Refactor jdbc storage implementations to reuse logics (apache#9904)

docs/en/changes/changes.md

@@ -96,7 +96,9 @@
   * Optimize data binary parse methods in *LogQueryDAO
   * Support different indexType
   * Support configuration for TTL and (block|segment) intervals
-* Optimize MQ Topology analysis. Use entry span's peer from the consumer side as source service when no producer instrumentation(no cross-process reference). 
+* Optimize MQ Topology analysis. Use entry span's peer from the consumer side as source service when no producer instrumentation(no cross-process reference).
+* Refactor JDBC storage implementations to reuse logics.
+* Fix `ClassCastException` in `LoggingConfigWatcher`.
 
 #### UI
 

docs/en/setup/backend/backend-setup.md

@@ -39,14 +39,14 @@ Before deploying Skywalking in your distributed environment, you should learn ab
 
 
 ## Startup script
-The default startup scripts are `/bin/oapService.sh`(.bat). 
+The default startup scripts are `/bin/oapService.sh`(.bat).
 Read the [start up mode](backend-start-up-mode.md) document to learn other ways to start up the backend.
 
 
 ## application.yml
 SkyWalking backend startup behaviours are driven by `config/application.yml`. Understanding the settings file will help you read this document.
 
-The core concept behind this setting file is that the SkyWalking collector is based on a pure modular design. 
+The core concept behind this setting file is that the SkyWalking collector is based on a pure modular design.
 End-users can switch or assemble the collector features according to their unique requirements.
 
 In `application.yml`, there are three levels.
@@ -60,9 +60,9 @@ Example:
 storage:
   selector: mysql # the mysql storage will actually be activated, while the h2 storage takes no effect
   h2:
-    driver: ${SW_STORAGE_H2_DRIVER:org.h2.jdbcx.JdbcDataSource}
-    url: ${SW_STORAGE_H2_URL:jdbc:h2:mem:skywalking-oap-db}
-    user: ${SW_STORAGE_H2_USER:sa}
+    properties:
+      jdbcUrl: ${SW_STORAGE_H2_URL:jdbc:h2:mem:skywalking-oap-db;DB_CLOSE_DELAY=-1}
+      dataSource.user: ${SW_STORAGE_H2_USER:sa}
     metadataQueryMaxSize: ${SW_STORAGE_H2_QUERY_MAX_SIZE:5000}
   mysql:
     properties:
@@ -82,7 +82,7 @@ storage:
 1. **`default`** is the default implementor of the core module.
 1. `driver`, `url`, ... `metadataQueryMaxSize` are all setting items of the implementor.
 
-At the same time, there are two types of modules: required and optional. The required modules provide the skeleton of the backend. 
+At the same time, there are two types of modules: required and optional. The required modules provide the skeleton of the backend.
 Even though their modular design supports pluggability, removing those modules does not serve any purpose. For optional modules, some of them have
 a provider implementation called `none`, meaning that it only provides a shell with no actual logic, typically such as telemetry.
 Setting `-` to the `selector` means that this whole module will be excluded at runtime.
@@ -98,20 +98,20 @@ capabilities. See [**Cluster Management**](backend-cluster.md) for more details.
 
 ## FAQs
 #### Why do we need to set the timezone? And when do we do it?
-SkyWalking provides downsampling time-series metrics features. 
+SkyWalking provides downsampling time-series metrics features.
 Query and store at each time dimension (minute, hour, day, month metrics indexes)
 related to timezone when time formatting.
 
 For example, metrics time will be formatted like yyyyMMddHHmm in minute dimension metrics, which is timezone-related.
-  
+
 By default, SkyWalking's OAP backend chooses the **OS default timezone**.
 Please follow the Java and OS documents if you want to override the timezone.
 
 #### How to query the storage directly from a 3rd party tool?
-SkyWalking provides different options based on browser UI, CLI and GraphQL to support extensions. But some users may want to query data 
+SkyWalking provides different options based on browser UI, CLI and GraphQL to support extensions. But some users may want to query data
 directly from the storage. For example, in the case of ElasticSearch, Kibana is a great tool for doing this.
 
-By default, SkyWalking saves based64-encoded ID(s) only in metrics entities to reduce memory, network and storage space usages. 
+By default, SkyWalking saves based64-encoded ID(s) only in metrics entities to reduce memory, network and storage space usages.
 But these tools usually don't support nested queries and are not convenient to work with. For these exceptional reasons,
 SkyWalking provides a config to add all necessary name column(s) into the final metrics entities with ID as a trade-off.
 

docs/en/setup/backend/configuration-vocabulary.md

@@ -116,7 +116,6 @@ The Configuration Vocabulary lists all available configurations provided by `app
 | -                       | -             | advanced                                                                                                                                                                 | All settings of ElasticSearch index creation. The value should be in JSON format.                                                                                                                                                                                                                                                                                                                                                                               | SW_STORAGE_ES_ADVANCED                            | -                                                                       |
 | -                       | -             | logicSharding                                                                                                                                          | Shard metrics and records indices into multi-physical indices, one index template per metric/meter aggregation function or record.                                                                                                                                                                                        | SW_STORAGE_ES_LOGIC_SHARDING       | false                                                                     |
 | -                       | h2            | -                                                                                                                                                                        | H2 storage is designed for demonstration and running in short term (i.e. 1-2 hours) only.                                                                                                                                                                                                                                                                                                                                                                       | -                                                 | -                                                                       |
-| -                       | -             | driver                                                                                                                                                                   | H2 JDBC driver.                                                                                                                                                                                                                                                                                                                                                                                                                                                 | SW_STORAGE_H2_DRIVER                              | org.h2.jdbcx.JdbcDataSource                                             |
 | -                       | -             | url                                                                                                                                                                      | H2 connection URL. Defaults to H2 memory mode.                                                                                                                                                                                                                                                                                                                                                                                                                  | SW_STORAGE_H2_URL                                 | jdbc:h2:mem:skywalking-oap-db                                           |
 | -                       | -             | user                                                                                                                                                                     | Username of H2 database.                                                                                                                                                                                                                                                                                                                                                                                                                                        | SW_STORAGE_H2_USER                                | sa                                                                      |
 | -                       | -             | password                                                                                                                                                                 | Password of H2 database.                                                                                                                                                                                                                                                                                                                                                                                                                                        | -                                                 | -                                                                       |

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/logging/LoggingConfigWatcher.java

@@ -25,6 +25,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.core.LoggerContext;
 import org.apache.logging.log4j.core.config.ConfigurationSource;
+import org.apache.logging.log4j.core.config.xml.XmlConfiguration;
 import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher;
 import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.logging.log4j.OapConfiguration;
@@ -37,13 +38,13 @@
 @Slf4j
 public class LoggingConfigWatcher extends ConfigChangeWatcher {
     private final LoggerContext ctx;
-    private final OapConfiguration originConfiguration;
+    private final XmlConfiguration originConfiguration;
     private volatile String content;
 
     public LoggingConfigWatcher(final ModuleProvider provider) {
         super(CoreModule.NAME, provider, "log4j-xml");
         this.ctx = (LoggerContext) LogManager.getContext(false);
-        this.originConfiguration = (OapConfiguration) ctx.getConfiguration();
+        this.originConfiguration = (XmlConfiguration) ctx.getConfiguration();
     }
 
     @Override

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java

@@ -33,7 +33,7 @@
 @Slf4j
 public abstract class ModelInstaller implements ModelCreator.CreatingListener {
     protected final Client client;
-    private final ModuleManager moduleManager;
+    protected final ModuleManager moduleManager;
 
     @Override
     public void whenCreating(Model model) throws StorageException {
@@ -42,8 +42,7 @@ public void whenCreating(Model model) throws StorageException {
                 try {
                     log.info(
                         "table: {} does not exist. OAP is running in 'no-init' mode, waiting... retry 3s later.",
-                        model
-                            .getName()
+                        model.getName()
                     );
                     Thread.sleep(3000L);
                 } catch (InterruptedException e) {
@@ -58,6 +57,9 @@ public void whenCreating(Model model) throws StorageException {
         }
     }
 
+    public void start() {
+    }
+
     /**
      * Installer implementation could use this API to request a column name replacement. This method delegates for
      * {@link ModelManipulator}.
@@ -72,10 +74,10 @@ protected final void overrideColumnName(String columnName, String newName) {
     /**
      * Check whether the storage entity exists. Need to implement based on the real storage.
      */
-    protected abstract boolean isExists(Model model) throws StorageException;
+    public abstract boolean isExists(Model model) throws StorageException;
 
     /**
      * Create the storage entity. All creations should be after the {@link #isExists(Model)} check.
      */
-    protected abstract void createTable(Model model) throws StorageException;
+    public abstract void createTable(Model model) throws StorageException;
 }

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java

@@ -129,7 +129,7 @@ public ResultSet executeQuery(Connection connection, String sql, Object... param
                 }
             }
             healthChecker.unHealth(e);
-            throw new JDBCClientException(e.getMessage(), e);
+            throw new JDBCClientException(sql, e);
         }
 
         return rs;

oap-server/server-starter/src/main/resources/application.yml

@@ -172,9 +172,9 @@ storage:
     # Enable shard metrics and records indices into multi-physical indices, one index template per metric/meter aggregation function or record.
     logicSharding: ${SW_STORAGE_ES_LOGIC_SHARDING:false}
   h2:
-    driver: ${SW_STORAGE_H2_DRIVER:org.h2.jdbcx.JdbcDataSource}
-    url: ${SW_STORAGE_H2_URL:jdbc:h2:mem:skywalking-oap-db;DB_CLOSE_DELAY=-1}
-    user: ${SW_STORAGE_H2_USER:sa}
+    properties:
+      jdbcUrl: ${SW_STORAGE_H2_URL:jdbc:h2:mem:skywalking-oap-db;DB_CLOSE_DELAY=-1}
+      dataSource.user: ${SW_STORAGE_H2_USER:sa}
     metadataQueryMaxSize: ${SW_STORAGE_H2_QUERY_MAX_SIZE:5000}
     maxSizeOfBatchSql: ${SW_STORAGE_MAX_SIZE_OF_BATCH_SQL:100}
     asyncBatchPersistentPoolSize: ${SW_STORAGE_ASYNC_BATCH_PERSISTENT_POOL_SIZE:1}
@@ -540,4 +540,4 @@ receiver-ebpf:
 receiver-telegraf:
   selector: ${SW_RECEIVER_TELEGRAF:default}
   default:
-    activeFiles: ${SW_RECEIVER_TELEGRAF_ACTIVE_FILES:vm}
+    activeFiles: ${SW_RECEIVER_TELEGRAF_ACTIVE_FILES:vm}

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java

@@ -36,16 +36,15 @@
 @Slf4j
 public class BanyanDBIndexInstaller extends ModelInstaller {
     private final BanyanDBStorageConfig config;
-    private final ConfigService configService;
 
     public BanyanDBIndexInstaller(Client client, ModuleManager moduleManager, BanyanDBStorageConfig config) {
         super(client, moduleManager);
         this.config = config;
-        this.configService = moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class);
     }
 
     @Override
-    protected boolean isExists(Model model) throws StorageException {
+    public boolean isExists(Model model) throws StorageException {
+        final ConfigService configService = moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class);
         final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(model, config, configService);
         try {
             final BanyanDBClient c = ((BanyanDBStorageClient) this.client).client;
@@ -73,8 +72,9 @@ protected boolean isExists(Model model) throws StorageException {
     }
 
     @Override
-    protected void createTable(Model model) throws StorageException {
+    public void createTable(Model model) throws StorageException {
         try {
+            ConfigService configService = moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class);
             if (model.isTimeSeries() && model.isRecord()) { // stream
                 Stream stream = MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
                 if (stream != null) {

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java

@@ -27,6 +27,7 @@
 import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
 import org.apache.skywalking.oap.server.core.storage.management.UITemplateManagementDAO;
 import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
+import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller;
 import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
 import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingScheduleDAO;
 import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
@@ -78,6 +79,7 @@
 public class BanyanDBStorageProvider extends ModuleProvider {
     private BanyanDBStorageConfig config;
     private BanyanDBStorageClient client;
+    private ModelInstaller modelInstaller;
 
     @Override
     public String name() {
@@ -109,6 +111,7 @@ public void prepare() throws ServiceNotProvidedException, ModuleStartException {
         this.registerServiceImplementation(StorageBuilderFactory.class, new StorageBuilderFactory.Default());
 
         this.client = new BanyanDBStorageClient(config.getHost(), config.getPort());
+        this.modelInstaller = new BanyanDBIndexInstaller(client, getManager(), this.config);
 
         // Stream
         this.registerServiceImplementation(
@@ -161,8 +164,9 @@ public void start() throws ServiceNotProvidedException, ModuleStartException {
         this.client.registerChecker(healthChecker);
         try {
             this.client.connect();
-            BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, getManager(), this.config);
-            getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
+            this.modelInstaller.start();
+
+            getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(modelInstaller);
         } catch (Exception e) {
             throw new ModuleStartException(e.getMessage(), e);
         }

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java

@@ -31,6 +31,7 @@
 import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
 import org.apache.skywalking.oap.server.core.storage.management.UITemplateManagementDAO;
 import org.apache.skywalking.oap.server.core.storage.model.ModelCreator;
+import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller;
 import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingDataDAO;
 import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingScheduleDAO;
 import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingTaskDAO;
@@ -97,6 +98,7 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider {
 
     protected StorageModuleElasticsearchConfig config;
     protected ElasticSearchClient elasticSearchClient;
+    protected ModelInstaller modelInstaller;
 
     @Override
     public String name() {
@@ -176,6 +178,8 @@ public void prepare() throws ServiceNotProvidedException {
             config.getSocketTimeout(), config.getResponseTimeout(),
             config.getNumHttpClientThread()
         );
+        modelInstaller = new StorageEsInstaller(elasticSearchClient, getManager(), config);
+
         this.registerServiceImplementation(
             IBatchDAO.class,
             new BatchProcessEsDAO(elasticSearchClient, config.getBulkActions(), config
@@ -235,9 +239,9 @@ public void start() throws ModuleStartException {
         elasticSearchClient.registerChecker(healthChecker);
         try {
             elasticSearchClient.connect();
-            StorageEsInstaller installer = new StorageEsInstaller(elasticSearchClient, getManager(), config);
+            modelInstaller.start();
 
-            getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(installer);
+            getManager().find(CoreModule.NAME).provider().getService(ModelCreator.class).addModelListener(modelInstaller);
         } catch (Exception e) {
             throw new ModuleStartException(e.getMessage(), e);
         }