You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2020/02/13 09:27:47 UTC
[skywalking] branch comments created (now 6c1bdcc)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a change to branch comments
in repository https://gitbox.apache.org/repos/asf/skywalking.git.
at 6c1bdcc Add more comments in the core module, should help the source code readers a lot.
This branch includes the following new commits:
new 6c1bdcc Add more comments in the core module, should help the source code readers a lot.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking] 01/01: Add more comments in the core module,
should help the source code readers a lot.
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch comments
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 6c1bdccfe9968bb06048c10616fdb55a681b4e1e
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Thu Feb 13 17:27:27 2020 +0800
Add more comments in the core module, should help the source code readers a lot.
---
.../oap/server/starter/OAPServerBootstrap.java | 12 +++--
.../skywalking/oap/server/core/CoreModule.java | 3 ++
.../oap/server/core/CoreModuleProvider.java | 55 ++++++++++++++------
.../oap/server/core/analysis/Stream.java | 26 +++++++++-
.../core/analysis/StreamAnnotationListener.java | 3 ++
.../server/core/analysis/worker/ExportWorker.java | 4 ++
.../analysis/worker/MetricsAggregateWorker.java | 16 ++++--
.../analysis/worker/MetricsPersistentWorker.java | 28 +++++++---
.../core/analysis/worker/MetricsRemoteWorker.java | 10 ++--
.../analysis/worker/MetricsStreamProcessor.java | 60 ++++++++++++++++++----
.../core/analysis/worker/MetricsTransWorker.java | 37 +++++++++----
.../oap/server/core/command/CommandService.java | 10 ++--
.../server/core/query/AggregationQueryService.java | 12 ++---
.../oap/server/core/query/MetricQueryService.java | 10 ++--
.../register/worker/InventoryStreamProcessor.java | 31 +++++++++--
.../oap/server/core/remote/Deserializable.java | 3 ++
.../server/core/remote/RemoteSenderService.java | 14 ++++-
.../server/core/remote/RemoteServiceHandler.java | 35 +++++++++++--
.../oap/server/core/remote/Serializable.java | 3 ++
.../skywalking/oap/server/core/storage/DAO.java | 3 ++
.../oap/server/core/storage/IBatchDAO.java | 19 ++++++-
.../oap/server/core/storage/IHistoryDeleteDAO.java | 12 ++++-
.../oap/server/core/storage/IMetricsDAO.java | 24 ++++++++-
.../oap/server/core/storage/IRecordDAO.java | 10 +++-
.../oap/server/core/storage/IRegisterDAO.java | 15 +++++-
.../oap/server/core/storage/StorageBuilder.java | 5 ++
.../oap/server/core/storage/StorageDAO.java | 3 ++
.../oap/server/core/storage/StorageModule.java | 4 ++
.../oap/server/core/storage/annotation/Column.java | 8 +++
...alueColumnIds.java => ValueColumnMetadata.java} | 19 +++++--
.../server/core/storage/model/ModelInstaller.java | 36 +++++++++----
.../server/core/storage/model/StorageModels.java | 4 +-
.../core/storage/ttl/DataTTLKeeperTimer.java | 20 +++++++-
.../oap/server/core/worker/AbstractWorker.java | 9 ++++
.../server/core/worker/IWorkerInstanceGetter.java | 4 +-
.../server/core/worker/IWorkerInstanceSetter.java | 9 ++++
.../handler/v6/grpc/RegisterServiceHandler.java | 18 +++++--
.../v6/grpc/ServiceInstancePingServiceHandler.java | 23 +++++----
.../oap/server/starter/OAPServerStartUp.java | 5 +-
.../oap/server/starter/OAPServerStartUp.java | 4 +-
.../plugin/jdbc/h2/dao/H2TableInstaller.java | 24 +++++----
.../plugin/jdbc/mysql/MySQLTableInstaller.java | 22 +++++---
42 files changed, 535 insertions(+), 137 deletions(-)
diff --git a/oap-server/server-bootstrap/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerBootstrap.java b/oap-server/server-bootstrap/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerBootstrap.java
index 3614427..704c278 100644
--- a/oap-server/server-bootstrap/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerBootstrap.java
+++ b/oap-server/server-bootstrap/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerBootstrap.java
@@ -18,6 +18,7 @@
package org.apache.skywalking.oap.server.starter;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.RunningMode;
import org.apache.skywalking.oap.server.library.module.ApplicationConfiguration;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
@@ -28,10 +29,11 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Starter core. Load the core configuration file, and initialize the startup sequence through {@link ModuleManager}.
+ */
+@Slf4j
public class OAPServerBootstrap {
-
- private static final Logger logger = LoggerFactory.getLogger(OAPServerBootstrap.class);
-
public static void start() {
String mode = System.getProperty("mode");
RunningMode.setMode(mode);
@@ -50,11 +52,11 @@ public class OAPServerBootstrap {
.setValue(System.currentTimeMillis() / 1000d);
if (RunningMode.isInitMode()) {
- logger.info("OAP starts up in init mode successfully, exit now...");
+ log.info("OAP starts up in init mode successfully, exit now...");
System.exit(0);
}
} catch (Throwable t) {
- logger.error(t.getMessage(), t);
+ log.error(t.getMessage(), t);
System.exit(1);
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java
index ce71372..e7fa4af 100755
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java
@@ -55,6 +55,9 @@ import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceGetter;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+/**
+ * Core module definition. Define all open services to other modules.
+ */
public class CoreModule extends ModuleDefine {
public static final String NAME = "core";
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
index 9069cb0..8f8d2b9 100755
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
@@ -93,6 +93,14 @@ import org.apache.skywalking.oap.server.library.server.grpc.GRPCServer;
import org.apache.skywalking.oap.server.library.server.jetty.JettyServer;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+/**
+ * Core module provider includes the recommended and default implementations of {@link CoreModule#services()}. All
+ * services with these default implementations are widely used including data receiver, data analysis, streaming
+ * process, storage and query.
+ *
+ * NOTICE. In our experiences, no one should re-implement the core module service implementations, unless we are very
+ * familiar with all mechanisms of SkyWalking.
+ */
public class CoreModuleProvider extends ModuleProvider {
private final CoreModuleConfig moduleConfig;
@@ -169,12 +177,14 @@ public class CoreModuleProvider extends ModuleProvider {
}
grpcServer.initialize();
- jettyServer = new JettyServer(moduleConfig.getRestHost(), moduleConfig.getRestPort(), moduleConfig.getRestContextPath(), moduleConfig
+ jettyServer = new JettyServer(
+ moduleConfig.getRestHost(), moduleConfig.getRestPort(), moduleConfig.getRestContextPath(), moduleConfig
.getJettySelectors());
jettyServer.initialize();
this.registerServiceImplementation(ConfigService.class, new ConfigService(moduleConfig));
- this.registerServiceImplementation(DownsamplingConfigService.class, new DownsamplingConfigService(moduleConfig.getDownsampling()));
+ this.registerServiceImplementation(
+ DownsamplingConfigService.class, new DownsamplingConfigService(moduleConfig.getDownsampling()));
this.registerServiceImplementation(GRPCHandlerRegister.class, new GRPCHandlerRegisterImpl(grpcServer));
this.registerServiceImplementation(JettyHandlerRegister.class, new JettyHandlerRegisterImpl(jettyServer));
@@ -192,17 +202,24 @@ public class CoreModuleProvider extends ModuleProvider {
this.registerServiceImplementation(IModelGetter.class, storageModels);
this.registerServiceImplementation(IModelOverride.class, storageModels);
- this.registerServiceImplementation(ServiceInventoryCache.class, new ServiceInventoryCache(getManager(), moduleConfig));
+ this.registerServiceImplementation(
+ ServiceInventoryCache.class, new ServiceInventoryCache(getManager(), moduleConfig));
this.registerServiceImplementation(IServiceInventoryRegister.class, new ServiceInventoryRegister(getManager()));
- this.registerServiceImplementation(ServiceInstanceInventoryCache.class, new ServiceInstanceInventoryCache(getManager(), moduleConfig));
- this.registerServiceImplementation(IServiceInstanceInventoryRegister.class, new ServiceInstanceInventoryRegister(getManager()));
+ this.registerServiceImplementation(
+ ServiceInstanceInventoryCache.class, new ServiceInstanceInventoryCache(getManager(), moduleConfig));
+ this.registerServiceImplementation(
+ IServiceInstanceInventoryRegister.class, new ServiceInstanceInventoryRegister(getManager()));
- this.registerServiceImplementation(EndpointInventoryCache.class, new EndpointInventoryCache(getManager(), moduleConfig));
- this.registerServiceImplementation(IEndpointInventoryRegister.class, new EndpointInventoryRegister(getManager()));
+ this.registerServiceImplementation(
+ EndpointInventoryCache.class, new EndpointInventoryCache(getManager(), moduleConfig));
+ this.registerServiceImplementation(
+ IEndpointInventoryRegister.class, new EndpointInventoryRegister(getManager()));
- this.registerServiceImplementation(NetworkAddressInventoryCache.class, new NetworkAddressInventoryCache(getManager(), moduleConfig));
- this.registerServiceImplementation(INetworkAddressInventoryRegister.class, new NetworkAddressInventoryRegister(getManager()));
+ this.registerServiceImplementation(
+ NetworkAddressInventoryCache.class, new NetworkAddressInventoryCache(getManager(), moduleConfig));
+ this.registerServiceImplementation(
+ INetworkAddressInventoryRegister.class, new NetworkAddressInventoryRegister(getManager()));
this.registerServiceImplementation(TopologyQueryService.class, new TopologyQueryService(getManager()));
this.registerServiceImplementation(MetricQueryService.class, new MetricQueryService(getManager()));
@@ -214,8 +231,10 @@ public class CoreModuleProvider extends ModuleProvider {
this.registerServiceImplementation(TopNRecordsQueryService.class, new TopNRecordsQueryService(getManager()));
// add profile service implementations
- this.registerServiceImplementation(ProfileTaskMutationService.class, new ProfileTaskMutationService(getManager()));
- this.registerServiceImplementation(ProfileTaskQueryService.class, new ProfileTaskQueryService(getManager(), moduleConfig));
+ this.registerServiceImplementation(
+ ProfileTaskMutationService.class, new ProfileTaskMutationService(getManager()));
+ this.registerServiceImplementation(
+ ProfileTaskQueryService.class, new ProfileTaskQueryService(getManager(), moduleConfig));
this.registerServiceImplementation(ProfileTaskCache.class, new ProfileTaskCache(getManager(), moduleConfig));
this.registerServiceImplementation(CommandService.class, new CommandService(getManager()));
@@ -248,10 +267,13 @@ public class CoreModuleProvider extends ModuleProvider {
}
if (CoreModuleConfig.Role.Mixed.name()
- .equalsIgnoreCase(moduleConfig.getRole()) || CoreModuleConfig.Role.Aggregator.name()
- .equalsIgnoreCase(moduleConfig
- .getRole())) {
- RemoteInstance gRPCServerInstance = new RemoteInstance(new Address(moduleConfig.getGRPCHost(), moduleConfig.getGRPCPort(), true));
+ .equalsIgnoreCase(
+ moduleConfig.getRole()) || CoreModuleConfig.Role.Aggregator.name()
+ .equalsIgnoreCase(
+ moduleConfig
+ .getRole())) {
+ RemoteInstance gRPCServerInstance = new RemoteInstance(
+ new Address(moduleConfig.getGRPCHost(), moduleConfig.getGRPCPort(), true));
this.getManager()
.find(ClusterModule.NAME)
.provider()
@@ -261,7 +283,8 @@ public class CoreModuleProvider extends ModuleProvider {
DynamicConfigurationService dynamicConfigurationService = getManager().find(ConfigurationModule.NAME)
.provider()
- .getService(DynamicConfigurationService.class);
+ .getService(
+ DynamicConfigurationService.class);
dynamicConfigurationService.registerConfigChangeWatcher(apdexThresholdConfig);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java
index 74709f7..7f6bd68 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java
@@ -22,17 +22,41 @@ import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import java.util.Map;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamingProcessor;
+import org.apache.skywalking.oap.server.core.analysis.worker.RecordStreamProcessor;
+import org.apache.skywalking.oap.server.core.analysis.worker.TopNStreamProcessor;
+import org.apache.skywalking.oap.server.core.register.worker.InventoryStreamProcessor;
+import org.apache.skywalking.oap.server.core.source.ScopeDeclaration;
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
+/**
+ * Stream annotation represents a metadata definition. Include the key values of the distributed streaming calculation.
+ * See {@link MetricsStreamProcessor}, {@link RecordStreamProcessor}, {@link InventoryStreamProcessor}, {@link
+ * TopNStreamProcessor} and {@link NoneStreamingProcessor} for more details.
+ */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface Stream {
-
+ /**
+ * @return name of this stream definition.
+ */
String name();
+ /**
+ * @return scope id, see {@link ScopeDeclaration}
+ */
int scopeId();
+ /**
+ * @return the converter type between {@link StorageBuilder} and {@link Map<String, Object>} for persistence.
+ */
Class<? extends StorageBuilder> builder();
+ /**
+ * @return the stream processor type, see {@link MetricsStreamProcessor}, {@link RecordStreamProcessor}, {@link
+ * InventoryStreamProcessor}, {@link * TopNStreamProcessor} and {@link NoneStreamingProcessor} for more details.
+ */
Class<? extends StreamProcessor> processor();
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java
index 154a439..13abca6 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java
@@ -28,6 +28,9 @@ import org.apache.skywalking.oap.server.core.annotation.AnnotationListener;
import org.apache.skywalking.oap.server.core.register.worker.InventoryStreamProcessor;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+/**
+ * Stream annotation listener, process the class with {@link Stream} annotation.
+ */
public class StreamAnnotationListener implements AnnotationListener {
private final ModuleDefineHolder moduleDefineHolder;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ExportWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ExportWorker.java
index 822a44e..63090bc 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ExportWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/ExportWorker.java
@@ -24,6 +24,10 @@ import org.apache.skywalking.oap.server.core.exporter.MetricValuesExportService;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+/**
+ * A bridge worker. If the {@link ExporterModule} provider declared and provides a implementation of {@link
+ * MetricValuesExportService}, forward the export data to it.
+ */
public class ExportWorker extends AbstractWorker<ExportEvent> {
private MetricValuesExportService exportService;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java
index 7daebc8..f6f944a 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsAggregateWorker.java
@@ -36,6 +36,12 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * MetricsAggregateWorker provides an in-memory metrics merging capability. This aggregation is called L1 aggregation,
+ * it merges the data just after the receiver analysis. The metrics belonging to the same entity, metrics type and time
+ * bucket, the L1 aggregation will merge them into one metrics object to reduce the unnecessary memory and network
+ * payload.
+ */
public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
private static final Logger logger = LoggerFactory.getLogger(MetricsAggregateWorker.class);
@@ -46,14 +52,15 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
private CounterMetrics aggregationCounter;
MetricsAggregateWorker(ModuleDefineHolder moduleDefineHolder, AbstractWorker<Metrics> nextWorker,
- String modelName) {
+ String modelName) {
super(moduleDefineHolder);
this.nextWorker = nextWorker;
this.mergeDataCache = new MergeDataCache<>();
String name = "METRICS_L1_AGGREGATION";
this.dataCarrier = new DataCarrier<>("MetricsAggregateWorker." + modelName, name, 2, 10000);
- BulkConsumePool.Creator creator = new BulkConsumePool.Creator(name, BulkConsumePool.Creator.recommendMaxSize() * 2, 20);
+ BulkConsumePool.Creator creator = new BulkConsumePool.Creator(
+ name, BulkConsumePool.Creator.recommendMaxSize() * 2, 20);
try {
ConsumerPoolFactory.INSTANCE.createIfAbsent(name, creator);
} catch (Exception e) {
@@ -64,7 +71,10 @@ public class MetricsAggregateWorker extends AbstractWorker<Metrics> {
MetricsCreator metricsCreator = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class);
- aggregationCounter = metricsCreator.createCounter("metrics_aggregation", "The number of rows in aggregation", new MetricsTag.Keys("metricName", "level", "dimensionality"), new MetricsTag.Values(modelName, "1", "min"));
+ aggregationCounter = metricsCreator.createCounter(
+ "metrics_aggregation", "The number of rows in aggregation",
+ new MetricsTag.Keys("metricName", "level", "dimensionality"), new MetricsTag.Values(modelName, "1", "min")
+ );
}
@Override
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java
index 263cac4..18e808b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java
@@ -26,6 +26,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
import org.apache.skywalking.apm.commons.datacarrier.consumer.BulkConsumePool;
import org.apache.skywalking.apm.commons.datacarrier.consumer.ConsumerPoolFactory;
@@ -35,17 +36,17 @@ import org.apache.skywalking.oap.server.core.analysis.data.MergeDataCache;
import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.exporter.ExportEvent;
import org.apache.skywalking.oap.server.core.storage.IMetricsDAO;
+import org.apache.skywalking.oap.server.core.storage.annotation.IDColumn;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.client.request.PrepareRequest;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * MetricsPersistentWorker is an extension of {@link PersistenceWorker} and focuses on the Metrics data persistent.
+ */
+@Slf4j
public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDataCache<Metrics>> {
-
- private static final Logger logger = LoggerFactory.getLogger(MetricsPersistentWorker.class);
-
private final Model model;
private final Map<Metrics, Metrics> databaseSession;
private final MergeDataCache<Metrics> mergeDataCache;
@@ -90,6 +91,9 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
cacheData(metrics);
}
+ /**
+ * Accept all metrics data and push them into the queue for serial processing
+ */
@Override
public void in(Metrics metrics) {
dataCarrier.produce(metrics);
@@ -144,7 +148,7 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
}
}
} catch (Throwable t) {
- logger.error(t.getMessage(), t);
+ log.error(t.getMessage(), t);
}
}
@@ -152,7 +156,7 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
}
if (prepareRequests.size() > 0) {
- logger.debug(
+ log.debug(
"prepare batch requests for model {}, took time: {}", model.getName(),
System.currentTimeMillis() - start
);
@@ -184,6 +188,9 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
mergeDataCache.finishWriting();
}
+ /**
+ * Sync data to the cache if the {@link #enableDatabaseSession} == true.
+ */
private void syncStorageToCache(Metrics[] metrics) throws IOException {
if (!enableDatabaseSession) {
databaseSession.clear();
@@ -219,6 +226,11 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
}
}
+ /**
+ * Metrics queue processor, merge the received metrics if existing one with same ID(s) and time bucket.
+ *
+ * ID is declared through {@link IDColumn}
+ */
private class PersistentConsumer implements IConsumer<Metrics> {
private final MetricsPersistentWorker persistent;
@@ -239,7 +251,7 @@ public class MetricsPersistentWorker extends PersistenceWorker<Metrics, MergeDat
@Override
public void onError(List<Metrics> data, Throwable t) {
- logger.error(t.getMessage(), t);
+ log.error(t.getMessage(), t);
}
@Override
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java
index 782338b..d3b9f76 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java
@@ -18,6 +18,7 @@
package org.apache.skywalking.oap.server.core.analysis.worker;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.remote.RemoteSenderService;
@@ -27,10 +28,11 @@ import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * MetricsRemoteWorker forwards the metrics to the target OAP node.
+ */
+@Slf4j
public class MetricsRemoteWorker extends AbstractWorker<Metrics> {
-
- private static final Logger logger = LoggerFactory.getLogger(MetricsRemoteWorker.class);
-
private final RemoteSenderService remoteSender;
private final String remoteReceiverWorkerName;
@@ -45,7 +47,7 @@ public class MetricsRemoteWorker extends AbstractWorker<Metrics> {
try {
remoteSender.send(remoteReceiverWorkerName, metrics, Selector.HashCode);
} catch (Throwable e) {
- logger.error(e.getMessage(), e);
+ log.error(e.getMessage(), e);
}
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
index 2075b24..bef05de 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
@@ -41,13 +41,33 @@ import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+/**
+ * MetricsStreamProcessor represents the entrance and creator of the metrics streaming aggregation work flow.
+ *
+ * {@link #in(Metrics)} provides the major entrance for metrics streaming calculation.
+ *
+ * {@link #create(ModuleDefineHolder, Stream, Class<Metrics>)} creates the workers and work flow for every metrics.
+ */
public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
-
+ /**
+ * Singleton instance.
+ */
private final static MetricsStreamProcessor PROCESSOR = new MetricsStreamProcessor();
+ /**
+ * Worker table hosts all entrance workers.
+ */
private Map<Class<? extends Metrics>, MetricsAggregateWorker> entryWorkers = new HashMap<>();
+
+ /**
+ * Worker table hosts all persistent workers.
+ */
@Getter
private List<MetricsPersistentWorker> persistentWorkers = new ArrayList<>();
+
+ /**
+ * Hold and forward CoreModuleConfig#enableDatabaseSession to the persistent worker.
+ */
@Setter
@Getter
private boolean enableDatabaseSession;
@@ -63,6 +83,13 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
}
}
+ /**
+ * Create the workers and work flow for every metrics.
+ *
+ * @param moduleDefineHolder pointer of the module define.
+ * @param stream definition of the metrics class.
+ * @param metricsClass data type of the streaming calculation.
+ */
@SuppressWarnings("unchecked")
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends Metrics> metricsClass) {
if (DisableRegister.INSTANCE.include(stream.name())) {
@@ -87,22 +114,28 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
MetricsPersistentWorker monthPersistentWorker = null;
if (configService.shouldToHour()) {
- Model model = modelSetter.putIfAbsent(metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Hour), false);
+ Model model = modelSetter.putIfAbsent(
+ metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Hour), false);
hourPersistentWorker = worker(moduleDefineHolder, metricsDAO, model);
}
if (configService.shouldToDay()) {
- Model model = modelSetter.putIfAbsent(metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Day), false);
+ Model model = modelSetter.putIfAbsent(
+ metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Day), false);
dayPersistentWorker = worker(moduleDefineHolder, metricsDAO, model);
}
if (configService.shouldToMonth()) {
- Model model = modelSetter.putIfAbsent(metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Month), false);
+ Model model = modelSetter.putIfAbsent(
+ metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Month), false);
monthPersistentWorker = worker(moduleDefineHolder, metricsDAO, model);
}
- MetricsTransWorker transWorker = new MetricsTransWorker(moduleDefineHolder, stream.name(), hourPersistentWorker, dayPersistentWorker, monthPersistentWorker);
+ MetricsTransWorker transWorker = new MetricsTransWorker(
+ moduleDefineHolder, stream.name(), hourPersistentWorker, dayPersistentWorker, monthPersistentWorker);
- Model model = modelSetter.putIfAbsent(metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Minute), false);
- MetricsPersistentWorker minutePersistentWorker = minutePersistentWorker(moduleDefineHolder, metricsDAO, model, transWorker);
+ Model model = modelSetter.putIfAbsent(
+ metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Minute), false);
+ MetricsPersistentWorker minutePersistentWorker = minutePersistentWorker(
+ moduleDefineHolder, metricsDAO, model, transWorker);
String remoteReceiverWorkerName = stream.name() + "_rec";
IWorkerInstanceSetter workerInstanceSetter = moduleDefineHolder.find(CoreModule.NAME)
@@ -111,24 +144,29 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
workerInstanceSetter.put(remoteReceiverWorkerName, minutePersistentWorker, metricsClass);
MetricsRemoteWorker remoteWorker = new MetricsRemoteWorker(moduleDefineHolder, remoteReceiverWorkerName);
- MetricsAggregateWorker aggregateWorker = new MetricsAggregateWorker(moduleDefineHolder, remoteWorker, stream.name());
+ MetricsAggregateWorker aggregateWorker = new MetricsAggregateWorker(
+ moduleDefineHolder, remoteWorker, stream.name());
entryWorkers.put(metricsClass, aggregateWorker);
}
private MetricsPersistentWorker minutePersistentWorker(ModuleDefineHolder moduleDefineHolder,
- IMetricsDAO metricsDAO, Model model, MetricsTransWorker transWorker) {
+ IMetricsDAO metricsDAO,
+ Model model,
+ MetricsTransWorker transWorker) {
AlarmNotifyWorker alarmNotifyWorker = new AlarmNotifyWorker(moduleDefineHolder);
ExportWorker exportWorker = new ExportWorker(moduleDefineHolder);
- MetricsPersistentWorker minutePersistentWorker = new MetricsPersistentWorker(moduleDefineHolder, model, metricsDAO, alarmNotifyWorker, exportWorker, transWorker, enableDatabaseSession);
+ MetricsPersistentWorker minutePersistentWorker = new MetricsPersistentWorker(
+ moduleDefineHolder, model, metricsDAO, alarmNotifyWorker, exportWorker, transWorker, enableDatabaseSession);
persistentWorkers.add(minutePersistentWorker);
return minutePersistentWorker;
}
private MetricsPersistentWorker worker(ModuleDefineHolder moduleDefineHolder, IMetricsDAO metricsDAO, Model model) {
- MetricsPersistentWorker persistentWorker = new MetricsPersistentWorker(moduleDefineHolder, model, metricsDAO, null, null, null, enableDatabaseSession);
+ MetricsPersistentWorker persistentWorker = new MetricsPersistentWorker(
+ moduleDefineHolder, model, metricsDAO, null, null, null, enableDatabaseSession);
persistentWorkers.add(persistentWorker);
return persistentWorker;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsTransWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsTransWorker.java
index 5af1e14..1f8f964 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsTransWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsTransWorker.java
@@ -19,6 +19,7 @@
package org.apache.skywalking.oap.server.core.analysis.worker;
import java.util.Objects;
+import org.apache.skywalking.oap.server.core.CoreModuleConfig;
import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
@@ -29,10 +30,12 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * MetricsTransWorker is transferring the metrics for downsampling. All streaming process metrics are in the minute
+ * precision, but at the storage layer, in order to enhance the query performance, metrics could be saved in minute,
+ * hour, day and month, including some of them through {@link CoreModuleConfig#getDownsampling()}.
+ */
public class MetricsTransWorker extends AbstractWorker<Metrics> {
-
- private static final Logger logger = LoggerFactory.getLogger(MetricsTransWorker.class);
-
private final MetricsPersistentWorker hourPersistenceWorker;
private final MetricsPersistentWorker dayPersistenceWorker;
private final MetricsPersistentWorker monthPersistenceWorker;
@@ -41,9 +44,11 @@ public class MetricsTransWorker extends AbstractWorker<Metrics> {
private final CounterMetrics aggregationDayCounter;
private final CounterMetrics aggregationMonthCounter;
- public MetricsTransWorker(ModuleDefineHolder moduleDefineHolder, String modelName,
- MetricsPersistentWorker hourPersistenceWorker, MetricsPersistentWorker dayPersistenceWorker,
- MetricsPersistentWorker monthPersistenceWorker) {
+ public MetricsTransWorker(ModuleDefineHolder moduleDefineHolder,
+ String modelName,
+ MetricsPersistentWorker hourPersistenceWorker,
+ MetricsPersistentWorker dayPersistenceWorker,
+ MetricsPersistentWorker monthPersistenceWorker) {
super(moduleDefineHolder);
this.hourPersistenceWorker = hourPersistenceWorker;
this.dayPersistenceWorker = dayPersistenceWorker;
@@ -52,11 +57,25 @@ public class MetricsTransWorker extends AbstractWorker<Metrics> {
MetricsCreator metricsCreator = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class);
- aggregationHourCounter = metricsCreator.createCounter("metrics_aggregation", "The number of rows in aggregation", new MetricsTag.Keys("metricName", "level", "dimensionality"), new MetricsTag.Values(modelName, "2", "hour"));
- aggregationDayCounter = metricsCreator.createCounter("metrics_aggregation", "The number of rows in aggregation", new MetricsTag.Keys("metricName", "level", "dimensionality"), new MetricsTag.Values(modelName, "2", "day"));
- aggregationMonthCounter = metricsCreator.createCounter("metrics_aggregation", "The number of rows in aggregation", new MetricsTag.Keys("metricName", "level", "dimensionality"), new MetricsTag.Values(modelName, "2", "month"));
+ aggregationHourCounter = metricsCreator.createCounter(
+ "metrics_aggregation", "The number of rows in aggregation", new MetricsTag.Keys("metricName", "level",
+ "dimensionality"
+ ), new MetricsTag.Values(modelName, "2", "hour"));
+ aggregationDayCounter = metricsCreator.createCounter(
+ "metrics_aggregation", "The number of rows in aggregation", new MetricsTag.Keys("metricName", "level",
+ "dimensionality"
+ ), new MetricsTag.Values(modelName, "2", "day"));
+ aggregationMonthCounter = metricsCreator.createCounter(
+ "metrics_aggregation", "The number of rows in aggregation", new MetricsTag.Keys("metricName", "level",
+ "dimensionality"
+ ), new MetricsTag.Values(modelName, "2", "month"));
}
+ /**
+ * Use the {@link Metrics#toHour()}, {@link Metrics#toDay()} and {@link Metrics#toMonth()} to clone a new metrics
+ * instance then process the downsampling. Then forward the data to different works of different precisions for
+ * another round aggregation/merging.
+ */
@Override
public void in(Metrics metrics) {
if (Objects.nonNull(hourPersistenceWorker)) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/command/CommandService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/command/CommandService.java
index 9bd86eb..0f98e2d 100755
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/command/CommandService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/command/CommandService.java
@@ -25,6 +25,9 @@ import org.apache.skywalking.oap.server.core.query.entity.ProfileTask;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.Service;
+/**
+ * CommandService represents the command creation factory. All commands for downstream agents should be created here.
+ */
public class CommandService implements Service {
private final ModuleManager moduleManager;
@@ -33,19 +36,20 @@ public class CommandService implements Service {
}
public ServiceResetCommand newResetCommand(final int serviceInstanceId, final long time,
- final String serviceInstanceUUID) {
+ final String serviceInstanceUUID) {
final String serialNumber = generateSerialNumber(serviceInstanceId, time, serviceInstanceUUID);
return new ServiceResetCommand(serialNumber);
}
public ProfileTaskCommand newProfileTaskCommand(ProfileTask task) {
final String serialNumber = UUID.randomUUID().toString();
- return new ProfileTaskCommand(serialNumber, task.getId(), task.getEndpointName(), task.getDuration(), task.getMinDurationThreshold(), task
+ return new ProfileTaskCommand(
+ serialNumber, task.getId(), task.getEndpointName(), task.getDuration(), task.getMinDurationThreshold(), task
.getDumpPeriod(), task.getMaxSamplingCount(), task.getStartTime(), task.getCreateTime());
}
private String generateSerialNumber(final int serviceInstanceId, final long time,
- final String serviceInstanceUUID) {
+ final String serviceInstanceUUID) {
return UUID.randomUUID().toString(); // Simply generate a uuid without taking care of the parameters
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java
index 6397af0..7c8d644 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/AggregationQueryService.java
@@ -31,7 +31,7 @@ import org.apache.skywalking.oap.server.core.register.EndpointInventory;
import org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory;
import org.apache.skywalking.oap.server.core.register.ServiceInventory;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
-import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnIds;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
import org.apache.skywalking.oap.server.core.storage.query.IAggregationQueryDAO;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.Service;
@@ -56,7 +56,7 @@ public class AggregationQueryService implements Service {
public List<TopNEntity> getServiceTopN(final String indName, final int topN, final Downsampling downsampling,
final long startTB, final long endTB, final Order order) throws IOException {
- List<TopNEntity> topNEntities = getAggregationQueryDAO().getServiceTopN(indName, ValueColumnIds.INSTANCE.getValueCName(indName), topN, downsampling, startTB, endTB, order);
+ List<TopNEntity> topNEntities = getAggregationQueryDAO().getServiceTopN(indName, ValueColumnMetadata.INSTANCE.getValueCName(indName), topN, downsampling, startTB, endTB, order);
for (TopNEntity entity : topNEntities) {
ServiceInventory inventory = moduleManager.find(CoreModule.NAME)
.provider()
@@ -71,7 +71,7 @@ public class AggregationQueryService implements Service {
public List<TopNEntity> getAllServiceInstanceTopN(final String indName, final int topN,
final Downsampling downsampling, final long startTB, final long endTB, final Order order) throws IOException {
- List<TopNEntity> topNEntities = getAggregationQueryDAO().getAllServiceInstanceTopN(indName, ValueColumnIds.INSTANCE
+ List<TopNEntity> topNEntities = getAggregationQueryDAO().getAllServiceInstanceTopN(indName, ValueColumnMetadata.INSTANCE
.getValueCName(indName), topN, downsampling, startTB, endTB, order);
for (TopNEntity entity : topNEntities) {
ServiceInstanceInventory inventory = moduleManager.find(CoreModule.NAME)
@@ -87,7 +87,7 @@ public class AggregationQueryService implements Service {
public List<TopNEntity> getServiceInstanceTopN(final int serviceId, final String indName, final int topN,
final Downsampling downsampling, final long startTB, final long endTB, final Order order) throws IOException {
- List<TopNEntity> topNEntities = getAggregationQueryDAO().getServiceInstanceTopN(serviceId, indName, ValueColumnIds.INSTANCE
+ List<TopNEntity> topNEntities = getAggregationQueryDAO().getServiceInstanceTopN(serviceId, indName, ValueColumnMetadata.INSTANCE
.getValueCName(indName), topN, downsampling, startTB, endTB, order);
for (TopNEntity entity : topNEntities) {
ServiceInstanceInventory inventory = moduleManager.find(CoreModule.NAME)
@@ -103,7 +103,7 @@ public class AggregationQueryService implements Service {
public List<TopNEntity> getAllEndpointTopN(final String indName, final int topN, final Downsampling downsampling,
final long startTB, final long endTB, final Order order) throws IOException {
- List<TopNEntity> topNEntities = getAggregationQueryDAO().getAllEndpointTopN(indName, ValueColumnIds.INSTANCE.getValueCName(indName), topN, downsampling, startTB, endTB, order);
+ List<TopNEntity> topNEntities = getAggregationQueryDAO().getAllEndpointTopN(indName, ValueColumnMetadata.INSTANCE.getValueCName(indName), topN, downsampling, startTB, endTB, order);
for (TopNEntity entity : topNEntities) {
EndpointInventory inventory = moduleManager.find(CoreModule.NAME)
.provider()
@@ -118,7 +118,7 @@ public class AggregationQueryService implements Service {
public List<TopNEntity> getEndpointTopN(final int serviceId, final String indName, final int topN,
final Downsampling downsampling, final long startTB, final long endTB, final Order order) throws IOException {
- List<TopNEntity> topNEntities = getAggregationQueryDAO().getEndpointTopN(serviceId, indName, ValueColumnIds.INSTANCE
+ List<TopNEntity> topNEntities = getAggregationQueryDAO().getEndpointTopN(serviceId, indName, ValueColumnMetadata.INSTANCE
.getValueCName(indName), topN, downsampling, startTB, endTB, order);
for (TopNEntity entity : topNEntities) {
EndpointInventory inventory = moduleManager.find(CoreModule.NAME)
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java
index edfc818..ea7d903 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/MetricQueryService.java
@@ -31,7 +31,7 @@ import org.apache.skywalking.oap.server.core.query.entity.Thermodynamic;
import org.apache.skywalking.oap.server.core.query.sql.KeyValues;
import org.apache.skywalking.oap.server.core.query.sql.Where;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
-import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnIds;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
import org.apache.skywalking.oap.server.core.storage.query.IMetricsQueryDAO;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.Service;
@@ -75,7 +75,7 @@ public class MetricQueryService implements Service {
where.getKeyValues().add(intKeyValues);
ids.forEach(intKeyValues.getValues()::add);
- return getMetricQueryDAO().getValues(metricsName, downsampling, startTB, endTB, where, ValueColumnIds.INSTANCE.getValueCName(metricsName), ValueColumnIds.INSTANCE
+ return getMetricQueryDAO().getValues(metricsName, downsampling, startTB, endTB, where, ValueColumnMetadata.INSTANCE.getValueCName(metricsName), ValueColumnMetadata.INSTANCE
.getValueFunction(metricsName));
}
@@ -89,7 +89,7 @@ public class MetricQueryService implements Service {
durationPoints.forEach(durationPoint -> ids.add(durationPoint.getPoint() + Const.ID_SPLIT + id));
}
- return getMetricQueryDAO().getLinearIntValues(indName, downsampling, ids, ValueColumnIds.INSTANCE.getValueCName(indName));
+ return getMetricQueryDAO().getLinearIntValues(indName, downsampling, ids, ValueColumnMetadata.INSTANCE.getValueCName(indName));
}
public List<IntValues> getMultipleLinearIntValues(final String indName, final String id, final int numOfLinear,
@@ -113,7 +113,7 @@ public class MetricQueryService implements Service {
durationPoints.forEach(durationPoint -> ids.add(durationPoint.getPoint() + Const.ID_SPLIT + id));
}
- IntValues[] multipleLinearIntValues = getMetricQueryDAO().getMultipleLinearIntValues(indName, downsampling, ids, linearIndex, ValueColumnIds.INSTANCE
+ IntValues[] multipleLinearIntValues = getMetricQueryDAO().getMultipleLinearIntValues(indName, downsampling, ids, linearIndex, ValueColumnMetadata.INSTANCE
.getValueCName(indName));
List<IntValues> response = new ArrayList<>(linearIndex.size());
@@ -133,6 +133,6 @@ public class MetricQueryService implements Service {
}
});
- return getMetricQueryDAO().getThermodynamic(indName, downsampling, ids, ValueColumnIds.INSTANCE.getValueCName(indName));
+ return getMetricQueryDAO().getThermodynamic(indName, downsampling, ids, ValueColumnMetadata.INSTANCE.getValueCName(indName));
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java
index ab16bab..b23f12f 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java
@@ -37,10 +37,24 @@ import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+/**
+ * InventoryStreamProcessor represents the entrance and creator of the inventory register work flow.
+ *
+ * {@link #in(RegisterSource)} provides the major entrance for inventory streaming merge, eventually add or update the
+ * inventory data in the storage.
+ *
+ * {@link #create(ModuleDefineHolder, Stream, Class<RegisterSource>)} creates the workers and work flow for every
+ * inventory.
+ */
public class InventoryStreamProcessor implements StreamProcessor<RegisterSource> {
-
+ /**
+ * Singleton instance.
+ */
private static final InventoryStreamProcessor PROCESSOR = new InventoryStreamProcessor();
+ /**
+ * Worker table hosts all entrance workers.
+ */
private Map<Class<? extends RegisterSource>, RegisterDistinctWorker> entryWorkers = new HashMap<>();
public static InventoryStreamProcessor getInstance() {
@@ -51,9 +65,16 @@ public class InventoryStreamProcessor implements StreamProcessor<RegisterSource>
entryWorkers.get(registerSource.getClass()).in(registerSource);
}
+ /**
+ * Create the workers and work flow for every inventory.
+ *
+ * @param moduleDefineHolder pointer of the module define.
+ * @param stream definition of the inventory class.
+ * @param inventoryClass data type of the inventory.
+ */
@SuppressWarnings("unchecked")
public void create(ModuleDefineHolder moduleDefineHolder, Stream stream,
- Class<? extends RegisterSource> inventoryClass) {
+ Class<? extends RegisterSource> inventoryClass) {
StorageDAO storageDAO = moduleDefineHolder.find(StorageModule.NAME).provider().getService(StorageDAO.class);
IRegisterDAO registerDAO;
try {
@@ -63,9 +84,11 @@ public class InventoryStreamProcessor implements StreamProcessor<RegisterSource>
}
IModelSetter modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(IModelSetter.class);
- Model model = modelSetter.putIfAbsent(inventoryClass, stream.scopeId(), new Storage(stream.name(), false, false, Downsampling.None), false);
+ Model model = modelSetter.putIfAbsent(
+ inventoryClass, stream.scopeId(), new Storage(stream.name(), false, false, Downsampling.None), false);
- RegisterPersistentWorker persistentWorker = new RegisterPersistentWorker(moduleDefineHolder, model.getName(), registerDAO, stream
+ RegisterPersistentWorker persistentWorker = new RegisterPersistentWorker(
+ moduleDefineHolder, model.getName(), registerDAO, stream
.scopeId());
String remoteReceiverWorkerName = stream.name() + "_rec";
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Deserializable.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Deserializable.java
index 14bdb75..729125f 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Deserializable.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Deserializable.java
@@ -20,6 +20,9 @@ package org.apache.skywalking.oap.server.core.remote;
import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
+/**
+ * Covert the {@link RemoteData} received from the network to the current data entity.
+ */
public interface Deserializable {
void deserialize(RemoteData remoteData);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java
index 7c2fc5c..cd4a54e 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java
@@ -32,6 +32,10 @@ import org.apache.skywalking.oap.server.library.module.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * RemoteSenderService represents a gRPC client to send metrics from one OAP node to another through network. It
+ * provides several routing mode to select target OAP node.
+ */
public class RemoteSenderService implements Service {
private static final Logger logger = LoggerFactory.getLogger(RemoteSenderService.class);
@@ -47,6 +51,13 @@ public class RemoteSenderService implements Service {
this.rollingSelector = new RollingSelector();
}
+ /**
+ * Send data to the target based on the given selector
+ *
+ * @param nextWorkName points to the worker to process the data when {@link RemoteServiceHandler} received.
+ * @param streamData data to be sent
+ * @param selector strategy implementation to choose suitable OAP node.
+ */
public void send(String nextWorkName, StreamData streamData, Selector selector) {
RemoteClientManager clientManager = moduleManager.find(CoreModule.NAME)
.provider()
@@ -55,7 +66,8 @@ public class RemoteSenderService implements Service {
List<RemoteClient> clientList = clientManager.getRemoteClient();
if (clientList.size() == 0) {
- logger.warn("There is no available remote server for now, ignore the streaming data until the cluster metadata initialized.");
+ logger.warn(
+ "There is no available remote server for now, ignore the streaming data until the cluster metadata initialized.");
return;
}
switch (selector) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java
index c2c033e..c009753 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java
@@ -61,21 +61,43 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
remoteInCounter = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class)
- .createCounter("remote_in_count", "The number(server side) of inside remote inside aggregate rpc.", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+ .createCounter(
+ "remote_in_count",
+ "The number(server side) of inside remote inside aggregate rpc.",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
+ );
remoteInErrorCounter = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class)
- .createCounter("remote_in_error_count", "The error number(server side) of inside remote inside aggregate rpc.", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+ .createCounter(
+ "remote_in_error_count",
+ "The error number(server side) of inside remote inside aggregate rpc.",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
+ );
remoteInTargetNotFoundCounter = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class)
- .createCounter("remote_in_target_not_found_count", "The error number(server side) of inside remote handler target worker not found. May be caused by unmatched OAL scrips.", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+ .createCounter(
+ "remote_in_target_not_found_count",
+ "The error number(server side) of inside remote handler target worker not found. May be caused by unmatched OAL scrips.",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
+ );
remoteInHistogram = moduleDefineHolder.find(TelemetryModule.NAME)
.provider()
.getService(MetricsCreator.class)
- .createHistogramMetric("remote_in_latency", "The latency(server side) of inside remote inside aggregate rpc.", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+ .createHistogramMetric(
+ "remote_in_latency",
+ "The latency(server side) of inside remote inside aggregate rpc.",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE
+ );
}
+ /**
+ * gRPC handler of {@link RemoteServiceGrpc}. Continue the distributed aggregation at the current OAP node.
+ *
+ * @param responseObserver
+ * @return
+ */
@Override
public StreamObserver<RemoteMessage> call(StreamObserver<Empty> responseObserver) {
if (Objects.isNull(workerInstanceGetter)) {
@@ -106,7 +128,10 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
nextWorker.in(streamData);
} else {
remoteInTargetNotFoundCounter.inc();
- logger.warn("Work name [{}] not found. Check OAL script, make sure they are same in the whole cluster.", nextWorkerName);
+ logger.warn(
+ "Work name [{}] not found. Check OAL script, make sure they are same in the whole cluster.",
+ nextWorkerName
+ );
}
} catch (Throwable t) {
remoteInErrorCounter.inc();
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Serializable.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Serializable.java
index 382371c..d5dc58a 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Serializable.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/Serializable.java
@@ -20,6 +20,9 @@ package org.apache.skywalking.oap.server.core.remote;
import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
+/**
+ * Covert the interface implementation to {@link RemoteData.Builder}, in order to send the data through network.
+ */
public interface Serializable {
RemoteData.Builder serialize();
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/DAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/DAO.java
index dab0d49..9d066e4 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/DAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/DAO.java
@@ -20,5 +20,8 @@ package org.apache.skywalking.oap.server.core.storage;
import org.apache.skywalking.oap.server.library.module.Service;
+/**
+ * A specific interface for storage layer services.
+ */
public interface DAO extends Service {
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java
index 883b984..c6bb736 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java
@@ -22,9 +22,26 @@ import java.util.List;
import org.apache.skywalking.oap.server.library.client.request.InsertRequest;
import org.apache.skywalking.oap.server.library.client.request.PrepareRequest;
+/**
+ * IBatchDAO provides two modes of data persistence supported by most databases, including synchronous and
+ * asynchronous.
+ */
public interface IBatchDAO extends DAO {
-
+ /**
+ * Push data into the database in async mode. This method is driven by streaming process. This method doesn't
+ * request the data queryable immediately after the method finished.
+ *
+ * All data are in the additional mode, no modification.
+ *
+ * @param insertRequest data to insert.
+ */
void asynchronous(InsertRequest insertRequest);
+ /**
+ * Make all given PrepareRequest efficient in the sync mode. All requests could be confirmed by the database. All
+ * changes are required queryable after method returns.
+ *
+ * @param prepareRequests data to insert or update. No delete happens in streaming mode.
+ */
void synchronous(List<PrepareRequest> prepareRequests);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IHistoryDeleteDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IHistoryDeleteDAO.java
index 2b7d826..d778af0 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IHistoryDeleteDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IHistoryDeleteDAO.java
@@ -19,9 +19,19 @@
package org.apache.skywalking.oap.server.core.storage;
import java.io.IOException;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
import org.apache.skywalking.oap.server.core.storage.model.Model;
+/**
+ * Remove all expired data based on TTL configurations.
+ */
public interface IHistoryDeleteDAO extends DAO {
-
+ /**
+ * Delete the data
+ *
+ * @param model data entity.
+ * @param timeBucketColumnName column name represents the time. Right now, always {@link Metrics#TIME_BUCKET}
+ * @throws IOException when error happens in the deletion process.
+ */
void deleteHistory(Model model, String timeBucketColumnName) throws IOException;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IMetricsDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IMetricsDAO.java
index 2cc35dc..c5add94 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IMetricsDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IMetricsDAO.java
@@ -25,11 +25,33 @@ import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.library.client.request.InsertRequest;
import org.apache.skywalking.oap.server.library.client.request.UpdateRequest;
+/**
+ * Metrics related DAO.
+ */
public interface IMetricsDAO extends DAO {
-
+ /**
+ * Read data from the storage by given IDs.
+ *
+ * @param model target entity of this query.
+ * @param ids ID list.
+ * @return the data of all given IDs. Only include existing data. Don't require to keep the same order of ids list.
+ * @throws IOException when error occurs in data query.
+ */
List<Metrics> multiGet(Model model, List<String> ids) throws IOException;
+ /**
+ * Transfer the given metrics to an executable insert statement.
+ *
+ * @return InsertRequest should follow the database client driver datatype, in order to make sure it could be
+ * executed ASAP.
+ */
InsertRequest prepareBatchInsert(Model model, Metrics metrics) throws IOException;
+ /**
+ * Transfer the given metrics to an executable update statement.
+ *
+ * @return UpdateRequest should follow the database client driver datatype, in order to make sure it could be
+ * executed ASAP.
+ */
UpdateRequest prepareBatchUpdate(Model model, Metrics metrics) throws IOException;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java
index 97343d5..8503b87 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java
@@ -23,7 +23,15 @@ import org.apache.skywalking.oap.server.core.analysis.record.Record;
import org.apache.skywalking.oap.server.core.storage.model.Model;
import org.apache.skywalking.oap.server.library.client.request.InsertRequest;
+/**
+ * DAO specifically for {@link Record} implementations.
+ */
public interface IRecordDAO extends DAO {
-
+ /**
+ * Transfer the given metrics to an executable insert statement.
+ *
+ * @return InsertRequest should follow the database client driver datatype, in order to make sure it could be
+ * executed ASAP.
+ */
InsertRequest prepareBatchInsert(Model model, Record record) throws IOException;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRegisterDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRegisterDAO.java
index 87fda70..5afb452 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRegisterDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRegisterDAO.java
@@ -21,11 +21,24 @@ package org.apache.skywalking.oap.server.core.storage;
import java.io.IOException;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
+/**
+ * DAO specifically for {@link RegisterSource} implementations.
+ */
public interface IRegisterDAO extends DAO {
-
+ /**
+ * Read the RegisterSource by the given ID.
+ *
+ * @return RegisterSource instance or NULL if id doesn't exist.
+ */
RegisterSource get(String modelName, String id) throws IOException;
+ /**
+ * Do a blocking insert operation.
+ */
void forceInsert(String modelName, RegisterSource source) throws IOException;
+ /**
+ * Do a blocking update operation.
+ */
void forceUpdate(String modelName, RegisterSource source) throws IOException;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageBuilder.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageBuilder.java
index 0a07a17..f4746b8 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageBuilder.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageBuilder.java
@@ -20,6 +20,11 @@ package org.apache.skywalking.oap.server.core.storage;
import java.util.Map;
+/**
+ * Converter between the give T and Map.
+ *
+ * @param <T> A storage entity implementation.
+ */
public interface StorageBuilder<T extends StorageData> {
T map2Data(Map<String, Object> dbMap);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java
index 45bbb72..89525a8 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java
@@ -24,6 +24,9 @@ import org.apache.skywalking.oap.server.core.analysis.record.Record;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
import org.apache.skywalking.oap.server.library.module.Service;
+/**
+ * StorageDAO is a DAO factory for storage layer. Provide the implementations of typical DAO interfaces.
+ */
public interface StorageDAO extends Service {
IMetricsDAO newMetricsDao(StorageBuilder<Metrics> storageBuilder);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java
index 9f889db..a3194b6 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageModule.java
@@ -35,6 +35,10 @@ import org.apache.skywalking.oap.server.core.storage.query.ITopologyQueryDAO;
import org.apache.skywalking.oap.server.core.storage.query.ITraceQueryDAO;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+/**
+ * StorageModule provides the capabilities(services) to interact with the database. With different databases, this
+ * module could have different providers, such as currently, H2, MySQL, ES, TiDB.
+ */
public class StorageModule extends ModuleDefine {
public static final String NAME = "storage";
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/Column.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/Column.java
index f52a74d..d04e2c1 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/Column.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/Column.java
@@ -23,10 +23,18 @@ import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.apache.skywalking.oap.server.core.query.sql.Function;
+import org.apache.skywalking.oap.server.core.storage.model.IModelOverride;
+/**
+ * Data column of all persistent entity.
+ */
@Target({ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
public @interface Column {
+ /**
+ * column name in the storage. Most of the storage will keep the name consistently. But in same cases, this name
+ * could be a keyword, then, the implementation will use {@link IModelOverride} to replace the column name.
+ */
String columnName();
/**
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/ValueColumnIds.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/ValueColumnMetadata.java
similarity index 75%
rename from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/ValueColumnIds.java
rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/ValueColumnMetadata.java
index 2b4767d..174f277 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/ValueColumnIds.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/ValueColumnMetadata.java
@@ -22,19 +22,32 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.skywalking.oap.server.core.query.sql.Function;
-public enum ValueColumnIds {
+/**
+ * ValueColumnMetadata holds the metadata for column values of metrics. The metadata of ValueColumn is declared through
+ * {@link Column} annotation.
+ */
+public enum ValueColumnMetadata {
INSTANCE;
private Map<String, ValueColumn> mapping = new HashMap<>();
- public void putIfAbsent(String indName, String valueCName, Function function) {
- mapping.putIfAbsent(indName, new ValueColumn(valueCName, function));
+ /**
+ * Register the new metadata for the given model name.
+ */
+ public void putIfAbsent(String modelName, String valueCName, Function function) {
+ mapping.putIfAbsent(modelName, new ValueColumn(valueCName, function));
}
+ /**
+ * Fetch the value column name of the given metrics name.
+ */
public String getValueCName(String metricsName) {
return findColumn(metricsName).valueCName;
}
+ /**
+ * Fetch the function for the value column of the given metrics name.
+ */
public Function getValueFunction(String metricsName) {
return findColumn(metricsName).function;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java
index 11f78ee..0de46bd 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java
@@ -19,24 +19,27 @@
package org.apache.skywalking.oap.server.core.storage.model;
import java.util.List;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.RunningMode;
import org.apache.skywalking.oap.server.core.storage.StorageException;
import org.apache.skywalking.oap.server.library.client.Client;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * The core module installation controller.
+ */
+@Slf4j
public abstract class ModelInstaller {
-
- private static final Logger logger = LoggerFactory.getLogger(ModelInstaller.class);
-
private final ModuleManager moduleManager;
public ModelInstaller(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
+ /**
+ * Entrance of the storage entity installation work.
+ */
public final void install(Client client) throws StorageException {
IModelGetter modelGetter = moduleManager.find(CoreModule.NAME).provider().getService(IModelGetter.class);
@@ -46,30 +49,43 @@ public abstract class ModelInstaller {
for (Model model : models) {
while (!isExists(client, model)) {
try {
- logger.info("table: {} does not exist. OAP is running in 'no-init' mode, waiting... retry 3s later.", model
- .getName());
+ log.info(
+ "table: {} does not exist. OAP is running in 'no-init' mode, waiting... retry 3s later.",
+ model
+ .getName()
+ );
Thread.sleep(3000L);
} catch (InterruptedException e) {
- logger.error(e.getMessage());
+ log.error(e.getMessage());
}
}
}
} else {
for (Model model : models) {
if (!isExists(client, model)) {
- logger.info("table: {} does not exist", model.getName());
+ log.info("table: {} does not exist", model.getName());
createTable(client, model);
}
}
}
}
- public final void overrideColumnName(String columnName, String newName) {
+ /**
+ * Installer implementation could use this API to request a column name replacement. This method delegates for
+ * {@link IModelOverride}.
+ */
+ protected final void overrideColumnName(String columnName, String newName) {
IModelOverride modelOverride = moduleManager.find(CoreModule.NAME).provider().getService(IModelOverride.class);
modelOverride.overrideColumnName(columnName, newName);
}
+ /**
+ * Check whether the storage entity exists. Need to implement based on the real storage.
+ */
protected abstract boolean isExists(Client client, Model model) throws StorageException;
+ /**
+ * Create the storage entity. All creations should be after the {@link #isExists(Client, Model)} check.
+ */
protected abstract void createTable(Client client, Model model) throws StorageException;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
index 5f7b790..ad21f2a 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
@@ -25,7 +25,7 @@ import lombok.Getter;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
-import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnIds;
+import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,7 +73,7 @@ public class StorageModels implements IModelGetter, IModelSetter, IModelOverride
logger.debug("The field named {} with the {} type", column.columnName(), field.getType());
}
if (column.isValue()) {
- ValueColumnIds.INSTANCE.putIfAbsent(modelName, column.columnName(), column.function());
+ ValueColumnMetadata.INSTANCE.putIfAbsent(modelName, column.columnName(), column.function());
}
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java
index 33d8630..d83049a 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java
@@ -38,6 +38,14 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * TTL = Time To Live
+ *
+ * DataTTLKeeperTimer is an internal timer, it drives the {@link IHistoryDeleteDAO} to remove the expired data. TTL
+ * configurations are provided in {@link CoreModuleConfig}, some storage implementations, such as ES6/ES7, provides an
+ * override TTL, which could be more suitable for the implementation. No matter which TTL configurations are set, they
+ * are all driven by this timer.
+ */
public enum DataTTLKeeperTimer {
INSTANCE;
@@ -51,10 +59,18 @@ public enum DataTTLKeeperTimer {
this.clusterNodesQuery = moduleManager.find(ClusterModule.NAME).provider().getService(ClusterNodesQuery.class);
Executors.newSingleThreadScheduledExecutor()
- .scheduleAtFixedRate(new RunnableWithExceptionProtection(this::delete, t -> logger.error("Remove data in background failure.", t)), moduleConfig
- .getDataKeeperExecutePeriod(), moduleConfig.getDataKeeperExecutePeriod(), TimeUnit.MINUTES);
+ .scheduleAtFixedRate(
+ new RunnableWithExceptionProtection(
+ this::delete,
+ t -> logger.error("Remove data in background failure.", t)
+ ), moduleConfig
+ .getDataKeeperExecutePeriod(), moduleConfig.getDataKeeperExecutePeriod(), TimeUnit.MINUTES);
}
+ /**
+ * DataTTLKeeperTimer starts in every OAP node, but the deletion only work when it is as the first node in the OAP
+ * node list from {@link ClusterNodesQuery}.
+ */
private void delete() {
List<RemoteInstance> remoteInstances = clusterNodesQuery.queryRemoteNodes();
if (CollectionUtils.isNotEmpty(remoteInstances) && !remoteInstances.get(0).getAddress().isSelf()) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java
index 60f77db..8ee41b4 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java
@@ -21,6 +21,12 @@ package org.apache.skywalking.oap.server.core.worker;
import lombok.Getter;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+/**
+ * Abstract worker definition. Provide the {@link ModuleDefineHolder} to make sure the worker could find and access
+ * services in different modules. Also, {@link #in(Object)} is provided as the primary entrance of every worker.
+ *
+ * @param <INPUT> the datatype this worker implementation processes.
+ */
public abstract class AbstractWorker<INPUT> {
@Getter
@@ -30,5 +36,8 @@ public abstract class AbstractWorker<INPUT> {
this.moduleDefineHolder = moduleDefineHolder;
}
+ /**
+ * Main entrance of this worker.
+ */
public abstract void in(INPUT input);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java
index c98175a..3ba0112 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java
@@ -20,8 +20,10 @@ package org.apache.skywalking.oap.server.core.worker;
import org.apache.skywalking.oap.server.library.module.Service;
+/**
+ * Worker instance finder interface. Find work instance from all registered work instance based on worker name.
+ */
public interface IWorkerInstanceGetter extends Service {
-
RemoteHandleWorker get(String nextWorkerName);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java
index 6336a81..c71698a 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java
@@ -18,9 +18,18 @@
package org.apache.skywalking.oap.server.core.worker;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
import org.apache.skywalking.oap.server.core.remote.data.StreamData;
import org.apache.skywalking.oap.server.library.module.Service;
+/**
+ * Worker instance register interface. Push the worker name, instance and class type having {@link Stream} annotation.
+ */
public interface IWorkerInstanceSetter extends Service {
+ /**
+ * @param remoteReceiverWorkName worker name
+ * @param instance The worker instance processes the given streamDataClass.
+ * @param streamDataClass Type of metrics.
+ */
void put(String remoteReceiverWorkName, AbstractWorker instance, Class<? extends StreamData> streamDataClass);
}
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/RegisterServiceHandler.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/RegisterServiceHandler.java
index ec72b32..a7c250e 100644
--- a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/RegisterServiceHandler.java
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/RegisterServiceHandler.java
@@ -61,6 +61,12 @@ import static org.apache.skywalking.oap.server.core.register.ServiceInstanceInve
import static org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory.PropertyUtil.OS_NAME;
import static org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory.PropertyUtil.PROCESS_NO;
+/**
+ * RegisterServiceHandler responses the requests of multiple inventory entities register, including service, instance,
+ * endpoint, network address and address-service mapping. Responses of service, instance and endpoint register include
+ * the IDs to represents these entities. Agent could use them in the header and data report to reduce the network
+ * bandwidth resource costs.
+ */
public class RegisterServiceHandler extends RegisterGrpc.RegisterImplBase implements GRPCHandler {
private static final Logger logger = LoggerFactory.getLogger(RegisterServiceHandler.class);
@@ -109,7 +115,8 @@ public class RegisterServiceHandler extends RegisterGrpc.RegisterImplBase implem
serviceType = ServiceType.normal;
}
- int serviceId = serviceInventoryRegister.getOrCreate(serviceName, NodeType.fromRegisterServiceType(serviceType), null);
+ int serviceId = serviceInventoryRegister.getOrCreate(
+ serviceName, NodeType.fromRegisterServiceType(serviceType), null);
if (serviceId != Const.NONE) {
KeyIntValuePair value = KeyIntValuePair.newBuilder().setKey(serviceName).setValue(serviceId).build();
@@ -123,7 +130,7 @@ public class RegisterServiceHandler extends RegisterGrpc.RegisterImplBase implem
@Override
public void doServiceInstanceRegister(ServiceInstances request,
- StreamObserver<ServiceInstanceRegisterMapping> responseObserver) {
+ StreamObserver<ServiceInstanceRegisterMapping> responseObserver) {
ServiceInstanceRegisterMapping.Builder builder = ServiceInstanceRegisterMapping.newBuilder();
@@ -176,8 +183,9 @@ public class RegisterServiceHandler extends RegisterGrpc.RegisterImplBase implem
}
}
- int serviceInstanceId = serviceInstanceInventoryRegister.getOrCreate(instance.getServiceId(), instanceName, instanceUUID, instance
- .getTime(), instanceProperties);
+ int serviceInstanceId = serviceInstanceInventoryRegister.getOrCreate(
+ instance.getServiceId(), instanceName, instanceUUID, instance
+ .getTime(), instanceProperties);
if (serviceInstanceId != Const.NONE) {
logger.info("register service instance id={} [UUID:{}]", serviceInstanceId, instanceUUID);
@@ -237,7 +245,7 @@ public class RegisterServiceHandler extends RegisterGrpc.RegisterImplBase implem
@Override
public void doServiceAndNetworkAddressMappingRegister(ServiceAndNetworkAddressMappings request,
- StreamObserver<Commands> responseObserver) {
+ StreamObserver<Commands> responseObserver) {
request.getMappingsList().forEach(mapping -> {
int serviceId = mapping.getServiceId();
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/ServiceInstancePingServiceHandler.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/ServiceInstancePingServiceHandler.java
index 24e8b55..c7aec32 100755
--- a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/ServiceInstancePingServiceHandler.java
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/grpc/ServiceInstancePingServiceHandler.java
@@ -19,6 +19,8 @@
package org.apache.skywalking.oap.server.receiver.register.provider.handler.v6.grpc;
import io.grpc.stub.StreamObserver;
+import java.util.Objects;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.apm.network.common.Command;
import org.apache.skywalking.apm.network.common.Commands;
import org.apache.skywalking.apm.network.register.v2.ServiceInstancePingGrpc;
@@ -32,14 +34,13 @@ import org.apache.skywalking.oap.server.core.register.service.IServiceInstanceIn
import org.apache.skywalking.oap.server.core.register.service.IServiceInventoryRegister;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Objects;
+/**
+ * ServiceInstancePingServiceHandler responses the requests for instance ping. Trigger the heartbeat update and push the
+ * commands to the downstream.
+ */
+@Slf4j
public class ServiceInstancePingServiceHandler extends ServiceInstancePingGrpc.ServiceInstancePingImplBase implements GRPCHandler {
- private static final Logger logger = LoggerFactory.getLogger(ServiceInstancePingServiceHandler.class);
-
private final ServiceInstanceInventoryCache serviceInstanceInventoryCache;
private final IServiceInventoryRegister serviceInventoryRegister;
private final IServiceInstanceInventoryRegister serviceInstanceInventoryRegister;
@@ -69,10 +70,14 @@ public class ServiceInstancePingServiceHandler extends ServiceInstancePingGrpc.S
serviceInventoryRegister.heartbeat(serviceInstanceInventory.getServiceId(), heartBeatTime);
responseObserver.onNext(Commands.getDefaultInstance());
} else {
- logger.warn("Can't find service by service instance id from cache," + " service instance id is: {}, will send a reset command to agent side", serviceInstanceId);
+ log.warn(
+ "Can't find service by service instance id from cache," + " service instance id is: {}, will send a reset command to agent side",
+ serviceInstanceId
+ );
- final ServiceResetCommand resetCommand = commandService.newResetCommand(request.getServiceInstanceId(), request
- .getTime(), request.getServiceInstanceUUID());
+ final ServiceResetCommand resetCommand = commandService.newResetCommand(
+ request.getServiceInstanceId(), request
+ .getTime(), request.getServiceInstanceUUID());
final Command command = resetCommand.serialize().build();
final Commands nextCommands = Commands.newBuilder().addCommands(command).build();
responseObserver.onNext(nextCommands);
diff --git a/oap-server/server-starter-es7/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java b/oap-server/server-starter-es7/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java
index 6dd180a..f91f5f8 100644
--- a/oap-server/server-starter-es7/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java
+++ b/oap-server/server-starter-es7/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java
@@ -18,8 +18,11 @@
package org.apache.skywalking.oap.server.starter;
+/**
+ * OAP starter specific for the ES7 storage. This includes the same code of OAPServerStartUp in the `server-starter`
+ * module.
+ */
public class OAPServerStartUp {
-
public static void main(String[] args) {
OAPServerBootstrap.start();
}
diff --git a/oap-server/server-starter/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java b/oap-server/server-starter/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java
index 6dd180a..84eebb0 100644
--- a/oap-server/server-starter/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java
+++ b/oap-server/server-starter/src/main/java/org/apache/skywalking/oap/server/starter/OAPServerStartUp.java
@@ -18,8 +18,10 @@
package org.apache.skywalking.oap.server.starter;
+/**
+ * OAP starter
+ */
public class OAPServerStartUp {
-
public static void main(String[] args) {
OAPServerBootstrap.start();
}
diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java
index 1fa6eac..f16f8b3 100644
--- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java
+++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java
@@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
import org.apache.skywalking.oap.server.core.analysis.metrics.IntKeyLongValueHashMap;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
@@ -35,12 +36,13 @@ import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariC
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLBuilder;
import org.apache.skywalking.oap.server.storage.plugin.jdbc.TableMetaInfo;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * H2 table initialization. Create tables without Indexes. H2 is for the demonstration only, so, keep the logic as
+ * simple as possible.
+ */
+@Slf4j
public class H2TableInstaller extends ModelInstaller {
- private static final Logger logger = LoggerFactory.getLogger(H2TableInstaller.class);
-
public H2TableInstaller(ModuleManager moduleManager) {
super(moduleManager);
}
@@ -71,14 +73,15 @@ public class H2TableInstaller extends ModelInstaller {
for (int i = 0; i < model.getColumns().size(); i++) {
ModelColumn column = model.getColumns().get(i);
ColumnName name = column.getColumnName();
- tableCreateSQL.appendLine(name.getStorageName() + " " + getColumnType(model, name, column.getType()) + (i != model
- .getColumns()
- .size() - 1 ? "," : ""));
+ tableCreateSQL.appendLine(
+ name.getStorageName() + " " + getColumnType(model, name, column.getType()) + (i != model
+ .getColumns()
+ .size() - 1 ? "," : ""));
}
tableCreateSQL.appendLine(")");
- if (logger.isDebugEnabled()) {
- logger.debug("creating table: " + tableCreateSQL.toStringInNewLine());
+ if (log.isDebugEnabled()) {
+ log.debug("creating table: " + tableCreateSQL.toStringInNewLine());
}
try (Connection connection = h2Client.getConnection()) {
@@ -91,6 +94,9 @@ public class H2TableInstaller extends ModelInstaller {
}
+ /**
+ * Set up the data type mapping between Java type and H2 database type
+ */
protected String getColumnType(Model model, ColumnName name, Class<?> type) {
if (Integer.class.equals(type) || int.class.equals(type)) {
return "INT";
diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java
index 55404d9..6960883 100644
--- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java
+++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java
@@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql;
import java.sql.Connection;
import java.sql.SQLException;
+import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord;
import org.apache.skywalking.oap.server.core.analysis.metrics.IntKeyLongValueHashMap;
import org.apache.skywalking.oap.server.core.register.RegisterSource;
@@ -33,8 +34,6 @@ import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariC
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLBuilder;
import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TableInstaller;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.ALARM;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.ENDPOINT_INVENTORY;
@@ -46,10 +45,8 @@ import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SE
/**
* Extend H2TableInstaller but match MySQL SQL syntax.
*/
+@Slf4j
public class MySQLTableInstaller extends H2TableInstaller {
-
- private static final Logger logger = LoggerFactory.getLogger(MySQLTableInstaller.class);
-
public MySQLTableInstaller(ModuleManager moduleManager) {
super(moduleManager);
/*
@@ -66,6 +63,9 @@ public class MySQLTableInstaller extends H2TableInstaller {
this.createIndexes(jdbcHikariCPClient, model);
}
+ /**
+ * Based on MySQL features, provide a specific data type mappings.
+ */
@Override
protected String getColumnType(Model model, ColumnName name, Class<?> type) {
if (Integer.class.equals(type) || int.class.equals(type)) {
@@ -92,6 +92,12 @@ public class MySQLTableInstaller extends H2TableInstaller {
}
}
+ /**
+ * Create indexes of all tables. Due to MySQL storage is suitable for middle size use case and also compatible with
+ * TiDB users, Indexes are required for the UI query.
+ *
+ * Based on different Model, provide different index creation strategy.
+ */
protected void createIndexes(JDBCHikariCPClient client, Model model) throws StorageException {
switch (model.getScopeId()) {
case SERVICE_INVENTORY:
@@ -198,9 +204,9 @@ public class MySQLTableInstaller extends H2TableInstaller {
}
private void createIndex(JDBCHikariCPClient client, Connection connection, Model model,
- SQLBuilder indexSQL) throws JDBCClientException {
- if (logger.isDebugEnabled()) {
- logger.debug("create index for table {}, sql: {} ", model.getName(), indexSQL.toStringInNewLine());
+ SQLBuilder indexSQL) throws JDBCClientException {
+ if (log.isDebugEnabled()) {
+ log.debug("create index for table {}, sql: {} ", model.getName(), indexSQL.toStringInNewLine());
}
client.execute(connection, indexSQL.toString());
}