(skywalking) 01/01: Add self observability metrics for otel handler

[ke...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kezhenxu94 pushed a commit to branch otelso117
in repository https://gitbox.apache.org/repos/asf/skywalking.git

commit a3e832cf0644f96f5675fcdf2ac01aa5be43a270
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Thu Sep 5 15:56:00 2024 +0800

    Add self observability metrics for otel handler
---
 .github/workflows/publish-docker.yaml              |  1 +
 .../otlp/OpenTelemetryMetricRequestProcessor.java  | 27 ++++++++++++++++++++++
 .../src/main/resources/otel-rules/oap.yaml         | 27 ++++++++++++++--------
 3 files changed, 45 insertions(+), 10 deletions(-)

diff --git a/.github/workflows/publish-docker.yaml b/.github/workflows/publish-docker.yaml
index 01c7e0b20d..ca6278dc33 100644
--- a/.github/workflows/publish-docker.yaml
+++ b/.github/workflows/publish-docker.yaml
@@ -20,6 +20,7 @@ on:
   push:
     branches:
       - master
+      - otelso117
   release:
     types:
       - released
diff --git a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricRequestProcessor.java b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricRequestProcessor.java
index 0be3aaaf2c..c85087338f 100644
--- a/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricRequestProcessor.java
+++ b/oap-server/server-receiver-plugin/otel-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricRequestProcessor.java
@@ -42,6 +42,10 @@ import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Histogra
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Metric;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Summary;
 import org.apache.skywalking.oap.server.receiver.otel.OtelMetricReceiverConfig;
+import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -49,6 +53,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 import java.util.stream.Stream;
+import lombok.Getter;
 
 import static io.opentelemetry.proto.metrics.v1.AggregationTemporality.AGGREGATION_TEMPORALITY_DELTA;
 import static io.opentelemetry.proto.metrics.v1.AggregationTemporality.AGGREGATION_TEMPORALITY_UNSPECIFIED;
@@ -73,7 +78,27 @@ public class OpenTelemetryMetricRequestProcessor implements Service {
             .build();
     private List<PrometheusMetricConverter> converters;
 
+    @Getter(lazy = true)
+    private final MetricsCreator metricsCreator = manager.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class);
+
+    @Getter(lazy = true)
+    private final CounterMetrics receivedCounter = getMetricsCreator().createCounter(
+        "otel_metrics_received",
+        "The number of metrics received by the OTLP receiver",
+        new MetricsTag.Keys("protocol"),
+        new MetricsTag.Values("otlp")
+    );
+    @Getter(lazy = true)
+    private final CounterMetrics processedCounter = getMetricsCreator().createCounter(
+        "otel_metrics_processed",
+        "The number of metrics processed by the OTLP receiver",
+        new MetricsTag.Keys("protocol"),
+        new MetricsTag.Values("otlp")
+    );
+
     public void processMetricsRequest(final ExportMetricsServiceRequest requests) {
+        getReceivedCounter().inc(requests.getResourceMetricsList().size());
+
         requests.getResourceMetricsList().forEach(request -> {
             if (log.isDebugEnabled()) {
                 log.debug("Resource attributes: {}", request.getResource().getAttributesList());
@@ -104,6 +129,8 @@ public class OpenTelemetryMetricRequestProcessor implements Service {
                                 tryIt,
                                 "Convert OTEL metric to prometheus metric"
                             )))));
+
+            getProcessedCounter().inc();
         });
 
     }
diff --git a/oap-server/server-starter/src/main/resources/otel-rules/oap.yaml b/oap-server/server-starter/src/main/resources/otel-rules/oap.yaml
index 6de0e00b53..95143353bc 100644
--- a/oap-server/server-starter/src/main/resources/otel-rules/oap.yaml
+++ b/oap-server/server-starter/src/main/resources/otel-rules/oap.yaml
@@ -37,15 +37,17 @@ metricsRules:
   - name: instance_jvm_memory_bytes_used
     exp: jvm_memory_bytes_used.sum(['service', 'host_name'])
   - name: instance_jvm_gc_count
-    exp: "jvm_gc_collection_seconds_count.tagMatch('gc', 'PS Scavenge|Copy|ParNew|G1 Young Generation|PS MarkSweep|MarkSweepCompact|ConcurrentMarkSweep|G1 Old Generation')
-    .sum(['service', 'host_name', 'gc']).increase('PT1M')
-    .tag({tags -> if (tags['gc'] == 'PS Scavenge' || tags['gc'] == 'Copy' || tags['gc'] == 'ParNew' || tags['gc'] == 'G1 Young Generation') {tags.gc = 'young_gc_count'} })
-    .tag({tags -> if (tags['gc'] == 'PS MarkSweep' || tags['gc'] == 'MarkSweepCompact' || tags['gc'] == 'ConcurrentMarkSweep' || tags['gc'] == 'G1 Old Generation') {tags.gc = 'old_gc_count'} })"
+    exp: >
+      jvm_gc_collection_seconds_count.tagMatch('gc', 'PS Scavenge|Copy|ParNew|G1 Young Generation|PS MarkSweep|MarkSweepCompact|ConcurrentMarkSweep|G1 Old Generation')
+      .sum(['service', 'host_name', 'gc']).increase('PT1M')
+      .tag({tags -> if (tags['gc'] == 'PS Scavenge' || tags['gc'] == 'Copy' || tags['gc'] == 'ParNew' || tags['gc'] == 'G1 Young Generation') {tags.gc = 'young_gc_count'} })
+      .tag({tags -> if (tags['gc'] == 'PS MarkSweep' || tags['gc'] == 'MarkSweepCompact' || tags['gc'] == 'ConcurrentMarkSweep' || tags['gc'] == 'G1 Old Generation') {tags.gc = 'old_gc_count'} })"
   - name: instance_jvm_gc_time
-    exp: "(jvm_gc_collection_seconds_sum * 1000).tagMatch('gc', 'PS Scavenge|Copy|ParNew|G1 Young Generation|PS MarkSweep|MarkSweepCompact|ConcurrentMarkSweep|G1 Old Generation')
-    .sum(['service', 'host_name', 'gc']).increase('PT1M')
-    .tag({tags -> if (tags['gc'] == 'PS Scavenge' || tags['gc'] == 'Copy' || tags['gc'] == 'ParNew' || tags['gc'] == 'G1 Young Generation') {tags.gc = 'young_gc_time'} })
-    .tag({tags -> if (tags['gc'] == 'PS MarkSweep' || tags['gc'] == 'MarkSweepCompact' || tags['gc'] == 'ConcurrentMarkSweep' || tags['gc'] == 'G1 Old Generation') {tags.gc = 'old_gc_time'} })"
+    exp: >
+      (jvm_gc_collection_seconds_sum * 1000).tagMatch('gc', 'PS Scavenge|Copy|ParNew|G1 Young Generation|PS MarkSweep|MarkSweepCompact|ConcurrentMarkSweep|G1 Old Generation')
+      .sum(['service', 'host_name', 'gc']).increase('PT1M')
+      .tag({tags -> if (tags['gc'] == 'PS Scavenge' || tags['gc'] == 'Copy' || tags['gc'] == 'ParNew' || tags['gc'] == 'G1 Young Generation') {tags.gc = 'young_gc_time'} })
+      .tag({tags -> if (tags['gc'] == 'PS MarkSweep' || tags['gc'] == 'MarkSweepCompact' || tags['gc'] == 'ConcurrentMarkSweep' || tags['gc'] == 'G1 Old Generation') {tags.gc = 'old_gc_time'} })"
   - name: instance_trace_count
     exp: trace_in_latency_count.sum(['service', 'host_name']).increase('PT1M')
   - name: instance_trace_latency_percentile
@@ -59,8 +61,9 @@ metricsRules:
   - name: instance_mesh_analysis_error_count
     exp: mesh_analysis_error_count.sum(['service', 'host_name']).increase('PT1M')
   - name: instance_metrics_aggregation
-    exp: "metrics_aggregation.tagEqual('dimensionality', 'minute').sum(['service', 'host_name', 'level']).increase('PT1M')
-    .tag({tags -> if (tags['level'] == '1') {tags.level = 'L1 aggregation'} }).tag({tags -> if (tags['level'] == '2') {tags.level = 'L2 aggregation'} })"
+    exp: >
+      metrics_aggregation.tagEqual('dimensionality', 'minute').sum(['service', 'host_name', 'level']).increase('PT1M')
+      .tag({tags -> if (tags['level'] == '1') {tags.level = 'L1 aggregation'} }).tag({tags -> if (tags['level'] == '2') {tags.level = 'L2 aggregation'} })"
   - name: instance_persistence_execute_percentile
     exp: persistence_timer_bulk_execute_latency.sum(['le', 'service', 'host_name']).increase('PT5M').histogram().histogram_percentile([50,70,90,99])
   - name: instance_persistence_prepare_percentile
@@ -99,3 +102,7 @@ metricsRules:
     exp: k8s_als_drop_count.sum(['service', 'host_name']).increase('PT1M')
   - name: instance_k8s_als_latency_percentile
     exp: k8s_als_in_latency.sum(['le', 'service', 'host_name']).increase('PT1M').histogram().histogram_percentile([50,70,90,99])
+  - name: otel_metrics_received
+    exp: otel_metrics_received.sum(['service', 'host_name']).increase('PT1M')
+  - name: otel_metrics_processed
+    exp: otel_metrics_processed.sum(['service', 'host_name']).increase('PT1M')