Add telemetry tracer/metric enable flag and integ test (#10395) (#10676)

* Add telemetry tracer/metric enable flag and integ test * Add Changelog * Fix compilation issue * Empty-Commit * Add component flag to traceable wrappers * Address review comment * Address review comment * Address review comment * Address review comment * Address review comment * Address review comment --------- (cherry picked from commit a36ab39) Signed-off-by: Gagan Juneja <gjjuneja@amazon.com> Signed-off-by: Gagan Juneja <gagandeepjuneja@gmail.com> Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: Gagan Juneja <gjjuneja@amazon.com>
opensearch-project · Oct 18, 2023 · 1970085 · 1970085
1 parent 3763207
commit 1970085
Show file tree

Hide file tree

Showing 33 changed files with 429 additions and 58 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -22,6 +22,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Backport the PR #9107 for updating CONCURRENT_SEGMENT_SEARCH_TARGET_MAX_SLICE_COUNT_KEY setting to a dynamic setting ([#10606](https://github.com/opensearch-project/OpenSearch/pull/10606))
 - [Remote Store] Add Remote Store backpressure rejection stats to `_nodes/stats` ([#10524](https://github.com/opensearch-project/OpenSearch/pull/10524))
 - [BUG] Fix java.lang.SecurityException in repository-gcs plugin ([#10642](https://github.com/opensearch-project/OpenSearch/pull/10642))
+- Add telemetry tracer/metric enable flag and integ test. ([#10395](https://github.com/opensearch-project/OpenSearch/pull/10395))
 
 ### Deprecated
 

diff --git a/libs/telemetry/src/main/java/org/opensearch/telemetry/metrics/MetricsTelemetry.java b/libs/telemetry/src/main/java/org/opensearch/telemetry/metrics/MetricsTelemetry.java
@@ -10,14 +10,12 @@
 
 import org.opensearch.common.annotation.ExperimentalApi;
 
-import java.io.Closeable;
-
 /**
  * Interface for metrics telemetry providers
   *
  * @opensearch.experimental
  */
 @ExperimentalApi
-public interface MetricsTelemetry extends MetricsRegistry, Closeable {
+public interface MetricsTelemetry extends MetricsRegistry {
 
 }
diff --git a/libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/DefaultTracer.java b/libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/DefaultTracer.java
@@ -85,6 +85,11 @@ public SpanScope withSpanInScope(Span span) {
         return DefaultSpanScope.create(span, tracerContextStorage).attach();
     }
 
+    @Override
+    public boolean isRecording() {
+        return true;
+    }
+
     private Span createSpan(SpanCreationContext spanCreationContext, Span parentSpan) {
         return tracingTelemetry.createSpan(spanCreationContext, parentSpan);
     }

diff --git a/libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/Tracer.java b/libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/Tracer.java
@@ -53,4 +53,10 @@ public interface Tracer extends HttpTracer, Closeable {
      */
     SpanScope withSpanInScope(Span span);
 
+    /**
+     * Tells if the traces are being recorded or not
+     * @return boolean
+     */
+    boolean isRecording();
+
 }
diff --git a/libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/noop/NoopTracer.java b/libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/noop/NoopTracer.java
@@ -54,6 +54,11 @@ public SpanScope withSpanInScope(Span span) {
         return SpanScope.NO_OP;
     }
 
+    @Override
+    public boolean isRecording() {
+        return false;
+    }
+
     @Override
     public void close() {
 

diff --git a/libs/telemetry/src/test/java/org/opensearch/telemetry/tracing/DefaultTracerTests.java b/libs/telemetry/src/test/java/org/opensearch/telemetry/tracing/DefaultTracerTests.java
@@ -62,6 +62,7 @@ public void testCreateSpan() {
 
         String spanName = defaultTracer.getCurrentSpan().getSpan().getSpanName();
         assertEquals("span_name", spanName);
+        assertTrue(defaultTracer.isRecording());
     }
 
     @SuppressWarnings("unchecked")

diff --git a/...g/IntegrationTestOTelTelemetryPlugin.java → ...y/IntegrationTestOTelTelemetryPlugin.java b/...g/IntegrationTestOTelTelemetryPlugin.java → ...y/IntegrationTestOTelTelemetryPlugin.java
@@ -6,12 +6,9 @@
  * compatible open source license.
  */
 
-package org.opensearch.telemetry.tracing;
+package org.opensearch.telemetry;
 
 import org.opensearch.common.settings.Settings;
-import org.opensearch.telemetry.OTelTelemetryPlugin;
-import org.opensearch.telemetry.Telemetry;
-import org.opensearch.telemetry.TelemetrySettings;
 
 import java.util.Optional;
 

diff --git a/...alClusterTest/java/org/opensearch/telemetry/metrics/InMemorySingletonMetricsExporter.java b/...alClusterTest/java/org/opensearch/telemetry/metrics/InMemorySingletonMetricsExporter.java
@@ -0,0 +1,65 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.telemetry.metrics;
+
+import java.util.Collection;
+import java.util.List;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.export.MetricExporter;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter;
+
+public class InMemorySingletonMetricsExporter implements MetricExporter {
+
+    public static final InMemorySingletonMetricsExporter INSTANCE = new InMemorySingletonMetricsExporter(InMemoryMetricExporter.create());
+
+    private static InMemoryMetricExporter delegate;
+
+    public static InMemorySingletonMetricsExporter create() {
+        return INSTANCE;
+    }
+
+    private InMemorySingletonMetricsExporter(InMemoryMetricExporter delegate) {
+        InMemorySingletonMetricsExporter.delegate = delegate;
+    }
+
+    @Override
+    public CompletableResultCode export(Collection<MetricData> metrics) {
+        return delegate.export(metrics);
+    }
+
+    @Override
+    public CompletableResultCode flush() {
+        return delegate.flush();
+    }
+
+    @Override
+    public CompletableResultCode shutdown() {
+        return delegate.shutdown();
+    }
+
+    public List<MetricData> getFinishedMetricItems() {
+        return delegate.getFinishedMetricItems();
+    }
+
+    /**
+     * Clears the state.
+     */
+    public void reset() {
+        delegate.reset();
+    }
+
+    @Override
+    public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
+        return delegate.getAggregationTemporality(instrumentType);
+    }
+}
diff --git a/...alClusterTest/java/org/opensearch/telemetry/metrics/TelemetryMetricsDisabledSanityIT.java b/...alClusterTest/java/org/opensearch/telemetry/metrics/TelemetryMetricsDisabledSanityIT.java
@@ -0,0 +1,62 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.telemetry.metrics;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.telemetry.IntegrationTestOTelTelemetryPlugin;
+import org.opensearch.telemetry.OTelTelemetrySettings;
+import org.opensearch.telemetry.TelemetrySettings;
+import org.opensearch.telemetry.metrics.noop.NoopCounter;
+import org.opensearch.telemetry.metrics.noop.NoopMetricsRegistry;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, minNumDataNodes = 1)
+public class TelemetryMetricsDisabledSanityIT extends OpenSearchIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(TelemetrySettings.METRICS_FEATURE_ENABLED_SETTING.getKey(), false)
+            .put(
+                OTelTelemetrySettings.OTEL_METRICS_EXPORTER_CLASS_SETTING.getKey(),
+                "org.opensearch.telemetry.metrics.InMemorySingletonMetricsExporter"
+            )
+            .put(TelemetrySettings.METRICS_PUBLISH_INTERVAL_SETTING.getKey(), TimeValue.timeValueSeconds(1))
+            .build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(IntegrationTestOTelTelemetryPlugin.class);
+    }
+
+    @Override
+    protected boolean addMockTelemetryPlugin() {
+        return false;
+    }
+
+    public void testSanityChecksWhenMetricsDisabled() throws Exception {
+        MetricsRegistry metricsRegistry = internalCluster().getInstance(MetricsRegistry.class);
+
+        Counter counter = metricsRegistry.createCounter("test-counter", "test", "1");
+        counter.add(1.0);
+
+        Thread.sleep(2000);
+
+        assertTrue(metricsRegistry instanceof NoopMetricsRegistry);
+        assertTrue(counter instanceof NoopCounter);
+    }
+
+}
diff --git a/...nalClusterTest/java/org/opensearch/telemetry/metrics/TelemetryMetricsEnabledSanityIT.java b/...nalClusterTest/java/org/opensearch/telemetry/metrics/TelemetryMetricsEnabledSanityIT.java
@@ -0,0 +1,99 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.telemetry.metrics;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.telemetry.IntegrationTestOTelTelemetryPlugin;
+import org.opensearch.telemetry.OTelTelemetrySettings;
+import org.opensearch.telemetry.TelemetrySettings;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.junit.After;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.stream.Collectors;
+
+import io.opentelemetry.sdk.metrics.data.DoublePointData;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.SUITE, minNumDataNodes = 1)
+public class TelemetryMetricsEnabledSanityIT extends OpenSearchIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(TelemetrySettings.METRICS_FEATURE_ENABLED_SETTING.getKey(), true)
+            .put(
+                OTelTelemetrySettings.OTEL_METRICS_EXPORTER_CLASS_SETTING.getKey(),
+                "org.opensearch.telemetry.metrics.InMemorySingletonMetricsExporter"
+            )
+            .put(TelemetrySettings.METRICS_PUBLISH_INTERVAL_SETTING.getKey(), TimeValue.timeValueSeconds(1))
+            .build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(IntegrationTestOTelTelemetryPlugin.class);
+    }
+
+    @Override
+    protected boolean addMockTelemetryPlugin() {
+        return false;
+    }
+
+    public void testCounter() throws Exception {
+        MetricsRegistry metricsRegistry = internalCluster().getInstance(MetricsRegistry.class);
+        InMemorySingletonMetricsExporter.INSTANCE.reset();
+
+        Counter counter = metricsRegistry.createCounter("test-counter", "test", "1");
+        counter.add(1.0);
+        // Sleep for about 2s to wait for metrics to be published.
+        Thread.sleep(2000);
+
+        InMemorySingletonMetricsExporter exporter = InMemorySingletonMetricsExporter.INSTANCE;
+        double value = ((DoublePointData) ((ArrayList) exporter.getFinishedMetricItems()
+            .stream()
+            .filter(a -> a.getName().equals("test-counter"))
+            .collect(Collectors.toList())
+            .get(0)
+            .getDoubleSumData()
+            .getPoints()).get(0)).getValue();
+        assertEquals(1.0, value, 0.0);
+    }
+
+    public void testUpDownCounter() throws Exception {
+
+        MetricsRegistry metricsRegistry = internalCluster().getInstance(MetricsRegistry.class);
+        InMemorySingletonMetricsExporter.INSTANCE.reset();
+
+        Counter counter = metricsRegistry.createUpDownCounter("test-up-down-counter", "test", "1");
+        counter.add(1.0);
+        counter.add(-2.0);
+        // Sleep for about 2s to wait for metrics to be published.
+        Thread.sleep(2000);
+
+        InMemorySingletonMetricsExporter exporter = InMemorySingletonMetricsExporter.INSTANCE;
+        double value = ((DoublePointData) ((ArrayList) exporter.getFinishedMetricItems()
+            .stream()
+            .filter(a -> a.getName().equals("test-up-down-counter"))
+            .collect(Collectors.toList())
+            .get(0)
+            .getDoubleSumData()
+            .getPoints()).get(0)).getValue();
+        assertEquals(-1.0, value, 0.0);
+    }
+
+    @After
+    public void reset() {
+        InMemorySingletonMetricsExporter.INSTANCE.reset();
+    }
+}
diff --git a/...nalClusterTest/java/org/opensearch/telemetry/tracing/TelemetryTracerDisabledSanityIT.java b/...nalClusterTest/java/org/opensearch/telemetry/tracing/TelemetryTracerDisabledSanityIT.java
@@ -12,6 +12,7 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.telemetry.IntegrationTestOTelTelemetryPlugin;
 import org.opensearch.telemetry.OTelTelemetrySettings;
 import org.opensearch.telemetry.TelemetrySettings;
 import org.opensearch.test.OpenSearchIntegTestCase;

diff --git a/...rnalClusterTest/java/org/opensearch/telemetry/tracing/TelemetryTracerEnabledSanityIT.java b/...rnalClusterTest/java/org/opensearch/telemetry/tracing/TelemetryTracerEnabledSanityIT.java
@@ -12,6 +12,7 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.telemetry.IntegrationTestOTelTelemetryPlugin;
 import org.opensearch.telemetry.OTelTelemetrySettings;
 import org.opensearch.telemetry.TelemetrySettings;
 import org.opensearch.telemetry.tracing.attributes.Attributes;
@@ -88,9 +89,7 @@ public void testSanityChecksWhenTracingEnabled() throws Exception {
         );
 
         InMemorySingletonSpanExporter exporter = InMemorySingletonSpanExporter.INSTANCE;
-        if (!exporter.getFinishedSpanItems().isEmpty()) {
-            validators.validate(exporter.getFinishedSpanItems(), 6);
-        }
+        validators.validate(exporter.getFinishedSpanItems(), 6);
     }
 
     private static void updateTelemetrySetting(Client client, boolean value) {

diff --git a/plugins/telemetry-otel/src/main/java/org/opensearch/telemetry/OTelTelemetryPlugin.java b/plugins/telemetry-otel/src/main/java/org/opensearch/telemetry/OTelTelemetryPlugin.java
@@ -8,14 +8,13 @@
 
 package org.opensearch.telemetry;
 
+import org.opensearch.common.concurrent.RefCountedReleasable;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.plugins.TelemetryPlugin;
-import org.opensearch.telemetry.metrics.OTelMetricsTelemetry;
 import org.opensearch.telemetry.tracing.OTelResourceProvider;
 import org.opensearch.telemetry.tracing.OTelTelemetry;
-import org.opensearch.telemetry.tracing.OTelTracingTelemetry;
 
 import java.util.Arrays;
 import java.util.List;
@@ -37,6 +36,8 @@ public class OTelTelemetryPlugin extends Plugin implements TelemetryPlugin {
 
     private final Settings settings;
 
+    private RefCountedReleasable<OpenTelemetrySdk> refCountedOpenTelemetry;
+
     /**
      * Creates Otel plugin
      * @param settings cluster settings
@@ -58,20 +59,32 @@ public List<Setting<?>> getSettings() {
 
     @Override
     public Optional<Telemetry> getTelemetry(TelemetrySettings telemetrySettings) {
+        initializeOpenTelemetrySdk(telemetrySettings);
         return Optional.of(telemetry(telemetrySettings));
     }
 
+    private void initializeOpenTelemetrySdk(TelemetrySettings telemetrySettings) {
+        if (refCountedOpenTelemetry != null) {
+            return;
+        }
+        OpenTelemetrySdk openTelemetrySdk = OTelResourceProvider.get(telemetrySettings, settings);
+        refCountedOpenTelemetry = new RefCountedReleasable<>("openTelemetry", openTelemetrySdk, openTelemetrySdk::close);
+    }
+
     @Override
     public String getName() {
         return OTEL_TRACER_NAME;
     }
 
     private Telemetry telemetry(TelemetrySettings telemetrySettings) {
-        final OpenTelemetrySdk openTelemetry = OTelResourceProvider.get(telemetrySettings, settings);
-        return new OTelTelemetry(
-            new OTelTracingTelemetry<>(openTelemetry, openTelemetry.getSdkTracerProvider()),
-            new OTelMetricsTelemetry<>(openTelemetry.getSdkMeterProvider())
-        );
+        return new OTelTelemetry(refCountedOpenTelemetry);
+    }
+
+    @Override
+    public void close() {
+        if (refCountedOpenTelemetry != null) {
+            refCountedOpenTelemetry.close();
+        }
     }
 
 }