[Backport] [2.x] - Add Tracing Instrumentation at Network and Rest layer

CHANGELOG.md

@@ -122,6 +122,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Move ZStd to a plugin ([#9658](https://github.com/opensearch-project/OpenSearch/pull/9658))
 - [Remote Store] Add support for Remote Translog Store upload stats in `_nodes/stats/` API ([#8908](https://github.com/opensearch-project/OpenSearch/pull/8908))
 - [Remote Store] Removing feature flag to mark feature GA ([#9761](https://github.com/opensearch-project/OpenSearch/pull/9761))
+- Add instrumentation in rest and network layer. ([#9415](https://github.com/opensearch-project/OpenSearch/pull/9415))
 
 ### Deprecated
 

libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/DefaultTracer.java

@@ -119,9 +119,13 @@ protected void addDefaultAttributes(Span span) {
     }
 
     @Override
-    public Span startSpan(String spanName, Map<String, List<String>> headers, Attributes attributes) {
+    public Span startSpan(SpanCreationContext spanCreationContext, Map<String, List<String>> headers) {
         Optional<Span> propagatedSpan = tracingTelemetry.getContextPropagator().extractFromHeaders(headers);
-        return startSpan(spanName, propagatedSpan.map(SpanContext::new).orElse(null), attributes);
+        return startSpan(
+            spanCreationContext.getSpanName(),
+            propagatedSpan.map(SpanContext::new).orElse(null),
+            spanCreationContext.getAttributes()
+        );
     }
 
 }

libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/http/HttpTracer.java

@@ -10,7 +10,7 @@
 
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.telemetry.tracing.Span;
-import org.opensearch.telemetry.tracing.attributes.Attributes;
+import org.opensearch.telemetry.tracing.SpanCreationContext;
 
 import java.util.List;
 import java.util.Map;
@@ -28,10 +28,9 @@ public interface HttpTracer {
     /**
      * Start the span with propagating the tracing info from the HttpRequest header.
      *
-     * @param spanName span name.
+     * @param spanCreationContext span name.
      * @param header http request header.
-     * @param attributes span attributes.
      * @return span.
      */
-    Span startSpan(String spanName, Map<String, List<String>> header, Attributes attributes);
+    Span startSpan(SpanCreationContext spanCreationContext, Map<String, List<String>> header);
 }

libs/telemetry/src/main/java/org/opensearch/telemetry/tracing/noop/NoopTracer.java

@@ -81,7 +81,7 @@ public void close() {
     }
 
     @Override
-    public Span startSpan(String spanName, Map<String, List<String>> header, Attributes attributes) {
+    public Span startSpan(SpanCreationContext spanCreationContext, Map<String, List<String>> header) {
         return NoopSpan.INSTANCE;
     }
 }

modules/transport-netty4/src/main/java/org/opensearch/http/netty4/Netty4HttpServerTransport.java

@@ -52,6 +52,7 @@
 import org.opensearch.http.HttpHandlingSettings;
 import org.opensearch.http.HttpReadTimeoutException;
 import org.opensearch.http.HttpServerChannel;
+import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.NettyAllocator;
 import org.opensearch.transport.NettyByteBufSizer;
@@ -174,9 +175,10 @@ public Netty4HttpServerTransport(
         NamedXContentRegistry xContentRegistry,
         Dispatcher dispatcher,
         ClusterSettings clusterSettings,
-        SharedGroupFactory sharedGroupFactory
+        SharedGroupFactory sharedGroupFactory,
+        Tracer tracer
     ) {
-        super(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher, clusterSettings);
+        super(settings, networkService, bigArrays, threadPool, xContentRegistry, dispatcher, clusterSettings, tracer);
         Netty4Utils.setAvailableProcessors(OpenSearchExecutors.NODE_PROCESSORS_SETTING.get(settings));
         NettyAllocator.logAllocatorDescriptionIfNeeded();
         this.sharedGroupFactory = sharedGroupFactory;

modules/transport-netty4/src/main/java/org/opensearch/transport/Netty4Plugin.java

@@ -48,6 +48,7 @@
 import org.opensearch.http.netty4.Netty4HttpServerTransport;
 import org.opensearch.plugins.NetworkPlugin;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.netty4.Netty4Transport;
 
@@ -122,7 +123,8 @@ public Map<String, Supplier<HttpServerTransport>> getHttpTransports(
         NamedXContentRegistry xContentRegistry,
         NetworkService networkService,
         HttpServerTransport.Dispatcher dispatcher,
-        ClusterSettings clusterSettings
+        ClusterSettings clusterSettings,
+        Tracer tracer
     ) {
         return Collections.singletonMap(
             NETTY_HTTP_TRANSPORT_NAME,
@@ -134,7 +136,8 @@ public Map<String, Supplier<HttpServerTransport>> getHttpTransports(
                 xContentRegistry,
                 dispatcher,
                 clusterSettings,
-                getSharedGroupFactory(settings)
+                getSharedGroupFactory(settings),
+                tracer
             )
         );
     }

modules/transport-netty4/src/test/java/org/opensearch/http/netty4/Netty4BadRequestTests.java

@@ -47,6 +47,7 @@
 import org.opensearch.rest.BytesRestResponse;
 import org.opensearch.rest.RestChannel;
 import org.opensearch.rest.RestRequest;
+import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.OpenSearchTestCase;
 import org.opensearch.threadpool.TestThreadPool;
 import org.opensearch.threadpool.ThreadPool;
@@ -112,7 +113,8 @@ public void dispatchBadRequest(RestChannel channel, ThreadContext threadContext,
                 xContentRegistry(),
                 dispatcher,
                 new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
-                new SharedGroupFactory(Settings.EMPTY)
+                new SharedGroupFactory(Settings.EMPTY),
+                NoopTracer.INSTANCE
             )
         ) {
             httpServerTransport.start();

modules/transport-netty4/src/test/java/org/opensearch/http/netty4/Netty4HttpServerPipeliningTests.java

@@ -45,6 +45,7 @@
 import org.opensearch.http.HttpResponse;
 import org.opensearch.http.HttpServerTransport;
 import org.opensearch.http.NullDispatcher;
+import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.OpenSearchTestCase;
 import org.opensearch.threadpool.TestThreadPool;
 import org.opensearch.threadpool.ThreadPool;
@@ -135,7 +136,8 @@ class CustomNettyHttpServerTransport extends Netty4HttpServerTransport {
                 xContentRegistry(),
                 new NullDispatcher(),
                 new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
-                new SharedGroupFactory(settings)
+                new SharedGroupFactory(settings),
+                NoopTracer.INSTANCE
             );
         }
 

modules/transport-netty4/src/test/java/org/opensearch/http/netty4/Netty4HttpServerTransportTests.java

@@ -55,6 +55,7 @@
 import org.opensearch.rest.BytesRestResponse;
 import org.opensearch.rest.RestChannel;
 import org.opensearch.rest.RestRequest;
+import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.OpenSearchTestCase;
 import org.opensearch.test.rest.FakeRestRequest;
 import org.opensearch.threadpool.TestThreadPool;
@@ -198,7 +199,8 @@ public void dispatchBadRequest(RestChannel channel, ThreadContext threadContext,
                 xContentRegistry(),
                 dispatcher,
                 clusterSettings,
-                new SharedGroupFactory(settings)
+                new SharedGroupFactory(settings),
+                NoopTracer.INSTANCE
             )
         ) {
             transport.start();
@@ -247,7 +249,8 @@ public void testBindUnavailableAddress() {
                 xContentRegistry(),
                 new NullDispatcher(),
                 clusterSettings,
-                new SharedGroupFactory(Settings.EMPTY)
+                new SharedGroupFactory(Settings.EMPTY),
+                NoopTracer.INSTANCE
             )
         ) {
             transport.start();
@@ -265,7 +268,8 @@ public void testBindUnavailableAddress() {
                     xContentRegistry(),
                     new NullDispatcher(),
                     clusterSettings,
-                    new SharedGroupFactory(settings)
+                    new SharedGroupFactory(settings),
+                    NoopTracer.INSTANCE
                 )
             ) {
                 BindHttpException bindHttpException = expectThrows(BindHttpException.class, otherTransport::start);
@@ -317,7 +321,8 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
                 xContentRegistry(),
                 dispatcher,
                 clusterSettings,
-                new SharedGroupFactory(settings)
+                new SharedGroupFactory(settings),
+                NoopTracer.INSTANCE
             )
         ) {
             transport.start();
@@ -379,7 +384,8 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
                 xContentRegistry(),
                 dispatcher,
                 clusterSettings,
-                new SharedGroupFactory(Settings.EMPTY)
+                new SharedGroupFactory(Settings.EMPTY),
+                NoopTracer.INSTANCE
             )
         ) {
             transport.start();
@@ -448,7 +454,8 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
                 xContentRegistry(),
                 dispatcher,
                 new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
-                new SharedGroupFactory(settings)
+                new SharedGroupFactory(settings),
+                NoopTracer.INSTANCE
             )
         ) {
             transport.start();
@@ -521,7 +528,8 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
                 xContentRegistry(),
                 dispatcher,
                 new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
-                new SharedGroupFactory(settings)
+                new SharedGroupFactory(settings),
+                NoopTracer.INSTANCE
             )
         ) {
             transport.start();

plugins/discovery-ec2/src/test/java/org/opensearch/discovery/ec2/Ec2DiscoveryTests.java

@@ -48,6 +48,7 @@
 import org.opensearch.common.util.PageCacheRecycler;
 import org.opensearch.core.common.transport.TransportAddress;
 import org.opensearch.core.indices.breaker.NoneCircuitBreakerService;
+import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.transport.MockTransportService;
 import org.opensearch.transport.Transport;
 import org.opensearch.transport.TransportService;
@@ -99,7 +100,14 @@ public TransportAddress[] addressesFromString(String address) {
                 return new TransportAddress[] { poorMansDNS.getOrDefault(address, buildNewFakeTransportAddress()) };
             }
         };
-        return new MockTransportService(Settings.EMPTY, transport, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR, null);
+        return new MockTransportService(
+            Settings.EMPTY,
+            transport,
+            threadPool,
+            TransportService.NOOP_TRANSPORT_INTERCEPTOR,
+            null,
+            NoopTracer.INSTANCE
+        );
     }
 
     protected List<TransportAddress> buildDynamicHosts(Settings nodeSettings, int nodes) {

plugins/discovery-ec2/src/test/java/org/opensearch/discovery/ec2/Ec2RetriesTests.java

@@ -47,6 +47,7 @@
 import org.opensearch.core.indices.breaker.NoneCircuitBreakerService;
 import org.opensearch.discovery.SeedHostsProvider;
 import org.opensearch.discovery.SeedHostsResolver;
+import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.transport.MockTransportService;
 import org.opensearch.transport.TransportService;
 import org.opensearch.transport.nio.MockNioTransport;
@@ -80,7 +81,8 @@ protected MockTransportService createTransportService() {
             ),
             threadPool,
             TransportService.NOOP_TRANSPORT_INTERCEPTOR,
-            null
+            null,
+            NoopTracer.INSTANCE
         );
     }
 

plugins/discovery-gce/src/test/java/org/opensearch/discovery/gce/GceDiscoveryTests.java

@@ -38,6 +38,7 @@
 import org.opensearch.common.network.NetworkService;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.common.transport.TransportAddress;
+import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.OpenSearchTestCase;
 import org.opensearch.test.transport.MockTransportService;
 import org.opensearch.threadpool.TestThreadPool;
@@ -109,7 +110,7 @@ public void setProjectName() {
 
     @Before
     public void createTransportService() {
-        transportService = MockTransportService.createNewService(Settings.EMPTY, Version.CURRENT, threadPool, null);
+        transportService = MockTransportService.createNewService(Settings.EMPTY, Version.CURRENT, threadPool, NoopTracer.INSTANCE);
     }
 
     @After

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

@@ -11,7 +11,9 @@
 import org.opensearch.test.telemetry.tracing.MockSpanData;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import io.opentelemetry.sdk.common.CompletableResultCode;
@@ -21,7 +23,7 @@
 
 public class InMemorySingletonSpanExporter implements SpanExporter {
 
-    private static final InMemorySingletonSpanExporter INSTANCE = new InMemorySingletonSpanExporter(InMemorySpanExporter.create());
+    public static final InMemorySingletonSpanExporter INSTANCE = new InMemorySingletonSpanExporter(InMemorySpanExporter.create());
 
     private static InMemorySpanExporter delegate;
 
@@ -62,10 +64,30 @@ private List<MockSpanData> convertSpanDataListToMockSpanDataList(List<SpanData>
                     spanData.getStartEpochNanos(),
                     spanData.getEndEpochNanos(),
                     spanData.hasEnded(),
-                    spanData.getName()
+                    spanData.getName(),
+                    getAttributes(spanData)
                 )
             )
             .collect(Collectors.toList());
         return mockSpanDataList;
     }
+
+    private Map<String, Object> getAttributes(SpanData spanData) {
+        if (spanData.getAttributes() != null) {
+            return spanData.getAttributes()
+                .asMap()
+                .entrySet()
+                .stream()
+                .collect(Collectors.toMap(e -> e.getKey().getKey(), e -> e.getValue()));
+        } else {
+            return Collections.emptyMap();
+        }
+    }
+
+    /**
+     * Clears the state.
+     */
+    public void reset() {
+        delegate.reset();
+    }
 }

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

@@ -63,14 +63,15 @@ public void testSanityCheckWhenTracingDisabled() throws Exception {
 
         ensureGreen();
         refresh();
+        InMemorySingletonSpanExporter exporter = InMemorySingletonSpanExporter.INSTANCE;
+        exporter.reset();
 
         // Make the search call;
         client.prepareSearch().setQuery(queryStringQuery("fox")).get();
 
         // Sleep for about 3s to wait for traces are published (the delay is 1s)
         Thread.sleep(3000);
 
-        InMemorySingletonSpanExporter exporter = InMemorySingletonSpanExporter.create();
         assertTrue(exporter.getFinishedSpanItems().isEmpty());
     }