Improve the performance of OTEL metrics handler. (#12645)

Benchmark for a single node k8s monitoring. | metrics (avg) | before | after | | ------------------------ | ------ | ----- | | cpu | 19 | 16 | | gc count | 16 | 1 | | gc time | 38.8 | 5.1 | | otel metrics latency P50 | 125 | 8 | | otel metrics latency P90 | 333.3 | 22.5 | | otel metrics latency P99 | 666.6 | 166.6 |
apache · Sep 25, 2024 · 3559e85 · 3559e85
1 parent 9cf79f7
commit 3559e85
Show file tree

Hide file tree

Showing 4 changed files with 35 additions and 48 deletions.
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
@@ -73,6 +73,7 @@
 * BanyanDB: support using native term searching for `keyword` in query `findEndpoint` and `getAlarm`.
 * BanyanDB: support TLS connection and configuration.
 * PromQL service: query API support RFC3399 time format.
+* Improve the performance of OTEL metrics handler.
 
 #### UI
 

diff --git a/...n/java/org/apache/skywalking/oap/meter/analyzer/prometheus/PrometheusMetricConverter.java b/...n/java/org/apache/skywalking/oap/meter/analyzer/prometheus/PrometheusMetricConverter.java
@@ -30,12 +30,9 @@
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.oap.meter.analyzer.MetricConvert;
 import org.apache.skywalking.oap.meter.analyzer.dsl.Sample;
 import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamily;
 import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamilyBuilder;
-import org.apache.skywalking.oap.meter.analyzer.prometheus.rule.Rule;
-import org.apache.skywalking.oap.server.core.analysis.meter.MeterSystem;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Counter;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Gauge;
 import org.apache.skywalking.oap.server.library.util.prometheus.metrics.Histogram;
@@ -55,39 +52,22 @@
  */
 @Slf4j
 public class PrometheusMetricConverter {
-    private final Pattern metricsNameEscapePattern;
+    private static final Pattern METRICS_NAME_ESCAPE_PATTERN = Pattern.compile("[/.]");
 
-    private final LoadingCache<String, String> escapedMetricsNameCache =
+    private static final LoadingCache<String, String> ESCAPED_METRICS_NAME_CACHE =
         CacheBuilder.newBuilder()
                     .maximumSize(1000)
                     .build(new CacheLoader<String, String>() {
                         @Override
                         public String load(final String name) {
-                            return metricsNameEscapePattern.matcher(name).replaceAll("_");
+                            return METRICS_NAME_ESCAPE_PATTERN.matcher(name).replaceAll("_");
                         }
                     });
 
-    private final MetricConvert convert;
-
-    public PrometheusMetricConverter(Rule rule, MeterSystem service) {
-        this.convert = new MetricConvert(rule, service);
-        this.metricsNameEscapePattern = Pattern.compile("[/.]");
-    }
-
-    /**
-     * toMeter transforms prometheus metrics to meter-system metrics.
-     *
-     * @param metricStream prometheus metrics stream.
-     */
-    public void toMeter(Stream<Metric> metricStream) {
-        ImmutableMap<String, SampleFamily> data = convertPromMetricToSampleFamily(metricStream);
-        convert.toMeter(data);
-    }
-
-    public ImmutableMap<String, SampleFamily> convertPromMetricToSampleFamily(Stream<Metric> metricStream) {
+    public static ImmutableMap<String, SampleFamily> convertPromMetricToSampleFamily(Stream<Metric> metricStream) {
         return metricStream
             .peek(metric -> log.debug("Prom metric to be convert to SampleFamily: {}", metric))
-            .flatMap(this::convertMetric)
+            .flatMap(PrometheusMetricConverter::convertMetric)
             .filter(t -> t != NIL && t._2.samples.length > 0)
             .peek(t -> log.debug("SampleFamily: {}", t))
             .collect(toImmutableMap(Tuple2::_1, Tuple2::_2, (a, b) -> {
@@ -99,7 +79,7 @@ public ImmutableMap<String, SampleFamily> convertPromMetricToSampleFamily(Stream
             }));
     }
 
-    private Stream<Tuple2<String, SampleFamily>> convertMetric(Metric metric) {
+    private static Stream<Tuple2<String, SampleFamily>> convertMetric(Metric metric) {
         return Match(metric).of(
             Case($(instanceOf(Histogram.class)), t -> Stream.of(
                 Tuple.of(escapedName(metric.getName() + "_count"), SampleFamilyBuilder.newBuilder(Sample.builder().name(escapedName(metric.getName() + "_count"))
@@ -117,7 +97,7 @@ private Stream<Tuple2<String, SampleFamily>> convertMetric(Metric metric) {
         );
     }
 
-    private Optional<Tuple2<String, SampleFamily>> convertToSample(Metric metric) {
+    private static Optional<Tuple2<String, SampleFamily>> convertToSample(Metric metric) {
         Sample[] ss = Match(metric).of(
             Case($(instanceOf(Counter.class)), t -> Collections.singletonList(Sample.builder()
                 .name(escapedName(t.getName()))
@@ -161,12 +141,12 @@ private Optional<Tuple2<String, SampleFamily>> convertToSample(Metric metric) {
     }
 
     // Returns the escaped name of the given one, with "." and "/" replaced by "_"
-    protected String escapedName(final String name) {
+    protected static String escapedName(final String name) {
         try {
-            return escapedMetricsNameCache.get(name);
+            return ESCAPED_METRICS_NAME_CACHE.get(name);
         } catch (ExecutionException e) {
             log.error("Failed to get escaped metrics name from cache", e);
-            return metricsNameEscapePattern.matcher(name).replaceAll("_");
+            return METRICS_NAME_ESCAPE_PATTERN.matcher(name).replaceAll("_");
         }
     }
 }
diff --git a/...c/main/java/org/apache/skywalking/oap/server/receiver/envoy/MetricServiceGRPCHandler.java b/...c/main/java/org/apache/skywalking/oap/server/receiver/envoy/MetricServiceGRPCHandler.java
@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.receiver.envoy;
 
+import com.google.common.collect.ImmutableMap;
 import io.envoyproxy.envoy.service.metrics.v2.MetricsServiceGrpc;
 import io.envoyproxy.envoy.service.metrics.v3.StreamMetricsMessage;
 import io.envoyproxy.envoy.service.metrics.v3.StreamMetricsResponse;
@@ -31,6 +32,8 @@
 import java.util.stream.Collectors;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.meter.analyzer.MetricConvert;
+import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamily;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.oap.meter.analyzer.prometheus.PrometheusMetricConverter;
 import org.apache.skywalking.oap.server.core.CoreModule;
@@ -50,7 +53,7 @@
 public class MetricServiceGRPCHandler extends MetricsServiceGrpc.MetricsServiceImplBase {
     private final CounterMetrics counter;
     private final HistogramMetrics histogram;
-    private final List<PrometheusMetricConverter> converters;
+    private final List<MetricConvert> converters;
 
     private final EnvoyMetricReceiverConfig config;
 
@@ -74,7 +77,7 @@ public MetricServiceGRPCHandler(final ModuleManager moduleManager,
 
         converters = config.rules()
                            .stream()
-                           .map(rule -> new PrometheusMetricConverter(rule, meterSystem))
+                           .map(rule -> new MetricConvert(rule, meterSystem))
                            .collect(Collectors.toList());
     }
 
@@ -123,8 +126,12 @@ public void onNext(StreamMetricsMessage message) {
                         }
                     }
                     groupingMetrics.forEach(
-                        (name, metrics) ->
-                            converters.forEach(converter -> converter.toMeter(metrics.stream())));
+                        (name, metrics) -> {
+                            ImmutableMap<String, SampleFamily> sampleFamilies = PrometheusMetricConverter
+                                .convertPromMetricToSampleFamily(metrics.stream());
+                            converters.forEach(converter -> converter.toMeter(sampleFamilies));
+                        }
+                    );
                 }
             }
 

diff --git a/.../apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricRequestProcessor.java b/.../apache/skywalking/oap/server/receiver/otel/otlp/OpenTelemetryMetricRequestProcessor.java
@@ -29,6 +29,7 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.oap.meter.analyzer.MetricConvert;
+import org.apache.skywalking.oap.meter.analyzer.dsl.SampleFamily;
 import org.apache.skywalking.oap.meter.analyzer.prometheus.PrometheusMetricConverter;
 import org.apache.skywalking.oap.meter.analyzer.prometheus.rule.Rule;
 import org.apache.skywalking.oap.meter.analyzer.prometheus.rule.Rules;
@@ -76,7 +77,7 @@ public class OpenTelemetryMetricRequestProcessor implements Service {
             .put("job", "job_name")
             .put("service.name", "job_name")
             .build();
-    private List<PrometheusMetricConverter> converters;
+    private List<MetricConvert> converters;
 
     @Getter(lazy = true)
     private final MetricsCreator metricsCreator = manager.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class);
@@ -110,18 +111,16 @@ public void processMetricsRequest(final ExportMetricsServiceRequest requests) {
                         (v1, v2) -> v1
                         ));
 
-                converters
-                    .forEach(convert -> convert.toMeter(
-                        request
-                            .getScopeMetricsList().stream()
-                            .flatMap(scopeMetrics -> scopeMetrics
-                                .getMetricsList().stream()
-                                .flatMap(metric -> adaptMetrics(nodeLabels, metric))
-                                .map(Function1.liftTry(Function.identity()))
-                                .flatMap(tryIt -> MetricConvert.log(
-                                    tryIt,
-                                    "Convert OTEL metric to prometheus metric"
-                                )))));
+                ImmutableMap<String, SampleFamily> sampleFamilies = PrometheusMetricConverter.convertPromMetricToSampleFamily(
+                    request.getScopeMetricsList().stream()
+                           .flatMap(scopeMetrics -> scopeMetrics
+                               .getMetricsList().stream()
+                               .flatMap(metric -> adaptMetrics(nodeLabels, metric))
+                               .map(Function1.liftTry(Function.identity()))
+                               .flatMap(tryIt -> MetricConvert.log(tryIt, "Convert OTEL metric to prometheus metric"))
+                           )
+                );
+                converters.forEach(convert -> convert.toMeter(sampleFamilies));
             });
         }
     }
@@ -146,7 +145,7 @@ public void start() throws ModuleStartException {
 
         converters = rules
             .stream()
-            .map(r -> new PrometheusMetricConverter(r, meterSystem))
+            .map(r -> new MetricConvert(r, meterSystem))
             .collect(toList());
     }