Added integration tests support for Telemetry span validation

CHANGELOG.md

@@ -51,6 +51,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Pass localNode info to all plugins on node start ([#7919](https://github.com/opensearch-project/OpenSearch/pull/7919))
 - Improved performance of parsing floating point numbers ([#7909](https://github.com/opensearch-project/OpenSearch/pull/7909))
 - Move span actions to Scope ([#8411](https://github.com/opensearch-project/OpenSearch/pull/8411))
+- Added Integration Tests for Distributed Tracing framework ([#8444](https://github.com/opensearch-project/OpenSearch/pull/8444))
 - Make Span exporter configurable ([#8620](https://github.com/opensearch-project/OpenSearch/issues/8620))
 - Add wrapper tracer implementation
 

plugins/telemetry-otel/build.gradle

@@ -11,6 +11,8 @@ import org.opensearch.gradle.Architecture
 import org.opensearch.gradle.OS
 import org.opensearch.gradle.info.BuildParams
 
+apply plugin: 'opensearch.internal-cluster-test'
+
 opensearchplugin {
   description 'Opentelemetry based telemetry implementation.'
   classname 'org.opensearch.telemetry.OTelTelemetryPlugin'

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/tracing/InMemorySpanExporter.java

@@ -0,0 +1,86 @@
+/*
+ * 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.tracing;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+import org.opensearch.test.telemetry.tracing.MockSpanData;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ * Writes the span to the list of MockSpanData, everytime a span is emitted.
+ */
+public class InMemorySpanExporter implements SpanExporter {
+
+    private final AtomicBoolean isShutdown = new AtomicBoolean();
+
+    private static List<MockSpanData> finishedSpanItems = new ArrayList<>();
+
+    /**
+     * No-args constructor
+     */
+    public InMemorySpanExporter() {}
+
+    @Override
+    public CompletableResultCode export(Collection<SpanData> spans) {
+        if (isShutdown.get()) {
+            return CompletableResultCode.ofFailure();
+        }
+        for (SpanData span : spans) {
+            finishedSpanItems.add(convertSpanDataToMockSpanData(span));
+        }
+        return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode flush() {
+        return CompletableResultCode.ofSuccess();
+    }
+
+    /**
+     * Returns All the spans generated at any point of time.
+     */
+    public List<MockSpanData> getFinishedSpanItems() {
+        return finishedSpanItems;
+    }
+
+    @Override
+    public CompletableResultCode shutdown() {
+        assertThat("Duplicate shutdown() calls.", isShutdown.compareAndSet(false, true), equalTo(true));
+        return CompletableResultCode.ofSuccess();
+    }
+
+    private MockSpanData convertSpanDataToMockSpanData(SpanData spanData) {
+        MockSpanData span = new MockSpanData(
+            spanData.getSpanId(),
+            spanData.getParentSpanId(),
+            spanData.getTraceId(),
+            spanData.getStartEpochNanos(),
+            spanData.getEndEpochNanos(),
+            spanData.hasEnded(),
+            spanData.getName()
+        );
+        return span;
+    }
+
+    /**
+     * Used by SpanExporterFactory to instantiate InMemorySpanExporter
+     */
+    public static InMemorySpanExporter create() {
+        return new InMemorySpanExporter();
+    }
+}

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/tracing/IntegrationTestOTelTelemetryPlugin.java

@@ -0,0 +1,40 @@
+/*
+ * 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.tracing;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+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;
+
+/**
+ * Telemetry plugin used for Integration tests.
+*/
+public class IntegrationTestOTelTelemetryPlugin extends OTelTelemetryPlugin {
+    /**
+     * Creates IntegrationTestOTelTelemetryPlugin
+     * @param settings cluster settings
+     */
+    public IntegrationTestOTelTelemetryPlugin(Settings settings) {
+        super(settings);
+    }
+
+    /**
+     * This method overrides getTelemetry() method in OTel plugin class, so we create only one instance of global OpenTelemetry
+     * resetForTest() will set OpenTelemetry to null again.
+     * @param settings cluster settings
+     */
+    public Optional<Telemetry> getTelemetry(TelemetrySettings settings) {
+        GlobalOpenTelemetry.resetForTest();
+        return super.getTelemetry(settings);
+    }
+}

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/tracing/TelemetryTracerDisabledSanityIT.java

@@ -0,0 +1,75 @@
+/*
+ * 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.tracing;
+
+import org.opensearch.client.Client;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.telemetry.OTelTelemetrySettings;
+import org.opensearch.telemetry.TelemetrySettings;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.opensearch.index.query.QueryBuilders.queryStringQuery;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, supportsDedicatedMasters = false, minNumDataNodes = 2)
+public class TelemetryTracerDisabledSanityIT extends OpenSearchIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(
+                OTelTelemetrySettings.OTEL_TRACER_SPAN_EXPORTER_CLASS_SETTING.getKey(),
+                "org.opensearch.telemetry.tracing.InMemorySpanExporter"
+            )
+            .build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(IntegrationTestOTelTelemetryPlugin.class);
+    }
+
+    @Override
+    protected boolean addMockTelemetryPlugin() {
+        return false;
+    }
+
+    public void testSanityCheckWhenTracingDisabled() throws Exception {
+        Client client = client();
+        // DISABLE TRACING
+        client.admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setTransientSettings(Settings.builder().put(TelemetrySettings.TRACER_ENABLED_SETTING.getKey(), false))
+            .get();
+
+        // Create Index and ingest data
+        String indexName = "test-index-11";
+        Settings basicSettings = Settings.builder().put("number_of_shards", 3).put("number_of_replicas", 1).build();
+        createIndex(indexName, basicSettings);
+        indexRandom(true, client.prepareIndex(indexName).setId("1").setSource("field1", "t`"));
+
+        ensureGreen();
+        refresh();
+
+        // Make the search call;
+        client.prepareSearch().setQuery(queryStringQuery("fox")).get();
+
+        // Sleep for about 2s to wait for traces are published
+        Thread.sleep(2000);
+
+        InMemorySpanExporter exporter = new InMemorySpanExporter();
+        assertTrue(exporter.getFinishedSpanItems().isEmpty());
+    }
+
+}

plugins/telemetry-otel/src/internalClusterTest/java/org/opensearch/telemetry/tracing/TelemetryTracerEnabledSanityIT.java

@@ -0,0 +1,94 @@
+/*
+ * 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.tracing;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.telemetry.TelemetrySettings;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.opensearch.telemetry.OTelTelemetrySettings;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.client.Client;
+import org.opensearch.test.telemetry.tracing.TelemetryValidators;
+import org.opensearch.test.telemetry.tracing.validators.AllSpansAreEndedProperly;
+import org.opensearch.test.telemetry.tracing.validators.AllSpansAreInOrder;
+import org.opensearch.test.telemetry.tracing.validators.AllSpansHaveUniqueId;
+import org.opensearch.test.telemetry.tracing.validators.NumberOfTraceIDsEqualToRequests;
+import org.opensearch.test.telemetry.tracing.validators.TotalParentSpansEqualToRequests;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.opensearch.index.query.QueryBuilders.queryStringQuery;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, minNumDataNodes = 2)
+public class TelemetryTracerEnabledSanityIT extends OpenSearchIntegTestCase {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(
+                OTelTelemetrySettings.OTEL_TRACER_SPAN_EXPORTER_CLASS_SETTING.getKey(),
+                "org.opensearch.telemetry.tracing.InMemorySpanExporter"
+            )
+            .build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(IntegrationTestOTelTelemetryPlugin.class);
+    }
+
+    @Override
+    protected boolean addMockTelemetryPlugin() {
+        return false;
+    }
+
+    public void testSanityChecksWhenTracingEnabled() throws Exception {
+        Client client = client();
+        // ENABLE TRACING
+        client.admin()
+            .cluster()
+            .prepareUpdateSettings()
+            .setTransientSettings(Settings.builder().put(TelemetrySettings.TRACER_ENABLED_SETTING.getKey(), true))
+            .get();
+
+        // Create Index and ingest data
+        String indexName = "test-index-11";
+        Settings basicSettings = Settings.builder().put("number_of_shards", 3).put("number_of_replicas", 0).build();
+        createIndex(indexName, basicSettings);
+        indexRandom(true, client.prepareIndex(indexName).setId("1").setSource("field1", "the fox jumps in the well"));
+        indexRandom(true, client.prepareIndex(indexName).setId("1").setSource("field2", "another fox did the same."));
+
+        ensureGreen();
+        refresh();
+
+        // Make the search calls;
+        client.prepareSearch().setQuery(queryStringQuery("fox")).get();
+        client.prepareSearch().setQuery(queryStringQuery("jumps")).get();
+
+        // Sleep for about 3s to wait for traces are published
+        Thread.sleep(3000);
+
+        TelemetryValidators validators = new TelemetryValidators(
+            Arrays.asList(
+                AllSpansAreEndedProperly.class,
+                AllSpansAreInOrder.class,
+                AllSpansHaveUniqueId.class,
+                NumberOfTraceIDsEqualToRequests.class,
+                TotalParentSpansEqualToRequests.class
+            )
+        );
+        InMemorySpanExporter exporter = new InMemorySpanExporter();
+        if (!exporter.getFinishedSpanItems().isEmpty()) {
+            validators.validate(exporter.getFinishedSpanItems(), 2);
+        }
+    }
+
+}

server/src/main/java/org/opensearch/node/Node.java

@@ -1419,6 +1419,7 @@ private Node stop() {
         // we should stop this last since it waits for resources to get released
         // if we had scroll searchers etc or recovery going on we wait for to finish.
         injector.getInstance(IndicesService.class).stop();
+        injector.getInstance(TracerFactory.class).close();
         logger.info("stopped");
 
         return this;