[Tracing Instrumentation] Add instrumentation in InboundHandler

CHANGELOG.md

@@ -125,6 +125,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Tracing Framework] Add support for SpanKind. ([#10122](https://github.com/opensearch-project/OpenSearch/pull/10122))
 - Pass parent filter to inner query in nested query ([#10246](https://github.com/opensearch-project/OpenSearch/pull/10246))
 - Disable concurrent segment search when terminate_after is used ([#10200](https://github.com/opensearch-project/OpenSearch/pull/10200))
+- Add instrumentation in Inbound Handler. ([#100143](https://github.com/opensearch-project/OpenSearch/pull/10143))
 - Enable remote segment upload backpressure by default ([#10356](https://github.com/opensearch-project/OpenSearch/pull/10356))
 
 ### Deprecated

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

@@ -96,7 +96,8 @@ public Map<String, Supplier<Transport>> getTransports(
         PageCacheRecycler pageCacheRecycler,
         CircuitBreakerService circuitBreakerService,
         NamedWriteableRegistry namedWriteableRegistry,
-        NetworkService networkService
+        NetworkService networkService,
+        Tracer tracer
     ) {
         return Collections.singletonMap(
             NETTY_TRANSPORT_NAME,
@@ -108,7 +109,8 @@ public Map<String, Supplier<Transport>> getTransports(
                 pageCacheRecycler,
                 namedWriteableRegistry,
                 circuitBreakerService,
-                getSharedGroupFactory(settings)
+                getSharedGroupFactory(settings),
+                tracer
             )
         );
     }

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

@@ -50,6 +50,7 @@
 import org.opensearch.core.common.unit.ByteSizeUnit;
 import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.core.indices.breaker.CircuitBreakerService;
+import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.Netty4NioSocketChannel;
 import org.opensearch.transport.NettyAllocator;
@@ -131,9 +132,10 @@ public Netty4Transport(
         PageCacheRecycler pageCacheRecycler,
         NamedWriteableRegistry namedWriteableRegistry,
         CircuitBreakerService circuitBreakerService,
-        SharedGroupFactory sharedGroupFactory
+        SharedGroupFactory sharedGroupFactory,
+        Tracer tracer
     ) {
-        super(settings, version, threadPool, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, networkService);
+        super(settings, version, threadPool, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, networkService, tracer);
         Netty4Utils.setAvailableProcessors(OpenSearchExecutors.NODE_PROCESSORS_SETTING.get(settings));
         NettyAllocator.logAllocatorDescriptionIfNeeded();
         this.sharedGroupFactory = sharedGroupFactory;

modules/transport-netty4/src/test/java/org/opensearch/transport/netty4/Netty4SizeHeaderFrameDecoderTests.java

@@ -40,6 +40,7 @@
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 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.OpenSearchTestCase;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.SharedGroupFactory;
@@ -86,7 +87,8 @@ public void startThreadPool() {
             recycler,
             new NamedWriteableRegistry(Collections.emptyList()),
             new NoneCircuitBreakerService(),
-            new SharedGroupFactory(settings)
+            new SharedGroupFactory(settings),
+            NoopTracer.INSTANCE
         );
         nettyTransport.start();
 

modules/transport-netty4/src/test/java/org/opensearch/transport/netty4/NettyTransportMultiPortTests.java

@@ -40,6 +40,7 @@
 import org.opensearch.common.util.PageCacheRecycler;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.indices.breaker.NoneCircuitBreakerService;
+import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.OpenSearchTestCase;
 import org.opensearch.threadpool.TestThreadPool;
 import org.opensearch.threadpool.ThreadPool;
@@ -141,7 +142,8 @@ private TcpTransport startTransport(Settings settings, ThreadPool threadPool) {
             recycler,
             new NamedWriteableRegistry(Collections.emptyList()),
             new NoneCircuitBreakerService(),
-            new SharedGroupFactory(settings)
+            new SharedGroupFactory(settings),
+            NoopTracer.INSTANCE
         );
         transport.start();
 

modules/transport-netty4/src/test/java/org/opensearch/transport/netty4/SimpleNetty4TransportTests.java

@@ -44,6 +44,7 @@
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 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.test.transport.StubbableTransport;
 import org.opensearch.transport.AbstractSimpleTransportTestCase;
@@ -82,7 +83,8 @@ protected Transport build(Settings settings, final Version version, ClusterSetti
             PageCacheRecycler.NON_RECYCLING_INSTANCE,
             namedWriteableRegistry,
             new NoneCircuitBreakerService(),
-            new SharedGroupFactory(settings)
+            new SharedGroupFactory(settings),
+            NoopTracer.INSTANCE
         ) {
 
             @Override

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

@@ -92,7 +92,8 @@ protected MockTransportService createTransportService() {
             new NetworkService(Collections.emptyList()),
             PageCacheRecycler.NON_RECYCLING_INSTANCE,
             writableRegistry(),
-            new NoneCircuitBreakerService()
+            new NoneCircuitBreakerService(),
+            NoopTracer.INSTANCE
         ) {
             @Override
             public TransportAddress[] addressesFromString(String address) {

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

@@ -77,7 +77,8 @@ protected MockTransportService createTransportService() {
                 networkService,
                 PageCacheRecycler.NON_RECYCLING_INSTANCE,
                 new NamedWriteableRegistry(Collections.emptyList()),
-                new NoneCircuitBreakerService()
+                new NoneCircuitBreakerService(),
+                NoopTracer.INSTANCE
             ),
             threadPool,
             TransportService.NOOP_TRANSPORT_INTERCEPTOR,

plugins/telemetry-otel/src/main/java/org/opensearch/telemetry/tracing/OTelSpan.java

@@ -57,7 +57,9 @@ public void addAttribute(String key, Boolean value) {
 
     @Override
     public void setError(Exception exception) {
-        delegateSpan.setStatus(StatusCode.ERROR, exception.getMessage());
+        if (exception != null) {
+            delegateSpan.setStatus(StatusCode.ERROR, exception.getMessage());
+        }
     }
 
     @Override

plugins/transport-nio/src/main/java/org/opensearch/transport/nio/NioTransport.java

@@ -52,6 +52,7 @@
 import org.opensearch.nio.NioSelector;
 import org.opensearch.nio.NioSocketChannel;
 import org.opensearch.nio.ServerChannelContext;
+import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TcpTransport;
 import org.opensearch.transport.TransportSettings;
@@ -84,9 +85,10 @@ protected NioTransport(
         PageCacheRecycler pageCacheRecycler,
         NamedWriteableRegistry namedWriteableRegistry,
         CircuitBreakerService circuitBreakerService,
-        NioGroupFactory groupFactory
+        NioGroupFactory groupFactory,
+        Tracer tracer
     ) {
-        super(settings, version, threadPool, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, networkService);
+        super(settings, version, threadPool, pageCacheRecycler, circuitBreakerService, namedWriteableRegistry, networkService, tracer);
         this.pageAllocator = new PageAllocator(pageCacheRecycler);
         this.groupFactory = groupFactory;
     }

plugins/transport-nio/src/main/java/org/opensearch/transport/nio/NioTransportPlugin.java

@@ -91,7 +91,8 @@ public Map<String, Supplier<Transport>> getTransports(
         PageCacheRecycler pageCacheRecycler,
         CircuitBreakerService circuitBreakerService,
         NamedWriteableRegistry namedWriteableRegistry,
-        NetworkService networkService
+        NetworkService networkService,
+        Tracer tracer
     ) {
         return Collections.singletonMap(
             NIO_TRANSPORT_NAME,
@@ -103,7 +104,8 @@ public Map<String, Supplier<Transport>> getTransports(
                 pageCacheRecycler,
                 namedWriteableRegistry,
                 circuitBreakerService,
-                getNioGroupFactory(settings)
+                getNioGroupFactory(settings),
+                tracer
             )
         );
     }

plugins/transport-nio/src/test/java/org/opensearch/transport/nio/SimpleNioTransportTests.java

@@ -44,6 +44,7 @@
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 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.test.transport.StubbableTransport;
 import org.opensearch.transport.AbstractSimpleTransportTestCase;
@@ -81,7 +82,8 @@ protected Transport build(Settings settings, final Version version, ClusterSetti
             new MockPageCacheRecycler(settings),
             namedWriteableRegistry,
             new NoneCircuitBreakerService(),
-            new NioGroupFactory(settings, logger)
+            new NioGroupFactory(settings, logger),
+            NoopTracer.INSTANCE
         ) {
 
             @Override

server/src/main/java/org/opensearch/common/network/NetworkModule.java

@@ -174,7 +174,8 @@ public NetworkModule(
                 pageCacheRecycler,
                 circuitBreakerService,
                 namedWriteableRegistry,
-                networkService
+                networkService,
+                tracer
             );
             for (Map.Entry<String, Supplier<Transport>> entry : transportFactory.entrySet()) {
                 registerTransport(entry.getKey(), entry.getValue());

server/src/main/java/org/opensearch/plugins/NetworkPlugin.java

@@ -83,7 +83,8 @@ default Map<String, Supplier<Transport>> getTransports(
         PageCacheRecycler pageCacheRecycler,
         CircuitBreakerService circuitBreakerService,
         NamedWriteableRegistry namedWriteableRegistry,
-        NetworkService networkService
+        NetworkService networkService,
+        Tracer tracer
     ) {
         return Collections.emptyMap();
     }

server/src/main/java/org/opensearch/telemetry/tracing/AttributeNames.java

@@ -60,6 +60,11 @@ private AttributeNames() {
      */
     public static final String TRANSPORT_TARGET_HOST = "target_host";
 
+    /**
+     * Transport Service send request local host.
+     */
+    public static final String TRANSPORT_HOST = "host";
+
     /**
      * Action Name.
      */

server/src/main/java/org/opensearch/telemetry/tracing/SpanBuilder.java

@@ -13,6 +13,7 @@
 import org.opensearch.http.HttpRequest;
 import org.opensearch.rest.RestRequest;
 import org.opensearch.telemetry.tracing.attributes.Attributes;
+import org.opensearch.transport.TcpChannel;
 import org.opensearch.transport.Transport;
 
 import java.util.Arrays;
@@ -127,4 +128,26 @@ private static Attributes buildSpanAttributes(String action, Transport.Connectio
         return attributes;
     }
 
+    /**
+     * Creates {@link SpanCreationContext} from Inbound Handler.
+     * @param action action.
+     * @param tcpChannel tcp channel.
+     * @return context
+     */
+    public static SpanCreationContext from(String action, TcpChannel tcpChannel) {
+        return SpanCreationContext.server().name(createSpanName(action, tcpChannel)).attributes(buildSpanAttributes(action, tcpChannel));
+    }
+
+    private static String createSpanName(String action, TcpChannel tcpChannel) {
+        return action + SEPARATOR + (tcpChannel.getRemoteAddress() != null
+            ? tcpChannel.getRemoteAddress().getHostString()
+            : tcpChannel.getLocalAddress().getHostString());
+    }
+
+    private static Attributes buildSpanAttributes(String action, TcpChannel tcpChannel) {
+        Attributes attributes = Attributes.create().addAttribute(AttributeNames.TRANSPORT_ACTION, action);
+        attributes.addAttribute(AttributeNames.TRANSPORT_HOST, tcpChannel.getLocalAddress().getHostString());
+        return attributes;
+    }
+
 }

server/src/main/java/org/opensearch/telemetry/tracing/channels/TraceableTransportChannel.java

@@ -0,0 +1,110 @@
+/*
+ * 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.channels;
+
+import org.opensearch.Version;
+import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.transport.TransportResponse;
+import org.opensearch.telemetry.tracing.Span;
+import org.opensearch.telemetry.tracing.SpanScope;
+import org.opensearch.telemetry.tracing.Tracer;
+import org.opensearch.transport.TcpChannel;
+import org.opensearch.transport.TransportChannel;
+
+import java.io.IOException;
+
+/**
+ * Tracer wrapped {@link TransportChannel}
+ */
+public class TraceableTransportChannel implements TransportChannel {
+
+    private final TransportChannel delegate;
+    private final Span span;
+    private final Tracer tracer;
+
+    /**
+     * Constructor.
+     * @param delegate delegate
+     * @param span span
+     * @param tracer tracer
+     */
+    public TraceableTransportChannel(TransportChannel delegate, Span span, Tracer tracer, TcpChannel tcpChannel) {
+        this.delegate = delegate;
+        this.span = span;
+        this.tracer = tracer;
+    }
+
+    /**
+     * Factory method.
+     *
+     * @param delegate delegate
+     * @param span     span
+     * @param tracer tracer
+     * @return transport channel
+     */
+    public static TransportChannel create(TransportChannel delegate, final Span span, final Tracer tracer, final TcpChannel tcpChannel) {
+        if (FeatureFlags.isEnabled(FeatureFlags.TELEMETRY) == true) {
+            tcpChannel.addCloseListener(new ActionListener<Void>() {
+                @Override
+                public void onResponse(Void unused) {
+                    onFailure(null);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    span.addEvent("The TransportChannel was closed without sending the response");
+                    span.setError(e);
+                    span.endSpan();
+                }
+            });
+
+            return new TraceableTransportChannel(delegate, span, tracer, tcpChannel);
+        } else {
+            return delegate;
+        }
+    }
+
+    @Override
+    public String getProfileName() {
+        return delegate.getProfileName();
+    }
+
+    @Override
+    public String getChannelType() {
+        return delegate.getChannelType();
+    }
+
+    @Override
+    public void sendResponse(TransportResponse response) throws IOException {
+        try (SpanScope scope = tracer.withSpanInScope(span)) {
+            delegate.sendResponse(response);
+        } catch (final IOException ex) {
+            span.setError(ex);
+            throw ex;
+        } finally {
+            span.endSpan();
+        }
+    }
+
+    @Override
+    public void sendResponse(Exception exception) throws IOException {
+        try (SpanScope scope = tracer.withSpanInScope(span)) {
+            delegate.sendResponse(exception);
+        } finally {
+            span.setError(exception);
+            span.endSpan();
+        }
+    }
+
+    @Override
+    public Version getVersion() {
+        return delegate.getVersion();
+    }
+}