Add OpenTelemetry tracing (#151)

Signed-off-by: Antoine Toulme <antoine@lunar-ocean.com>

Author: atoulme

README.md

@@ -413,5 +413,14 @@ by starting command have the -d parameter.
 
 Upload full logs to the JIRA not just where the issue occurred if possible
 
+### Tracing
+
+The SDK is set up to trace all gRPC communications:
+* All outgoing messages bear trace metadata, allowing correlation with peers.
+* Each request/response is computed as a span.
+* If you use OpenTelemetry in your program, the gRPC span will correlate using the thread local context to the parent context.
+
+The SDK accepts all environment variables as described in the [OpenTelemetry specification](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/sdk-environment-variables.md).
+
 
 <a rel="license" href="http://creativecommons.org/licenses/by/4.0/"><img alt="Creative Commons License" style="border-width:0" src="https://i.creativecommons.org/l/by/4.0/88x31.png" /></a><br />This work is licensed under a <a rel="license" href="http://creativecommons.org/licenses/by/4.0/">Creative Commons Attribution 4.0 International License</a>.

dependencies-suppressions.xml

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<suppressions xmlns="https://jeremylong.github.io/DependencyCheck/dependency-suppression.1.3.xsd">
+    <suppress>
+        <notes><![CDATA[
+   file name: opentelemetry-grpc-1.6-1.5.1-alpha.jar
+   ]]></notes>
+        <packageUrl regex="true">^pkg:maven/io\.opentelemetry\.instrumentation/opentelemetry\-grpc\-1\.6@.*$</packageUrl>
+        <cve>CVE-2020-7768</cve>
+    </suppress>
+</suppressions>

pom.xml

@@ -68,6 +68,18 @@
         </plugins>
     </reporting>
 
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>io.opentelemetry</groupId>
+                <artifactId>opentelemetry-bom</artifactId>
+                <version>1.5.0</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
     <dependencies>
         <dependency>
             <groupId>org.mockito</groupId>
@@ -220,6 +232,36 @@
             <version>1.3.2</version>
         </dependency>
 
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk-trace</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-otlp</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-extension-trace-propagators</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry.instrumentation</groupId>
+            <artifactId>opentelemetry-grpc-1.6</artifactId>
+            <version>1.5.1-alpha</version>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
+            <version>1.5.0-alpha</version>
+        </dependency>
     </dependencies>
 
 
@@ -576,8 +618,27 @@
                     </archive>
                 </configuration>
             </plugin>
-
-
+            <plugin>
+                <groupId>org.owasp</groupId>
+                <artifactId>dependency-check-maven</artifactId>
+                <version>6.2.2</version>
+                <configuration>
+                    <skipProvidedScope>true</skipProvidedScope>
+                    <skipTestScope>true</skipTestScope>
+                    <skipSystemScope>true</skipSystemScope>
+                    <failBuildOnCVSS>7</failBuildOnCVSS>
+                    <suppressionFiles>
+                        <suppressionFile>dependencies-suppressions.xml</suppressionFile>
+                    </suppressionFiles>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>check</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
     <distributionManagement>

src/main/java/org/hyperledger/fabric/sdk/Endpoint.java

@@ -36,13 +36,16 @@
 import javax.net.ssl.SSLException;
 
 import com.google.common.collect.ImmutableMap;
+import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
 import io.grpc.netty.shaded.io.grpc.netty.NegotiationType;
 import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslProvider;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.grpc.v1_6.GrpcTracing;
 import org.apache.commons.codec.binary.Hex;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -67,6 +70,8 @@ class Endpoint {
 
     private static final String SSLPROVIDER = Config.getConfig().getDefaultSSLProvider();
     private static final String SSLNEGOTIATION = Config.getConfig().getDefaultSSLNegotiationType();
+    private static final OpenTelemetry openTelemetry = Config.getConfig().getOpenTelemetry();
+    private static final GrpcTracing grpcTracing = GrpcTracing.create(openTelemetry);
 
     private final String addr;
     private final int port;
@@ -236,13 +241,14 @@ class Endpoint {
         }
 
         try {
+            ClientInterceptor clientInterceptor = grpcTracing.newClientInterceptor();
             if (protocol.equalsIgnoreCase("grpc")) {
-                this.channelBuilder = NettyChannelBuilder.forAddress(addr, port).negotiationType(NegotiationType.PLAINTEXT);
+                this.channelBuilder = NettyChannelBuilder.forAddress(addr, port).negotiationType(NegotiationType.PLAINTEXT).intercept(clientInterceptor);
                 addNettyBuilderProps(channelBuilder, properties);
             } else if (protocol.equalsIgnoreCase("grpcs")) {
                 if (pemBytes == null) {
                     // use root certificate
-                    this.channelBuilder = NettyChannelBuilder.forAddress(addr, port);
+                    this.channelBuilder = NettyChannelBuilder.forAddress(addr, port).intercept(clientInterceptor);
                     addNettyBuilderProps(channelBuilder, properties);
                 } else {
                     try {

src/main/java/org/hyperledger/fabric/sdk/helper/Config.java

@@ -23,6 +23,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.OpenTelemetrySdkBuilder;
+import io.opentelemetry.sdk.autoconfigure.OpenTelemetrySdkAutoConfiguration;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.log4j.Level;
@@ -111,6 +116,7 @@ public class Config {
     private static Config config;
     private static final Properties sdkProperties = new Properties();
     private static final AtomicLong count = new AtomicLong(0);
+    private final OpenTelemetrySdk openTelemetry;
 
     //Provides a unique id for logging to identify a specific instance.
     public String getNextID() {
@@ -241,6 +247,7 @@ private Config() {
 
             }
 
+            openTelemetry = OpenTelemetrySdkAutoConfiguration.initialize(false);
         }
 
     }
@@ -348,6 +355,10 @@ public String getDefaultSSLNegotiationType() {
 
     }
 
+    public OpenTelemetry getOpenTelemetry() {
+        return openTelemetry;
+    }
+
     private Map<Integer, String> curveMapping = null;
 
     /**

src/test/java/org/hyperledger/fabric/sdkintegration/End2endLifecycleIT.java

@@ -10,13 +10,15 @@
 import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
@@ -28,7 +30,16 @@
 import java.util.concurrent.TimeoutException;
 import java.util.function.Predicate;
 
+import com.google.common.io.Closer;
 import com.google.protobuf.InvalidProtocolBufferException;
+import io.grpc.Server;
+import io.grpc.Status;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import io.grpc.stub.StreamObserver;
+import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
+import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceResponse;
+import io.opentelemetry.proto.collector.trace.v1.TraceServiceGrpc;
+import io.opentelemetry.proto.trace.v1.ResourceSpans;
 import org.hyperledger.fabric.sdk.BlockEvent.TransactionEvent;
 import org.hyperledger.fabric.sdk.ChaincodeCollectionConfiguration;
 import org.hyperledger.fabric.sdk.ChaincodeResponse;
@@ -116,6 +127,38 @@ public class End2endLifecycleIT {
         TX_EXPECTED.put("writeset1", "Missing writeset for channel bar block 1");
     }
 
+    private static final class FakeCollector extends TraceServiceGrpc.TraceServiceImplBase {
+        private final List<ResourceSpans> receivedSpans = new ArrayList<>();
+        private Status returnedStatus = Status.OK;
+
+        @Override
+        public void export(
+                final ExportTraceServiceRequest request,
+                final StreamObserver<ExportTraceServiceResponse> responseObserver) {
+            receivedSpans.addAll(request.getResourceSpansList());
+            responseObserver.onNext(ExportTraceServiceResponse.newBuilder().build());
+            if (!returnedStatus.isOk()) {
+                if (returnedStatus.getCode() == Status.Code.DEADLINE_EXCEEDED) {
+                    // Do not call onCompleted to simulate a deadline exceeded.
+                    return;
+                }
+                responseObserver.onError(returnedStatus.asRuntimeException());
+                return;
+            }
+            responseObserver.onCompleted();
+        }
+
+        List<ResourceSpans> getReceivedSpans() {
+            return receivedSpans;
+        }
+
+        void setReturnedStatus(final Status returnedStatus) {
+            this.returnedStatus = returnedStatus;
+        }
+    }
+
+    private final Closer closer = Closer.create();
+    private final FakeCollector fakeTracesCollector = new FakeCollector();
     private final TestConfigHelper configHelper = new TestConfigHelper();
     String testName = "End2endLifecycleIT";
 
@@ -144,8 +187,15 @@ static void out(String format, Object... args) {
     @Before
     public void checkConfig() throws Exception {
         out("\n\n\nRUNNING: %s.\n", testName);
+        Server server =
+                NettyServerBuilder.forPort(4317)
+                        .addService(fakeTracesCollector)
+                        .build()
+                        .start();
+        closer.register(server::shutdownNow);
         //   configHelper.clearConfig();
         //   assertEquals(256, Config.getConfig().getSecurityLevel());
+        System.setProperty("OTEL_TRACES_SAMPLER", "always_on");
         resetConfig();
         configHelper.customizeConfig();
 
@@ -336,6 +386,8 @@ public void runFabricTest() throws Exception {
         assertNull(org1Client.getChannel(CHANNEL_NAME));
         out("\n");
 
+        assertFalse(fakeTracesCollector.receivedSpans.isEmpty());
+
         out("That's all folks!");
     }