diff --git a/docs/en/setup/backend/backend-setup.md b/docs/en/setup/backend/backend-setup.md index 5f3312dc0f..13edff0d4c 100644 --- a/docs/en/setup/backend/backend-setup.md +++ b/docs/en/setup/backend/backend-setup.md @@ -72,4 +72,8 @@ in storage based on rate. most of backend analysis capabilities based on the scripts. Here is the description of official scripts, which helps you to understand which metric data are in process, also could be used in alarm. 1. [Alarm](backend-alarm.md). Alarm provides a time-series based check mechanism. You could set alarm -rules targeting the analysis oal metric objects. \ No newline at end of file +rules targeting the analysis oal metric objects. + +## Telemetry for backend +OAP backend cluster itself underlying is a distributed streaming process system. For helping the Ops team, +we provide the telemetry for OAP backend itself. Follow [document](backend-telemetry.md) to use it. diff --git a/docs/en/setup/backend/backend-telemetry.md b/docs/en/setup/backend/backend-telemetry.md new file mode 100644 index 0000000000..842f8d2589 --- /dev/null +++ b/docs/en/setup/backend/backend-telemetry.md @@ -0,0 +1,20 @@ +# Telemetry for backend +In default, the telemetry is off, like this +```yaml +telemetry: + none: +``` + +In order to open, we should set `prometheus` to provider. The endpoint open at http://0.0.0.0:1234/ +```yaml +telemetry: + prometheus: +``` + +You could set host and port +```yaml +telemetry: + prometheus: + host: 127.0.0.1 + port: 1543 +``` \ No newline at end of file diff --git a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java index 7cf732c61c..5584a62ce3 100644 --- a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java +++ b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java @@ -19,22 +19,14 @@ package org.apache.skywalking.oap.server.cluster.plugin.consul; import com.google.common.base.Strings; -import com.orbitz.consul.AgentClient; -import com.orbitz.consul.Consul; -import com.orbitz.consul.HealthClient; -import com.orbitz.consul.model.agent.ImmutableRegistration; -import com.orbitz.consul.model.agent.Registration; +import com.orbitz.consul.*; +import com.orbitz.consul.model.agent.*; import com.orbitz.consul.model.health.ServiceHealth; -import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery; -import org.apache.skywalking.oap.server.core.cluster.ClusterRegister; -import org.apache.skywalking.oap.server.core.cluster.RemoteInstance; -import org.apache.skywalking.oap.server.core.cluster.ServiceRegisterException; +import java.util.*; +import org.apache.skywalking.oap.server.core.cluster.*; import org.apache.skywalking.oap.server.core.remote.client.Address; import org.apache.skywalking.oap.server.library.util.CollectionUtils; - -import java.util.ArrayList; -import java.util.List; -import java.util.Objects; +import org.apache.skywalking.oap.server.telemetry.api.TelemetryRelatedContext; /** * @author peng-yongsheng @@ -77,6 +69,7 @@ public class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery { AgentClient agentClient = client.agentClient(); this.selfAddress = remoteInstance.getAddress(); + TelemetryRelatedContext.INSTANCE.setId(selfAddress.toString()); Registration registration = ImmutableRegistration.builder() .id(remoteInstance.getAddress().toString()) diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java index 8fdc5f863f..730f4b28d2 100644 --- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java +++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java @@ -26,6 +26,7 @@ import java.util.function.Supplier; import javax.annotation.Nullable; import org.apache.skywalking.oap.server.core.cluster.*; import org.apache.skywalking.oap.server.core.remote.client.Address; +import org.apache.skywalking.oap.server.telemetry.api.TelemetryRelatedContext; import org.slf4j.*; /** @@ -49,6 +50,7 @@ public class KubernetesCoordinator implements ClusterRegister, ClusterNodesQuery KubernetesCoordinator(final ReusableWatch watch, final Supplier uidSupplier) { this.watch = watch; this.uid = uidSupplier.get(); + TelemetryRelatedContext.INSTANCE.setId(uid); } @Override public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException { diff --git a/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java b/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java index ef30870be9..b70873f858 100644 --- a/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java +++ b/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java @@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.cluster.plugin.standalone; import java.util.*; import org.apache.skywalking.oap.server.core.cluster.*; +import org.apache.skywalking.oap.server.telemetry.api.TelemetryRelatedContext; /** * A cluster manager simulator. Work in memory only. Also return the current instance. @@ -33,6 +34,7 @@ public class StandaloneManager implements ClusterNodesQuery, ClusterRegister { @Override public void registerRemote(RemoteInstance remoteInstance) { this.remoteInstance = remoteInstance; this.remoteInstance.getAddress().setSelf(true); + TelemetryRelatedContext.INSTANCE.setId("standalone"); } @Override diff --git a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java index 826bdbb4a4..837f3f338e 100644 --- a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java +++ b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java @@ -22,6 +22,7 @@ import java.util.*; import org.apache.curator.x.discovery.*; import org.apache.skywalking.oap.server.core.cluster.*; import org.apache.skywalking.oap.server.core.remote.client.Address; +import org.apache.skywalking.oap.server.telemetry.api.TelemetryRelatedContext; import org.slf4j.*; /** @@ -59,6 +60,7 @@ public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery serviceCache.start(); this.selfAddress = remoteInstance.getAddress(); + TelemetryRelatedContext.INSTANCE.setId(selfAddress.toString()); } catch (Exception e) { throw new ServiceRegisterException(e.getMessage()); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java index 0f026519fe..9d28c1c9b5 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java @@ -26,6 +26,9 @@ import org.apache.skywalking.oap.server.core.analysis.data.EndOfBatchContext; import org.apache.skywalking.oap.server.core.analysis.data.MergeDataCache; import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; import org.apache.skywalking.oap.server.core.worker.AbstractWorker; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,14 +44,19 @@ public class IndicatorAggregateWorker extends AbstractWorker { private final MergeDataCache mergeDataCache; private int messageNum; private final String modelName; + private CounterMetric aggregationCounter; - IndicatorAggregateWorker(int workerId, AbstractWorker nextWorker, String modelName) { + IndicatorAggregateWorker(ModuleManager moduleManager, int workerId, AbstractWorker nextWorker, String modelName) { super(workerId); this.modelName = modelName; this.nextWorker = nextWorker; this.mergeDataCache = new MergeDataCache<>(); this.dataCarrier = new DataCarrier<>("IndicatorAggregateWorker." + modelName, 1, 10000); this.dataCarrier.consume(new AggregatorConsumer(this), 1); + + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + aggregationCounter = metricCreator.createCounter("indicator_aggregation", "The number of rows in aggregation", + new MetricTag.Keys("metricName", "level", "dimensionality"), new MetricTag.Values(modelName, "1", "min")); } @Override public final void in(Indicator indicator) { @@ -57,6 +65,7 @@ public class IndicatorAggregateWorker extends AbstractWorker { } private void onWork(Indicator indicator) { + aggregationCounter.inc(); messageNum++; aggregate(indicator); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorProcess.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorProcess.java index 8b5a0b90b6..4baea5831b 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorProcess.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorProcess.java @@ -57,13 +57,13 @@ public enum IndicatorProcess { IndicatorPersistentWorker dayPersistentWorker = worker(moduleManager, indicatorDAO, modelName + Const.ID_SPLIT + Downsampling.Day.getName()); IndicatorPersistentWorker monthPersistentWorker = worker(moduleManager, indicatorDAO, modelName + Const.ID_SPLIT + Downsampling.Month.getName()); - IndicatorTransWorker transWorker = new IndicatorTransWorker(WorkerIdGenerator.INSTANCES.generate(), minutePersistentWorker, hourPersistentWorker, dayPersistentWorker, monthPersistentWorker); + IndicatorTransWorker transWorker = new IndicatorTransWorker(moduleManager, modelName, WorkerIdGenerator.INSTANCES.generate(), minutePersistentWorker, hourPersistentWorker, dayPersistentWorker, monthPersistentWorker); WorkerInstances.INSTANCES.put(transWorker.getWorkerId(), transWorker); IndicatorRemoteWorker remoteWorker = new IndicatorRemoteWorker(WorkerIdGenerator.INSTANCES.generate(), moduleManager, transWorker, modelName); WorkerInstances.INSTANCES.put(remoteWorker.getWorkerId(), remoteWorker); - IndicatorAggregateWorker aggregateWorker = new IndicatorAggregateWorker(WorkerIdGenerator.INSTANCES.generate(), remoteWorker, modelName); + IndicatorAggregateWorker aggregateWorker = new IndicatorAggregateWorker(moduleManager, WorkerIdGenerator.INSTANCES.generate(), remoteWorker, modelName); WorkerInstances.INSTANCES.put(aggregateWorker.getWorkerId(), aggregateWorker); entryWorkers.put(indicatorClass, aggregateWorker); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorTransWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorTransWorker.java index 537d6a4579..526254d4b7 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorTransWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorTransWorker.java @@ -21,6 +21,9 @@ package org.apache.skywalking.oap.server.core.analysis.worker; import java.util.Objects; import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; import org.apache.skywalking.oap.server.core.worker.AbstractWorker; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.*; /** @@ -35,7 +38,14 @@ public class IndicatorTransWorker extends AbstractWorker { private final IndicatorPersistentWorker dayPersistenceWorker; private final IndicatorPersistentWorker monthPersistenceWorker; - public IndicatorTransWorker(int workerId, + private CounterMetric aggregationMinCounter; + private CounterMetric aggregationHourCounter; + private CounterMetric aggregationDayCounter; + private CounterMetric aggregationMonthCounter; + + public IndicatorTransWorker(ModuleManager moduleManager, + String modelName, + int workerId, IndicatorPersistentWorker minutePersistenceWorker, IndicatorPersistentWorker hourPersistenceWorker, IndicatorPersistentWorker dayPersistenceWorker, @@ -45,16 +55,29 @@ public class IndicatorTransWorker extends AbstractWorker { this.hourPersistenceWorker = hourPersistenceWorker; this.dayPersistenceWorker = dayPersistenceWorker; this.monthPersistenceWorker = monthPersistenceWorker; + + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + aggregationMinCounter = metricCreator.createCounter("indicator_aggregation", "The number of rows in aggregation", + new MetricTag.Keys("metricName", "level", "dimensionality"), new MetricTag.Values(modelName, "2", "min")); + aggregationHourCounter = metricCreator.createCounter("indicator_aggregation", "The number of rows in aggregation", + new MetricTag.Keys("metricName", "level", "dimensionality"), new MetricTag.Values(modelName, "2", "hour")); + aggregationDayCounter = metricCreator.createCounter("indicator_aggregation", "The number of rows in aggregation", + new MetricTag.Keys("metricName", "level", "dimensionality"), new MetricTag.Values(modelName, "2", "day")); + aggregationMonthCounter = metricCreator.createCounter("indicator_aggregation", "The number of rows in aggregation", + new MetricTag.Keys("metricName", "level", "dimensionality"), new MetricTag.Values(modelName, "2", "month")); } @Override public void in(Indicator indicator) { if (Objects.nonNull(hourPersistenceWorker)) { + aggregationMonthCounter.inc(); hourPersistenceWorker.in(indicator.toHour()); } if (Objects.nonNull(dayPersistenceWorker)) { + aggregationDayCounter.inc(); dayPersistenceWorker.in(indicator.toDay()); } if (Objects.nonNull(monthPersistenceWorker)) { + aggregationHourCounter.inc(); monthPersistenceWorker.in(indicator.toMonth()); } /** @@ -62,6 +85,7 @@ public class IndicatorTransWorker extends AbstractWorker { * Because #toHour, #toDay, #toMonth include clone inside, which could avoid concurrency situation. */ if (Objects.nonNull(minutePersistenceWorker)) { + aggregationMinCounter.inc(); minutePersistenceWorker.in(indicator); } } 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 de71610652..89d301c53c 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 @@ -27,13 +27,14 @@ import org.apache.skywalking.oap.server.core.remote.grpc.proto.*; import org.apache.skywalking.oap.server.core.worker.WorkerInstances; import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder; import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.*; /** - * This class is Server-side streaming RPC implementation. It's a common service for OAP servers - * to receive message from each others. - * The stream data id is used to find the object to deserialize message. - * The next worker id is used to find the worker to process message. + * This class is Server-side streaming RPC implementation. It's a common service for OAP servers to receive message from + * each others. The stream data id is used to find the object to deserialize message. The next worker id is used to find + * the worker to process message. * * @author peng-yongsheng */ @@ -43,9 +44,22 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas private final ModuleDefineHolder moduleDefineHolder; private StreamDataClassGetter streamDataClassGetter; + private CounterMetric remoteInCounter; + private CounterMetric remoteInErrorCounter; + private HistogramMetric remoteInHistogram; public RemoteServiceHandler(ModuleDefineHolder moduleDefineHolder) { this.moduleDefineHolder = moduleDefineHolder; + + remoteInCounter = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricCreator.class) + .createCounter("remote_in_count", "The number(server side) of inside remote inside aggregate rpc.", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + remoteInErrorCounter = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricCreator.class) + .createCounter("remote_in_error_count", "The error number(server side) of inside remote inside aggregate rpc.", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + remoteInHistogram = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricCreator.class) + .createHistogramMetric("remote_in_latency", "The latency(server side) of inside remote inside aggregate rpc.", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); } @Override public StreamObserver call(StreamObserver responseObserver) { @@ -59,17 +73,24 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas return new StreamObserver() { @Override public void onNext(RemoteMessage message) { - int streamDataId = message.getStreamDataId(); - int nextWorkerId = message.getNextWorkerId(); - RemoteData remoteData = message.getRemoteData(); - - Class streamDataClass = streamDataClassGetter.findClassById(streamDataId); + remoteInCounter.inc(); + HistogramMetric.Timer timer = remoteInHistogram.createTimer(); try { - StreamData streamData = streamDataClass.newInstance(); - streamData.deserialize(remoteData); - WorkerInstances.INSTANCES.get(nextWorkerId).in(streamData); - } catch (Throwable t) { - logger.error(t.getMessage(), t); + int streamDataId = message.getStreamDataId(); + int nextWorkerId = message.getNextWorkerId(); + RemoteData remoteData = message.getRemoteData(); + + Class streamDataClass = streamDataClassGetter.findClassById(streamDataId); + try { + StreamData streamData = streamDataClass.newInstance(); + streamData.deserialize(remoteData); + WorkerInstances.INSTANCES.get(nextWorkerId).in(streamData); + } catch (Throwable t) { + remoteInErrorCounter.inc(); + logger.error(t.getMessage(), t); + } + } finally { + timer.finish(); } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java index b061ca69cc..6b3cf4f1fc 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java @@ -29,6 +29,9 @@ import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataClassGe import org.apache.skywalking.oap.server.core.remote.data.StreamData; import org.apache.skywalking.oap.server.core.remote.grpc.proto.*; import org.apache.skywalking.oap.server.library.client.grpc.GRPCClient; +import org.apache.skywalking.oap.server.library.module.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.*; /** @@ -49,13 +52,23 @@ public class GRPCRemoteClient implements RemoteClient { private GRPCClient client; private DataCarrier carrier; private boolean isConnect; + private CounterMetric remoteOutCounter; + private CounterMetric remoteOutErrorCounter; - public GRPCRemoteClient(StreamDataClassGetter streamDataClassGetter, Address address, int channelSize, + + public GRPCRemoteClient(ModuleDefineHolder moduleDefineHolder, StreamDataClassGetter streamDataClassGetter, Address address, int channelSize, int bufferSize) { this.streamDataClassGetter = streamDataClassGetter; this.address = address; this.channelSize = channelSize; this.bufferSize = bufferSize; + + remoteOutCounter = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricCreator.class) + .createCounter("remote_out_count", "The number(client side) of inside remote inside aggregate rpc.", + new MetricTag.Keys("dest", "self"), new MetricTag.Values(address.toString(), "N")); + remoteOutErrorCounter = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricCreator.class) + .createCounter("remote_out_error_count", "The error number(client side) of inside remote inside aggregate rpc.", + new MetricTag.Keys("dest", "self"), new MetricTag.Values(address.toString(), "N")); } @Override public void connect() { @@ -126,10 +139,12 @@ public class GRPCRemoteClient implements RemoteClient { try { StreamObserver streamObserver = createStreamObserver(); for (RemoteMessage remoteMessage : remoteMessages) { + remoteOutCounter.inc(); streamObserver.onNext(remoteMessage); } streamObserver.onCompleted(); } catch (Throwable t) { + remoteOutErrorCounter.inc(); logger.error(t.getMessage(), t); } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java index 2510a11881..07f7ec562a 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java @@ -24,12 +24,13 @@ import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.cluster.*; import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataClassGetter; import org.apache.skywalking.oap.server.library.module.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.*; /** - * This class manages the connections between OAP servers. There is a task schedule that will - * automatically query a server list from the cluster module. Such as Zookeeper cluster module - * or Kubernetes cluster module. + * This class manages the connections between OAP servers. There is a task schedule that will automatically query a + * server list from the cluster module. Such as Zookeeper cluster module or Kubernetes cluster module. * * @author peng-yongsheng */ @@ -43,6 +44,7 @@ public class RemoteClientManager implements Service { private final List clientsA; private final List clientsB; private volatile List usingClients; + private GaugeMetric gauge; public RemoteClientManager(ModuleDefineHolder moduleDefineHolder) { this.moduleDefineHolder = moduleDefineHolder; @@ -56,11 +58,15 @@ public class RemoteClientManager implements Service { } /** - * Query OAP server list from the cluster module and create a new connection - * for the new node. Make the OAP server orderly because of each of the server - * will send stream data to each other by hash code. + * Query OAP server list from the cluster module and create a new connection for the new node. Make the OAP server + * orderly because of each of the server will send stream data to each other by hash code. */ void refresh() { + if (gauge == null) { + gauge = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricCreator.class) + .createGauge("cluster_size", "Cluster size of current oap node", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + } try { if (Objects.isNull(clusterNodesQuery)) { synchronized (RemoteClientManager.class) { @@ -86,6 +92,8 @@ public class RemoteClientManager implements Service { instanceList = distinct(instanceList); Collections.sort(instanceList); + gauge.setValue(instanceList.size()); + if (logger.isDebugEnabled()) { instanceList.forEach(instance -> logger.debug("Cluster instance: {}", instance.toString())); } @@ -115,11 +123,10 @@ public class RemoteClientManager implements Service { } /** - * Because of OAP server register by the UUID which one-to-one mapping with process number. - * The register information not delete immediately after process shutdown because of there - * is always happened network fault, not really process shutdown. So, cluster module must - * wait a few seconds to confirm it. Then there are more than one register information in - * the cluster. + * Because of OAP server register by the UUID which one-to-one mapping with process number. The register information + * not delete immediately after process shutdown because of there is always happened network fault, not really + * process shutdown. So, cluster module must wait a few seconds to confirm it. Then there are more than one register + * information in the cluster. * * @param instanceList the instances query from cluster module. * @return distinct remote instances @@ -156,9 +163,8 @@ public class RemoteClientManager implements Service { } /** - * Compare clients between exist clients and remote instance collection. Move - * the clients into new client collection which are alive to avoid create a - * new channel. Shutdown the clients which could not find in cluster config. + * Compare clients between exist clients and remote instance collection. Move the clients into new client collection + * which are alive to avoid create a new channel. Shutdown the clients which could not find in cluster config. * * Create a gRPC client for remote instance except for self-instance. * @@ -190,10 +196,10 @@ public class RemoteClientManager implements Service { break; case Create: if (address.isSelf()) { - RemoteClient client = new SelfRemoteClient(address); + RemoteClient client = new SelfRemoteClient(moduleDefineHolder, address); getFreeClients().add(client); } else { - RemoteClient client = new GRPCRemoteClient(streamDataClassGetter, address, 1, 3000); + RemoteClient client = new GRPCRemoteClient(moduleDefineHolder, streamDataClassGetter, address, 1, 3000); client.connect(); getFreeClients().add(client); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java index ab25f5cde7..ad253ca967 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java @@ -21,6 +21,9 @@ package org.apache.skywalking.oap.server.core.remote.client; import org.apache.skywalking.oap.server.core.UnexpectedException; import org.apache.skywalking.oap.server.core.remote.data.StreamData; import org.apache.skywalking.oap.server.core.worker.WorkerInstances; +import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; /** * @author peng-yongsheng @@ -28,9 +31,13 @@ import org.apache.skywalking.oap.server.core.worker.WorkerInstances; public class SelfRemoteClient implements RemoteClient { private final Address address; + private CounterMetric remoteOutCounter; - public SelfRemoteClient(Address address) { + public SelfRemoteClient(ModuleDefineHolder moduleDefineHolder, Address address) { this.address = address; + remoteOutCounter = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricCreator.class) + .createCounter("remote_out_count", "The number(client side) of inside remote inside aggregate rpc.", + new MetricTag.Keys("dest", "self"), new MetricTag.Values(address.toString(), "Y")); } @Override public Address getAddress() { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java index 63d3a81140..522f9d5911 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java @@ -23,6 +23,8 @@ import java.util.concurrent.*; import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; import org.apache.skywalking.oap.server.core.analysis.worker.*; import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.*; /** @@ -35,6 +37,11 @@ public enum PersistenceTimer { private Boolean isStarted = false; private final Boolean debug; + private CounterMetric errorCounter; + private CounterMetric prepareCounter; + private HistogramMetric prepareLatency; + private CounterMetric executeCounter; + private HistogramMetric executeLatency; PersistenceTimer() { this.debug = System.getProperty("debug") != null; @@ -47,6 +54,18 @@ public enum PersistenceTimer { final long timeInterval = 3; IBatchDAO batchDAO = moduleManager.find(StorageModule.NAME).provider().getService(IBatchDAO.class); + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + errorCounter = metricCreator.createCounter("persistence_timer_bulk_error_count", "Error execution of the prepare stage in persistence timer", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + prepareCounter = metricCreator.createCounter("persistence_timer_bulk_prepare_count", "Execution of the prepare stage in persistence timer", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + prepareLatency = metricCreator.createHistogramMetric("persistence_timer_bulk_prepare_latency", "Latency of the prepare stage in persistence timer", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + executeCounter = metricCreator.createCounter("persistence_timer_bulk_execute_count", "Execution of the execute stage in persistence timer", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + executeLatency = metricCreator.createHistogramMetric("persistence_timer_bulk_execute_latency", "Latency of the execute stage in persistence timer", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + if (!isStarted) { Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( new RunnableWithExceptionProtection(() -> extractDataAndSave(batchDAO), @@ -64,31 +83,46 @@ public enum PersistenceTimer { long startTime = System.currentTimeMillis(); try { + prepareCounter.inc(); + HistogramMetric.Timer timer = prepareLatency.createTimer(); + List batchAllCollection = new LinkedList(); - List persistenceWorkers = new ArrayList<>(); - persistenceWorkers.addAll(IndicatorProcess.INSTANCE.getPersistentWorkers()); - persistenceWorkers.addAll(RecordProcess.INSTANCE.getPersistentWorkers()); - - persistenceWorkers.forEach(worker -> { - if (logger.isDebugEnabled()) { - logger.debug("extract {} worker data and save", worker.getClass().getName()); - } - - if (worker.flushAndSwitch()) { - List batchCollection = worker.buildBatchCollection(); + try { + List persistenceWorkers = new ArrayList<>(); + persistenceWorkers.addAll(IndicatorProcess.INSTANCE.getPersistentWorkers()); + persistenceWorkers.addAll(RecordProcess.INSTANCE.getPersistentWorkers()); + persistenceWorkers.forEach(worker -> { if (logger.isDebugEnabled()) { - logger.debug("extract {} worker data size: {}", worker.getClass().getName(), batchCollection.size()); + logger.debug("extract {} worker data and save", worker.getClass().getName()); } - batchAllCollection.addAll(batchCollection); - } - }); - if (debug) { - logger.info("build batch persistence duration: {} ms", System.currentTimeMillis() - startTime); + if (worker.flushAndSwitch()) { + List batchCollection = worker.buildBatchCollection(); + + if (logger.isDebugEnabled()) { + logger.debug("extract {} worker data size: {}", worker.getClass().getName(), batchCollection.size()); + } + batchAllCollection.addAll(batchCollection); + } + }); + + if (debug) { + logger.info("build batch persistence duration: {} ms", System.currentTimeMillis() - startTime); + } + } finally { + timer.finish(); + } + + executeCounter.inc(); + HistogramMetric.Timer executeLatencyTimer = executeLatency.createTimer(); + try { + batchDAO.batchPersistence(batchAllCollection); + } finally { + executeLatencyTimer.finish(); } - batchDAO.batchPersistence(batchAllCollection); } catch (Throwable e) { + errorCounter.inc(); logger.error(e.getMessage(), e); } finally { if (logger.isDebugEnabled()) { diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java index b2e1f03541..5b796864e0 100644 --- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java @@ -28,6 +28,8 @@ import org.apache.skywalking.oap.server.core.remote.data.StreamData; import org.apache.skywalking.oap.server.core.remote.grpc.proto.*; import org.apache.skywalking.oap.server.core.worker.*; import org.apache.skywalking.oap.server.library.module.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.apache.skywalking.oap.server.testing.module.*; import org.junit.*; @@ -59,6 +61,26 @@ public class RemoteServiceHandlerTestCase { WorkerInstances.INSTANCES.put(testWorkerId, new TestWorker()); String serverName = InProcessServerBuilder.generateName(); + MetricCreator metricCreator = mock(MetricCreator.class); + when(metricCreator.createCounter(any(), any(), any(), any())).thenReturn(new CounterMetric() { + @Override public void inc() { + + } + + @Override public void inc(double value) { + + } + }); + when(metricCreator.createHistogramMetric(any(), any(), any(), any(), any())).thenReturn( + new HistogramMetric() { + @Override public void observe(double value) { + + } + } + ); + ModuleDefineTesting telemetryModuleDefine = new ModuleDefineTesting(); + moduleManager.put(TelemetryModule.NAME, telemetryModuleDefine); + telemetryModuleDefine.provider().registerServiceImplementation(MetricCreator.class, metricCreator); gRPCCleanup.register(InProcessServerBuilder .forName(serverName).directExecutor().addService(new RemoteServiceHandler(moduleManager)).build().start()); diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java index 41953fb6d2..b704d7bc9d 100644 --- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java @@ -23,9 +23,12 @@ import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataClassGe import org.apache.skywalking.oap.server.core.remote.data.StreamData; import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData; import org.apache.skywalking.oap.server.core.worker.AbstractWorker; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; +import org.apache.skywalking.oap.server.testing.module.*; import org.junit.Assert; -import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.*; /** * @author peng-yongsheng @@ -34,7 +37,22 @@ public class GRPCRemoteClientRealClient { public static void main(String[] args) throws InterruptedException { Address address = new Address("localhost", 10000, false); - GRPCRemoteClient remoteClient = spy(new GRPCRemoteClient(new TestClassGetter(), address, 1, 10)); + ModuleManagerTesting moduleManager = new ModuleManagerTesting(); + MetricCreator metricCreator = mock(MetricCreator.class); + when(metricCreator.createCounter(any(), any(), any(), any())).thenReturn(new CounterMetric() { + @Override public void inc() { + + } + + @Override public void inc(double value) { + + } + }); + ModuleDefineTesting telemetryModuleDefine = new ModuleDefineTesting(); + moduleManager.put(TelemetryModule.NAME, telemetryModuleDefine); + telemetryModuleDefine.provider().registerServiceImplementation(MetricCreator.class, metricCreator); + + GRPCRemoteClient remoteClient = spy(new GRPCRemoteClient(moduleManager, new TestClassGetter(), address, 1, 10)); remoteClient.connect(); for (int i = 0; i < 10000; i++) { diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java index e14f29d743..550c45d387 100644 --- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java @@ -26,6 +26,8 @@ import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataClassGe import org.apache.skywalking.oap.server.core.remote.data.StreamData; import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData; import org.apache.skywalking.oap.server.core.worker.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.apache.skywalking.oap.server.testing.module.*; import org.junit.*; @@ -56,10 +58,24 @@ public class GRPCRemoteClientTestCase { @Test public void testPush() throws InterruptedException { + MetricCreator metricCreator = mock(MetricCreator.class); + when(metricCreator.createCounter(any(), any(), any(), any())).thenReturn(new CounterMetric() { + @Override public void inc() { + + } + + @Override public void inc(double value) { + + } + }); + ModuleDefineTesting telemetryModuleDefine = new ModuleDefineTesting(); + moduleManager.put(TelemetryModule.NAME, telemetryModuleDefine); + telemetryModuleDefine.provider().registerServiceImplementation(MetricCreator.class, metricCreator); + grpcServerRule.getServiceRegistry().addService(new RemoteServiceHandler(moduleManager)); Address address = new Address("not-important", 11, false); - GRPCRemoteClient remoteClient = spy(new GRPCRemoteClient(classGetter, address, 1, 10)); + GRPCRemoteClient remoteClient = spy(new GRPCRemoteClient(moduleManager, classGetter, address, 1, 10)); remoteClient.connect(); doReturn(grpcServerRule.getChannel()).when(remoteClient).getChannel(); diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java index 54b6cb7580..4b6317b499 100644 --- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManagerTestCase.java @@ -22,6 +22,8 @@ import java.util.*; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.cluster.*; import org.apache.skywalking.oap.server.core.remote.annotation.StreamDataClassGetter; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.apache.skywalking.oap.server.testing.module.*; import org.junit.*; @@ -47,6 +49,32 @@ public class RemoteClientManagerTestCase { StreamDataClassGetter streamDataClassGetter = mock(StreamDataClassGetter.class); coreModuleDefine.provider().registerServiceImplementation(StreamDataClassGetter.class, streamDataClassGetter); + MetricCreator metricCreator = mock(MetricCreator.class); + when(metricCreator.createGauge(any(), any(), any(), any())).thenReturn(new GaugeMetric() { + @Override public void inc() { + + } + + @Override public void inc(double value) { + + } + + @Override public void dec() { + + } + + @Override public void dec(double value) { + + } + + @Override public void setValue(double value) { + + } + }); + ModuleDefineTesting telemetryModuleDefine = new ModuleDefineTesting(); + moduleManager.put(TelemetryModule.NAME, telemetryModuleDefine); + telemetryModuleDefine.provider().registerServiceImplementation(MetricCreator.class, metricCreator); + RemoteClientManager clientManager = new RemoteClientManager(moduleManager); when(clusterNodesQuery.queryRemoteNodes()).thenReturn(groupOneInstances()); diff --git a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java index 6951e960d8..668698cf84 100644 --- a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java +++ b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java @@ -21,19 +21,18 @@ package org.apache.skywalking.oap.server.receiver.istio.telemetry.provider; import com.google.common.base.Joiner; import com.google.protobuf.Timestamp; import io.grpc.stub.StreamObserver; -import io.istio.HandleMetricServiceGrpc; -import io.istio.IstioMetricProto; +import io.istio.*; import io.istio.api.mixer.adapter.model.v1beta1.ReportProto; import io.istio.api.policy.v1beta1.TypeProto; -import java.time.Duration; -import java.time.Instant; +import java.time.*; import java.util.Map; import org.apache.skywalking.aop.server.receiver.mesh.TelemetryDataDispatcher; import org.apache.skywalking.apm.network.common.DetectPoint; -import org.apache.skywalking.apm.network.servicemesh.Protocol; -import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.apache.skywalking.apm.network.servicemesh.*; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; +import org.slf4j.*; /** * Handle istio telemetry data. @@ -46,49 +45,82 @@ public class IstioTelemetryGRPCHandler extends HandleMetricServiceGrpc.HandleMet private static final Joiner JOINER = Joiner.on("."); + private CounterMetric counter; + private HistogramMetric histogram; + + public IstioTelemetryGRPCHandler(ModuleManager moduleManager) { + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + counter = metricCreator.createCounter("istio_mesh_grpc_in_count", "The count of istio service mesh telemetry", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + histogram = metricCreator.createHistogramMetric("istio_mesh_grpc_in_latency", "The process latency of istio service mesh telemetry", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + } + @Override public void handleMetric(IstioMetricProto.HandleMetricRequest request, StreamObserver responseObserver) { if (logger.isDebugEnabled()) { logger.debug("Received msg {}", request); } for (IstioMetricProto.InstanceMsg i : request.getInstancesList()) { - String requestMethod = string(i, "requestMethod"); - String requestPath = string(i, "requestPath"); - String requestScheme = string(i, "requestScheme"); - long responseCode = int64(i, "responseCode"); - String reporter = string(i, "reporter"); - String protocol = string(i, "apiProtocol"); + counter.inc(); + HistogramMetric.Timer timer = histogram.createTimer(); - String endpoint; - boolean status = true; - Protocol netProtocol; - if (protocol.equals("http") || protocol.equals("https") || requestScheme.equals("http") || requestScheme.equals("https")) { - endpoint = requestScheme + "/" + requestMethod + "/" + requestPath; - status = responseCode >= 200 && responseCode < 400; - netProtocol = Protocol.HTTP; - } else { - //grpc - endpoint = protocol + "/" + requestPath; - netProtocol = Protocol.gRPC; + try { + String requestMethod = string(i, "requestMethod"); + String requestPath = string(i, "requestPath"); + String requestScheme = string(i, "requestScheme"); + long responseCode = int64(i, "responseCode"); + String reporter = string(i, "reporter"); + String protocol = string(i, "apiProtocol"); + + String endpoint; + boolean status = true; + Protocol netProtocol; + if (protocol.equals("http") || protocol.equals("https") || requestScheme.equals("http") || requestScheme.equals("https")) { + endpoint = requestScheme + "/" + requestMethod + "/" + requestPath; + status = responseCode >= 200 && responseCode < 400; + netProtocol = Protocol.HTTP; + } else { + //grpc + endpoint = protocol + "/" + requestPath; + netProtocol = Protocol.gRPC; + } + Instant requestTime = time(i, "requestTime"); + Instant responseTime = time(i, "responseTime"); + int latency = Math.toIntExact(Duration.between(requestTime, responseTime).toMillis()); + + DetectPoint detectPoint; + if (reporter.equals("source")) { + detectPoint = DetectPoint.client; + } else { + detectPoint = DetectPoint.server; + } + + String sourceServiceName; + if (has(i, "sourceNamespace")) { + sourceServiceName = JOINER.join(string(i, "sourceService"), string(i, "sourceNamespace")); + } else { + sourceServiceName = string(i, "sourceService"); + } + + String destServiceName; + if (has(i, "destinationNamespace")) { + destServiceName = JOINER.join(string(i, "destinationService"), string(i, "destinationNamespace")); + } else { + destServiceName = string(i, "destinationService"); + } + + ServiceMeshMetric metric = ServiceMeshMetric.newBuilder().setStartTime(requestTime.toEpochMilli()) + .setEndTime(responseTime.toEpochMilli()).setSourceServiceName(sourceServiceName) + .setSourceServiceInstance(string(i, "sourceUID")).setDestServiceName(destServiceName) + .setDestServiceInstance(string(i, "destinationUID")).setEndpoint(endpoint).setLatency(latency) + .setResponseCode(Math.toIntExact(responseCode)).setStatus(status).setProtocol(netProtocol).setDetectPoint(detectPoint).build(); + logger.debug("Transformed metric {}", metric); + + TelemetryDataDispatcher.preProcess(metric); + } finally { + timer.finish(); } - Instant requestTime = time(i, "requestTime"); - Instant responseTime = time(i, "responseTime"); - int latency = Math.toIntExact(Duration.between(requestTime, responseTime).toMillis()); - - DetectPoint detectPoint; - if (reporter.equals("source")) { - detectPoint = DetectPoint.client; - } else { - detectPoint = DetectPoint.server; - } - ServiceMeshMetric metric = ServiceMeshMetric.newBuilder().setStartTime(requestTime.toEpochMilli()) - .setEndTime(responseTime.toEpochMilli()).setSourceServiceName(JOINER.join(string(i, "sourceService"), string(i, "sourceNamespace"))) - .setSourceServiceInstance(string(i, "sourceUID")).setDestServiceName(JOINER.join(string(i, "destinationService"), string(i, "destinationNamespace"))) - .setDestServiceInstance(string(i, "destinationUID")).setEndpoint(endpoint).setLatency(latency) - .setResponseCode(Math.toIntExact(responseCode)).setStatus(status).setProtocol(netProtocol).setDetectPoint(detectPoint).build(); - logger.debug("Transformed metric {}", metric); - - TelemetryDataDispatcher.preProcess(metric); } responseObserver.onNext(ReportProto.ReportResult.newBuilder().build()); responseObserver.onCompleted(); @@ -118,4 +150,9 @@ public class IstioTelemetryGRPCHandler extends HandleMetricServiceGrpc.HandleMet throw new IllegalArgumentException(String.format("Lack dimension %s", key)); } } + + private boolean has(final IstioMetricProto.InstanceMsg instanceMsg, final String key) { + Map map = instanceMsg.getDimensionsMap(); + return map.containsKey(key); + } } diff --git a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java index 13a55ff1f8..3b5afdbc18 100644 --- a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java +++ b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java @@ -22,8 +22,8 @@ import org.apache.skywalking.aop.server.receiver.mesh.MeshReceiverModule; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister; import org.apache.skywalking.oap.server.library.module.*; -import org.apache.skywalking.oap.server.library.module.ModuleDefine; import org.apache.skywalking.oap.server.receiver.istio.telemetry.module.IstioTelemetryReceiverModule; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; public class IstioTelemetryReceiverProvider extends ModuleProvider { @Override public String name() { @@ -43,7 +43,7 @@ public class IstioTelemetryReceiverProvider extends ModuleProvider { @Override public void start() throws ServiceNotProvidedException, ModuleStartException { GRPCHandlerRegister service = getManager().find(CoreModule.NAME).provider().getService(GRPCHandlerRegister.class); - service.addHandler(new IstioTelemetryGRPCHandler()); + service.addHandler(new IstioTelemetryGRPCHandler(getManager())); } @Override public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException { @@ -51,6 +51,6 @@ public class IstioTelemetryReceiverProvider extends ModuleProvider { } @Override public String[] requiredModules() { - return new String[] {CoreModule.NAME, MeshReceiverModule.NAME}; + return new String[] {TelemetryModule.NAME, CoreModule.NAME, MeshReceiverModule.NAME}; } } \ No newline at end of file diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java index 8089fe23a0..92e97b4fe5 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java @@ -25,15 +25,28 @@ import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer; import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric; import org.apache.skywalking.oap.server.library.buffer.BufferStream; import org.apache.skywalking.oap.server.library.buffer.DataStreamReader; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; public class MeshDataBufferFileCache implements IConsumer, DataStreamReader.CallBack { private MeshModuleConfig config; private DataCarrier dataCarrier; private BufferStream stream; + private CounterMetric meshBufferFileIn; + private CounterMetric meshBufferFileRetry; + private CounterMetric meshBufferFileOut; - public MeshDataBufferFileCache(MeshModuleConfig config) { + public MeshDataBufferFileCache(MeshModuleConfig config, ModuleManager moduleManager) { this.config = config; dataCarrier = new DataCarrier<>("MeshDataBufferFileCache", 3, 1024); + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + meshBufferFileIn = metricCreator.createCounter("mesh_buffer_file_in", "The number of mesh telemetry into the buffer file", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + meshBufferFileRetry = metricCreator.createCounter("mesh_buffer_file_retry", "The number of retry mesh telemetry from the buffer file, but haven't registered successfully.", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + meshBufferFileOut = metricCreator.createCounter("mesh_buffer_file_out", "The number of mesh telemetry out of the buffer file", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); } void start() throws IOException { @@ -67,6 +80,7 @@ public class MeshDataBufferFileCache implements IConsumer collect(StreamObserver responseObserver) { return new StreamObserver() { @@ -35,7 +46,13 @@ public class MeshGRPCHandler extends ServiceMeshMetricServiceGrpc.ServiceMeshMet if (logger.isDebugEnabled()) { logger.debug("Received mesh metric: {}", metric); } - TelemetryDataDispatcher.preProcess(metric); + counter.inc(); + HistogramMetric.Timer timer = histogram.createTimer(); + try { + TelemetryDataDispatcher.preProcess(metric); + } finally { + timer.finish(); + } } @Override public void onError(Throwable throwable) { diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java index cbafbf7aa0..d221fe2b1c 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java @@ -23,6 +23,7 @@ import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister; import org.apache.skywalking.oap.server.library.module.*; import org.apache.skywalking.oap.server.library.module.ModuleDefine; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; public class MeshReceiverProvider extends ModuleProvider { private MeshModuleConfig config; @@ -47,7 +48,7 @@ public class MeshReceiverProvider extends ModuleProvider { } @Override public void start() throws ServiceNotProvidedException, ModuleStartException { - MeshDataBufferFileCache cache = new MeshDataBufferFileCache(config); + MeshDataBufferFileCache cache = new MeshDataBufferFileCache(config, getManager()); try { cache.start(); TelemetryDataDispatcher.setCache(cache, getManager()); @@ -56,7 +57,7 @@ public class MeshReceiverProvider extends ModuleProvider { } CoreRegisterLinker.setModuleManager(getManager()); GRPCHandlerRegister service = getManager().find(CoreModule.NAME).provider().getService(GRPCHandlerRegister.class); - service.addHandler(new MeshGRPCHandler()); + service.addHandler(new MeshGRPCHandler(getManager())); } @Override public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException { @@ -64,6 +65,6 @@ public class MeshReceiverProvider extends ModuleProvider { } @Override public String[] requiredModules() { - return new String[] {CoreModule.NAME}; + return new String[] {TelemetryModule.NAME, CoreModule.NAME}; } } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java index c30daac5bf..09fb310501 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java @@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener. import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.segment.SegmentSpanListener; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.service.ServiceMappingSpanListener; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardizationWorker; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; /** * @author peng-yongsheng @@ -90,13 +91,17 @@ public class TraceModuleProvider extends ModuleProvider { try { grpcHandlerRegister.addHandler(new TraceSegmentServiceHandler(segmentProducer)); - grpcHandlerRegister.addHandler(new TraceSegmentReportServiceHandler(segmentProducerV2)); + grpcHandlerRegister.addHandler(new TraceSegmentReportServiceHandler(segmentProducerV2, getManager())); jettyHandlerRegister.addHandler(new TraceSegmentServletHandler(segmentProducer)); - SegmentStandardizationWorker standardizationWorker = new SegmentStandardizationWorker(segmentProducer, moduleConfig.getBufferPath() + "v5", moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig.isBufferFileCleanWhenRestart()); + SegmentStandardizationWorker standardizationWorker = new SegmentStandardizationWorker(getManager(), segmentProducer, + moduleConfig.getBufferPath() + "v5", moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig.isBufferFileCleanWhenRestart(), + false); segmentProducer.setStandardizationWorker(standardizationWorker); - SegmentStandardizationWorker standardizationWorker2 = new SegmentStandardizationWorker(segmentProducer, moduleConfig.getBufferPath(), moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig.isBufferFileCleanWhenRestart()); + SegmentStandardizationWorker standardizationWorker2 = new SegmentStandardizationWorker(getManager(), segmentProducer, + moduleConfig.getBufferPath(), moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig.isBufferFileCleanWhenRestart(), + false); segmentProducerV2.setStandardizationWorker(standardizationWorker2); } catch (IOException e) { throw new ModuleStartException(e.getMessage(), e); @@ -108,6 +113,6 @@ public class TraceModuleProvider extends ModuleProvider { } @Override public String[] requiredModules() { - return new String[] {CoreModule.NAME}; + return new String[] {TelemetryModule.NAME, CoreModule.NAME}; } } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/grpc/TraceSegmentReportServiceHandler.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/grpc/TraceSegmentReportServiceHandler.java index 32d8473139..22bc0ec56d 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/grpc/TraceSegmentReportServiceHandler.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/grpc/TraceSegmentReportServiceHandler.java @@ -22,10 +22,13 @@ import io.grpc.stub.StreamObserver; import org.apache.skywalking.apm.network.common.Commands; import org.apache.skywalking.apm.network.language.agent.UpstreamSegment; import org.apache.skywalking.apm.network.language.agent.v2.TraceSegmentReportServiceGrpc; +import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler; import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v5.grpc.TraceSegmentServiceHandler; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParseV2; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentSource; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,9 +37,16 @@ public class TraceSegmentReportServiceHandler extends TraceSegmentReportServiceG private static final Logger logger = LoggerFactory.getLogger(TraceSegmentServiceHandler.class); private final SegmentParseV2.Producer segmentProducer; + private CounterMetric counter; + private HistogramMetric histogram; - public TraceSegmentReportServiceHandler(SegmentParseV2.Producer segmentProducer) { + public TraceSegmentReportServiceHandler(SegmentParseV2.Producer segmentProducer, ModuleManager moduleManager) { this.segmentProducer = segmentProducer; + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + counter = metricCreator.createCounter("trace_grpc_v6_in_count", "The count of service mesh telemetry", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + histogram = metricCreator.createHistogramMetric("trace_grpc_v6_in_latency", "The process latency of service mesh telemetry", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); } @Override public StreamObserver collect(StreamObserver responseObserver) { @@ -46,7 +56,13 @@ public class TraceSegmentReportServiceHandler extends TraceSegmentReportServiceG logger.debug("receive segment"); } - segmentProducer.send(segment, SegmentSource.Agent); + counter.inc(); + HistogramMetric.Timer timer = histogram.createTimer(); + try { + segmentProducer.send(segment, SegmentSource.Agent); + } finally { + timer.finish(); + } } @Override public void onError(Throwable throwable) { diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java index d4bd8617c3..24839017d2 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParse.java @@ -19,32 +19,18 @@ package org.apache.skywalking.oap.server.receiver.trace.provider.parser; import com.google.protobuf.InvalidProtocolBufferException; -import java.util.LinkedList; -import java.util.List; +import java.util.*; import lombok.Setter; -import org.apache.skywalking.apm.network.language.agent.SpanType; -import org.apache.skywalking.apm.network.language.agent.TraceSegmentObject; -import org.apache.skywalking.apm.network.language.agent.UniqueId; -import org.apache.skywalking.apm.network.language.agent.UpstreamSegment; +import org.apache.skywalking.apm.network.language.agent.*; import org.apache.skywalking.oap.server.library.buffer.DataStreamReader; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.TimeBucketUtils; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.ReferenceDecorator; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.SegmentCoreInfo; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.SegmentDecorator; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.SpanDecorator; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.EntrySpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.ExitSpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.FirstSpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.GlobalTraceIdsListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.LocalSpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.SpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.ReferenceIdExchanger; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardization; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardizationWorker; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SpanIdExchanger; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.*; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.*; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; +import org.slf4j.*; /** * @author peng-yongsheng @@ -58,6 +44,9 @@ public class SegmentParse { private final SegmentParserListenerManager listenerManager; private final SegmentCoreInfo segmentCoreInfo; @Setter private SegmentStandardizationWorker standardizationWorker; + private CounterMetric traceBufferFileRetry; + private CounterMetric traceBufferFileOut; + private CounterMetric traceParseError; private SegmentParse(ModuleManager moduleManager, SegmentParserListenerManager listenerManager) { this.moduleManager = moduleManager; @@ -67,6 +56,14 @@ public class SegmentParse { this.segmentCoreInfo.setStartTime(Long.MAX_VALUE); this.segmentCoreInfo.setEndTime(Long.MIN_VALUE); this.segmentCoreInfo.setV2(false); + + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + traceBufferFileRetry = metricCreator.createCounter("v5_trace_buffer_file_retry", "The number of retry trace segment from the buffer file, but haven't registered successfully.", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + traceBufferFileOut = metricCreator.createCounter("v5_trace_buffer_file_out", "The number of trace segment out of the buffer file", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + traceParseError = metricCreator.createCounter("v5_trace_parse_error", "The number of trace segment out of the buffer file", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); } public boolean parse(UpstreamSegment segment, Source source) { @@ -85,16 +82,21 @@ public class SegmentParse { if (source.equals(Source.Agent)) { writeToBufferFile(segmentCoreInfo.getSegmentId(), segment); + } else { + // from SegmentSource.Buffer + traceBufferFileRetry.inc(); } return false; } else { if (logger.isDebugEnabled()) { logger.debug("This segment id exchange success, id: {}", segmentCoreInfo.getSegmentId()); } + traceBufferFileOut.inc(); notifyListenerToBuild(); return true; } } catch (Throwable e) { + traceParseError.inc(); logger.error(e.getMessage(), e); return true; } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParseV2.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParseV2.java index 2058f61e80..13c478392c 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParseV2.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/SegmentParseV2.java @@ -19,32 +19,19 @@ package org.apache.skywalking.oap.server.receiver.trace.provider.parser; import com.google.protobuf.InvalidProtocolBufferException; -import java.util.LinkedList; -import java.util.List; +import java.util.*; import lombok.Setter; -import org.apache.skywalking.apm.network.language.agent.SpanType; -import org.apache.skywalking.apm.network.language.agent.UniqueId; -import org.apache.skywalking.apm.network.language.agent.UpstreamSegment; +import org.apache.skywalking.apm.network.language.agent.*; import org.apache.skywalking.apm.network.language.agent.v2.SegmentObject; import org.apache.skywalking.oap.server.library.buffer.DataStreamReader; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.TimeBucketUtils; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.ReferenceDecorator; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.SegmentCoreInfo; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.SegmentDecorator; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.SpanDecorator; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.EntrySpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.ExitSpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.FirstSpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.GlobalTraceIdsListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.LocalSpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.SpanListener; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.ReferenceIdExchanger; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardization; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardizationWorker; -import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SpanIdExchanger; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.*; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.*; +import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; +import org.slf4j.*; /** * SegmentParseV2 is a replication of SegmentParse, but be compatible with v2 trace protocol. @@ -60,6 +47,9 @@ public class SegmentParseV2 { private final SegmentParserListenerManager listenerManager; private final SegmentCoreInfo segmentCoreInfo; @Setter private SegmentStandardizationWorker standardizationWorker; + private CounterMetric traceBufferFileRetry; + private CounterMetric traceBufferFileOut; + private CounterMetric traceParseError; private SegmentParseV2(ModuleManager moduleManager, SegmentParserListenerManager listenerManager) { this.moduleManager = moduleManager; @@ -69,6 +59,14 @@ public class SegmentParseV2 { this.segmentCoreInfo.setStartTime(Long.MAX_VALUE); this.segmentCoreInfo.setEndTime(Long.MIN_VALUE); this.segmentCoreInfo.setV2(true); + + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + traceBufferFileRetry = metricCreator.createCounter("v6_trace_buffer_file_retry", "The number of retry trace segment from the buffer file, but haven't registered successfully.", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + traceBufferFileOut = metricCreator.createCounter("v6_trace_buffer_file_out", "The number of trace segment out of the buffer file", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); + traceParseError = metricCreator.createCounter("v6_trace_parse_error", "The number of trace segment out of the buffer file", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); } public boolean parse(UpstreamSegment segment, SegmentSource source) { @@ -87,16 +85,21 @@ public class SegmentParseV2 { if (source.equals(SegmentSource.Agent)) { writeToBufferFile(segmentCoreInfo.getSegmentId(), segment); + } else { + // from SegmentSource.Buffer + traceBufferFileRetry.inc(); } return false; } else { if (logger.isDebugEnabled()) { logger.debug("This segment id exchange success, id: {}", segmentCoreInfo.getSegmentId()); } + traceBufferFileOut.inc(); notifyListenerToBuild(); return true; } } catch (Throwable e) { + traceParseError.inc(); logger.error(e.getMessage(), e); return true; } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SegmentStandardizationWorker.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SegmentStandardizationWorker.java index 3adc4c8a67..47b4396657 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SegmentStandardizationWorker.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SegmentStandardizationWorker.java @@ -25,7 +25,10 @@ import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer; import org.apache.skywalking.apm.network.language.agent.UpstreamSegment; import org.apache.skywalking.oap.server.core.worker.AbstractWorker; import org.apache.skywalking.oap.server.library.buffer.BufferStream; +import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParse; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.*; /** @@ -36,9 +39,11 @@ public class SegmentStandardizationWorker extends AbstractWorker dataCarrier; + private CounterMetric traceBufferFileIn; - public SegmentStandardizationWorker(SegmentParse.Producer segmentParseCreator, String path, - int offsetFileMaxSize, int dataFileMaxSize, boolean cleanWhenRestart) throws IOException { + public SegmentStandardizationWorker(ModuleManager moduleManager, SegmentParse.Producer segmentParseCreator, + String path, + int offsetFileMaxSize, int dataFileMaxSize, boolean cleanWhenRestart, boolean isV6) throws IOException { super(Integer.MAX_VALUE); BufferStream.Builder builder = new BufferStream.Builder<>(path); @@ -53,6 +58,11 @@ public class SegmentStandardizationWorker extends AbstractWorker(1, 1024); dataCarrier.consume(new Consumer(stream), 1); + + MetricCreator metricCreator = moduleManager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class); + String metricNamePrefix = isV6 ? "v6_" : "v5_"; + traceBufferFileIn = metricCreator.createCounter(metricNamePrefix + "trace_buffer_file_in", "The number of trace segment into the buffer file", + MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE); } @Override @@ -75,6 +85,7 @@ public class SegmentStandardizationWorker extends AbstractWorker data) { for (SegmentStandardization aData : data) { + traceBufferFileIn.inc(); stream.write(aData.getUpstreamSegment()); } } diff --git a/oap-server/server-starter/pom.xml b/oap-server/server-starter/pom.xml index d02bf1517b..88f5272444 100644 --- a/oap-server/server-starter/pom.xml +++ b/oap-server/server-starter/pom.xml @@ -133,6 +133,13 @@ server-alarm-plugin ${project.version} + + + + org.apache.skywalking + telemetry-prometheus + ${project.version} + skywalking-oap 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 0bd2a20ad4..5e008be01b 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 @@ -20,7 +20,10 @@ package org.apache.skywalking.oap.server.starter; import org.apache.skywalking.oap.server.core.RunningMode; import org.apache.skywalking.oap.server.library.module.*; +import org.apache.skywalking.oap.server.library.util.TimeBucketUtils; import org.apache.skywalking.oap.server.starter.config.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.*; import org.slf4j.*; /** @@ -40,6 +43,10 @@ public class OAPServerStartUp { ApplicationConfiguration applicationConfiguration = configLoader.load(); manager.init(applicationConfiguration); + manager.find(TelemetryModule.NAME).provider().getService(MetricCreator.class).createGauge("uptime", + "oap server start up time", MetricTag.EMPTY_KEY, MetricTag.EMPTY_VALUE) + .setValue(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(System.currentTimeMillis())); + if (RunningMode.isInitMode()) { logger.info("OAP starts up in init mode successfully, exit now..."); System.exit(0); diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml index d352526d73..5b53d73a06 100644 --- a/oap-server/server-starter/src/main/resources/application.yml +++ b/oap-server/server-starter/src/main/resources/application.yml @@ -96,4 +96,4 @@ query: alarm: default: telemetry: - none: + prometheus: diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java index e01606e2b6..afec137d1a 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch; +import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.storage.*; import org.apache.skywalking.oap.server.core.storage.cache.*; import org.apache.skywalking.oap.server.core.storage.query.*; @@ -102,6 +103,6 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider { @Override public String[] requiredModules() { - return new String[0]; + return new String[] {CoreModule.NAME}; } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java index 3fb67fd59d..0dc7f25ef7 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java @@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; import java.io.IOException; import java.util.Properties; +import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.storage.IBatchDAO; import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO; import org.apache.skywalking.oap.server.core.storage.IRegisterLockDAO; @@ -138,6 +139,6 @@ public class MySQLStorageProvider extends ModuleProvider { } @Override public String[] requiredModules() { - return new String[0]; + return new String[] {CoreModule.NAME}; } } diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/CounterMetric.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/CounterMetric.java index 944872a1f6..3cd47fd673 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/CounterMetric.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/CounterMetric.java @@ -25,7 +25,7 @@ package org.apache.skywalking.oap.server.telemetry.api; * * @author wusheng */ -public interface CounterMetric extends TaggableMetric { +public interface CounterMetric { /** * Increase 1 to counter */ diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/GaugeMetric.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/GaugeMetric.java index 4fd9bfb92d..8a3cf45850 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/GaugeMetric.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/GaugeMetric.java @@ -23,28 +23,35 @@ package org.apache.skywalking.oap.server.telemetry.api; * * @author wusheng */ -public interface GaugeMetric extends TaggableMetric { +public interface GaugeMetric { /** - * Increase 1 to counter + * Increase 1 to gauge */ void inc(); /** - * Increase the given value to the counter + * Increase the given value to the gauge * * @param value */ void inc(double value); /** - * Decrease 1 to counter + * Decrease 1 to gauge */ void dec(); /** - * Decrease the given value to the counter + * Decrease the given value to the gauge * * @param value */ void dec(double value); + + /** + * Set the given value to the gauge + * + * @param value + */ + void setValue(double value); } diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HistogramMetric.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HistogramMetric.java index 240a318844..f5670e485c 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HistogramMetric.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HistogramMetric.java @@ -26,7 +26,7 @@ import java.io.*; * * @author wusheng */ -public abstract class HistogramMetric implements TaggableMetric { +public abstract class HistogramMetric { public Timer createTimer() { return new Timer(this); } @@ -38,7 +38,7 @@ public abstract class HistogramMetric implements TaggableMetric { */ public abstract void observe(double value); - class Timer implements Closeable { + public class Timer implements Closeable { private final HistogramMetric metric; private final long startNanos; private double duration; diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricCreator.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricCreator.java index 591d67bb2e..8dcb8cd223 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricCreator.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricCreator.java @@ -32,29 +32,29 @@ public interface MetricCreator extends Service { * * @param name * @param tips - * @param labels + * @param tagKeys * @return */ - CounterMetric createCounter(String name, String tips, MetricTag.Keys labels); + CounterMetric createCounter(String name, String tips, MetricTag.Keys tagKeys, MetricTag.Values tagValues); /** * Create a gauge type metric instance. * * @param name * @param tips - * @param labels + * @param tagKeys * @return */ - GaugeMetric createGauge(String name, String tips, MetricTag.Keys labels); + GaugeMetric createGauge(String name, String tips, MetricTag.Keys tagKeys, MetricTag.Values tagValues); /** * Create a Histogram type metric instance. * * @param name * @param tips - * @param labels + * @param tagKeys * @param buckets Time bucket for duration. * @return */ - HistogramMetric createHistogramMetric(String name, String tips, MetricTag.Keys labels, double... buckets); + HistogramMetric createHistogramMetric(String name, String tips, MetricTag.Keys tagKeys, MetricTag.Values tagValues, double... buckets); } diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricTag.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricTag.java index cd5b6b4b4b..1e8beb6d32 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricTag.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricTag.java @@ -24,7 +24,10 @@ package org.apache.skywalking.oap.server.telemetry.api; * The tag values should be set in putting value phase. */ public class MetricTag { - public class Keys { + public static final Keys EMPTY_KEY = new Keys(); + public static final Values EMPTY_VALUE = new Values(); + + public static class Keys { private String[] keys; public Keys() { @@ -40,7 +43,7 @@ public class MetricTag { } } - public class Values { + public static class Values { private String[] values; public Values(Keys keys) { diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/TelemetryRelatedContext.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/TelemetryRelatedContext.java index a37e647503..0ed338a8ed 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/TelemetryRelatedContext.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/TelemetryRelatedContext.java @@ -23,10 +23,10 @@ package org.apache.skywalking.oap.server.telemetry.api; * @author wusheng */ public enum TelemetryRelatedContext { - INTANCE; + INSTANCE; - private volatile String id = "default"; - private TelemetryRelatedContext(){} + private volatile String id = null; + TelemetryRelatedContext(){} /** * Set a global ID to represent the current oap instance @@ -36,9 +36,9 @@ public enum TelemetryRelatedContext { } /** - * Get the oap instance ID, if be set before, otherwise, return `default` string. + * Get the oap instance ID, if be set before. * - * @return id + * @return id or null. */ public String getId() { return id; diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricCreatorNoop.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricCreatorNoop.java index c33dda797f..10d3ba2a69 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricCreatorNoop.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/none/MetricCreatorNoop.java @@ -26,16 +26,51 @@ import org.apache.skywalking.oap.server.telemetry.api.*; * @author wusheng */ public class MetricCreatorNoop implements MetricCreator { - @Override public CounterMetric createCounter(String name, String tips, MetricTag.Keys labels) { - return null; - } + @Override + public CounterMetric createCounter(String name, String tips, MetricTag.Keys tagKeys, MetricTag.Values tagValues) { + return new CounterMetric() { + @Override public void inc() { - @Override public GaugeMetric createGauge(String name, String tips, MetricTag.Keys labels) { - return null; + } + + @Override public void inc(double value) { + + } + }; } @Override - public HistogramMetric createHistogramMetric(String name, String tips, MetricTag.Keys labels, double... buckets) { - return null; + public GaugeMetric createGauge(String name, String tips, MetricTag.Keys tagKeys, MetricTag.Values tagValues) { + return new GaugeMetric() { + @Override public void inc() { + + } + + @Override public void inc(double value) { + + } + + @Override public void dec() { + + } + + @Override public void dec(double value) { + + } + + @Override public void setValue(double value) { + + } + }; + } + + @Override + public HistogramMetric createHistogramMetric(String name, String tips, MetricTag.Keys tagKeys, + MetricTag.Values tagValues, double... buckets) { + return new HistogramMetric() { + @Override public void observe(double value) { + + } + }; } } diff --git a/oap-server/server-telemetry/telemetry-prometheus/pom.xml b/oap-server/server-telemetry/telemetry-prometheus/pom.xml index 7885f9bf96..4514c16b69 100644 --- a/oap-server/server-telemetry/telemetry-prometheus/pom.xml +++ b/oap-server/server-telemetry/telemetry-prometheus/pom.xml @@ -31,6 +31,11 @@ jar + + org.apache.skywalking + telemetry-api + ${project.version} + io.prometheus diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/BaseMetric.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/BaseMetric.java new file mode 100644 index 0000000000..536154da24 --- /dev/null +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/BaseMetric.java @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.telemetry.prometheus; + +import io.prometheus.client.SimpleCollector; +import java.util.*; +import java.util.concurrent.locks.ReentrantLock; +import org.apache.skywalking.oap.server.telemetry.api.*; + +/** + * BaseMetric parent class represents the me + * + * @author wusheng + */ +public abstract class BaseMetric { + private static Map ALL_METRICS = new HashMap<>(); + + private volatile C metricInstance; + protected final String name; + protected final String tips; + protected final MetricTag.Keys labels; + protected final MetricTag.Values values; + private ReentrantLock lock = new ReentrantLock(); + + public BaseMetric(String name, String tips, MetricTag.Keys labels, + MetricTag.Values values) { + this.name = name; + this.tips = tips; + this.labels = labels; + this.values = values; + } + + protected boolean isIDReady() { + return TelemetryRelatedContext.INSTANCE.getId() != null; + } + + protected C getMetric() { + if (metricInstance == null) { + if (isIDReady()) { + lock.lock(); + try { + if (metricInstance == null) { + String[] labelNames = new String[labels.getKeys().length + 1]; + labelNames[0] = "instance"; + for (int i = 0; i < labels.getKeys().length; i++) { + labelNames[i + 1] = labels.getKeys()[i]; + } + + String[] labelValues = new String[values.getValues().length + 1]; + labelValues[0] = TelemetryRelatedContext.INSTANCE.getId(); + for (int i = 0; i < values.getValues().length; i++) { + labelValues[i + 1] = values.getValues()[i]; + } + + if (!ALL_METRICS.containsKey(name)) { + synchronized (ALL_METRICS) { + if (!ALL_METRICS.containsKey(name)) { + ALL_METRICS.put(name, create(labelNames)); + } + } + } + + T metric = (T)ALL_METRICS.get(name); + + metricInstance = (C)metric.labels(labelValues); + } + } finally { + lock.unlock(); + } + } + } + + return metricInstance; + } + + protected abstract T create(String[] labelNames); +} diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/TaggableMetric.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusConfig.java similarity index 71% rename from oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/TaggableMetric.java rename to oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusConfig.java index ade3ff519b..670fb6e98b 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/TaggableMetric.java +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusConfig.java @@ -16,13 +16,19 @@ * */ -package org.apache.skywalking.oap.server.telemetry.api; +package org.apache.skywalking.oap.server.telemetry.prometheus; + +import lombok.*; +import org.apache.skywalking.oap.server.library.module.ModuleConfig; /** - * Set tag values to this metric instance. + * The Prometheus telemetry implementor settings. * * @author wusheng */ -public interface TaggableMetric { - void setTagValues(MetricTag.Values values); +@Setter +@Getter +public class PrometheusConfig extends ModuleConfig { + private String host = "0.0.0.0"; + private int port = 1234; } diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusCounterMetric.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusCounterMetric.java new file mode 100644 index 0000000000..cae0edb0c6 --- /dev/null +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusCounterMetric.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.telemetry.prometheus; + +import io.prometheus.client.Counter; +import org.apache.skywalking.oap.server.telemetry.api.*; + +/** + * Counter metric in Prometheus implementor. + * + * @author wusheng + */ +public class PrometheusCounterMetric extends BaseMetric implements CounterMetric { + + public PrometheusCounterMetric(String name, String tips, + MetricTag.Keys labels, MetricTag.Values values) { + super(name, tips, labels, values); + } + + @Override public void inc() { + Counter.Child metric = this.getMetric(); + if (metric != null) { + metric.inc(); + } + } + + @Override public void inc(double value) { + Counter.Child metric = this.getMetric(); + if (metric != null) { + metric.inc(value); + } + } + + @Override protected Counter create(String[] labelNames) { + return Counter.build() + .name(name).help(tips).labelNames(labelNames).register(); + } +} diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusGaugeMetric.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusGaugeMetric.java new file mode 100644 index 0000000000..c651725728 --- /dev/null +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusGaugeMetric.java @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.telemetry.prometheus; + +import io.prometheus.client.Gauge; +import org.apache.skywalking.oap.server.telemetry.api.*; + +/** + * Gauge metric in Prometheus implementor. + * + * @author wusheng + */ +public class PrometheusGaugeMetric extends BaseMetric implements GaugeMetric { + public PrometheusGaugeMetric(String name, String tips, + MetricTag.Keys labels, + MetricTag.Values values) { + super(name, tips, labels, values); + } + + @Override public void inc() { + Gauge.Child metric = this.getMetric(); + if (metric != null) { + metric.inc(); + } + } + + @Override public void inc(double value) { + Gauge.Child metric = this.getMetric(); + if (metric != null) { + metric.inc(value); + } + } + + @Override public void dec() { + Gauge.Child metric = this.getMetric(); + if (metric != null) { + metric.dec(); + } + } + + @Override public void dec(double value) { + Gauge.Child metric = this.getMetric(); + if (metric != null) { + metric.dec(value); + } + } + + @Override public void setValue(double value) { + Gauge.Child metric = this.getMetric(); + if (metric != null) { + metric.set(value); + } + } + + @Override protected Gauge create(String[] labelNames) { + return Gauge.build() + .name(name).help(tips).labelNames(labelNames).register(); + } +} diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusHistogramMetric.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusHistogramMetric.java new file mode 100644 index 0000000000..a2ad2ecb3d --- /dev/null +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusHistogramMetric.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.telemetry.prometheus; + +import io.prometheus.client.Histogram; +import org.apache.skywalking.oap.server.telemetry.api.*; + +/** + * HistogramMetric metric in Prometheus implementor. + * + * @author wusheng + */ +public class PrometheusHistogramMetric extends HistogramMetric { + private InnerMetricObject inner; + private final double[] buckets; + + public PrometheusHistogramMetric(String name, String tips, MetricTag.Keys labels, + MetricTag.Values values, double... buckets) { + inner = new InnerMetricObject(name, tips, labels, values); + this.buckets = buckets; + } + + @Override public void observe(double value) { + Histogram.Child metric = inner.getMetric(); + if (metric != null) { + metric.observe(value); + } + } + + class InnerMetricObject extends BaseMetric { + public InnerMetricObject(String name, String tips, MetricTag.Keys labels, + MetricTag.Values values) { + super(name, tips, labels, values); + } + + @Override protected Histogram create(String[] labelNames) { + Histogram.Builder builder = Histogram.build() + .name(name).help(tips); + if (builder != null && buckets.length > 0) { + builder = builder.buckets(buckets); + } + return builder.labelNames(labelNames).register(); + } + } +} diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusMetricCreator.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusMetricCreator.java new file mode 100644 index 0000000000..966a9ce259 --- /dev/null +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusMetricCreator.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.telemetry.prometheus; + +import org.apache.skywalking.oap.server.telemetry.api.*; + +/** + * Create metric instance for Prometheus exporter. + * + * @author wusheng + */ +public class PrometheusMetricCreator implements MetricCreator { + @Override + public CounterMetric createCounter(String name, String tips, MetricTag.Keys tagKeys, MetricTag.Values tagValues) { + return new PrometheusCounterMetric(name, tips, tagKeys, tagValues); + } + + @Override + public GaugeMetric createGauge(String name, String tips, MetricTag.Keys tagKeys, MetricTag.Values tagValues) { + return new PrometheusGaugeMetric(name, tips, tagKeys, tagValues); + } + + @Override + public HistogramMetric createHistogramMetric(String name, String tips, MetricTag.Keys tagKeys, + MetricTag.Values tagValues, double... buckets) { + return new PrometheusHistogramMetric(name, tips, tagKeys, tagValues, buckets); + } +} diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusTelemetryProvider.java b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusTelemetryProvider.java new file mode 100644 index 0000000000..da9b643fbf --- /dev/null +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/java/org/apache/skywalking/oap/server/telemetry/prometheus/PrometheusTelemetryProvider.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.telemetry.prometheus; + +import io.prometheus.client.exporter.HTTPServer; +import io.prometheus.client.hotspot.*; +import java.io.IOException; +import org.apache.skywalking.oap.server.library.module.*; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.MetricCreator; + +/** + * Start the Prometheus + * + * @author wusheng + */ +public class PrometheusTelemetryProvider extends ModuleProvider { + private PrometheusConfig config; + + public PrometheusTelemetryProvider() { + config = new PrometheusConfig(); + } + + @Override public String name() { + return "prometheus"; + } + + @Override public Class module() { + return TelemetryModule.class; + } + + @Override public ModuleConfig createConfigBeanIfAbsent() { + return config; + } + + @Override public void prepare() throws ServiceNotProvidedException, ModuleStartException { + this.registerServiceImplementation(MetricCreator.class, new PrometheusMetricCreator()); + try { + new HTTPServer(config.getHost(), config.getPort()); + } catch (IOException e) { + throw new ModuleStartException(e.getMessage(), e); + } + + DefaultExports.initialize(); + } + + @Override public void start() throws ServiceNotProvidedException, ModuleStartException { + + } + + @Override public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException { + } + + @Override public String[] requiredModules() { + return new String[0]; + } +} diff --git a/oap-server/server-telemetry/telemetry-prometheus/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider b/oap-server/server-telemetry/telemetry-prometheus/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider new file mode 100644 index 0000000000..1e72991caf --- /dev/null +++ b/oap-server/server-telemetry/telemetry-prometheus/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider @@ -0,0 +1,20 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# + + +org.apache.skywalking.oap.server.telemetry.prometheus.PrometheusTelemetryProvider