HDDS-3399. Update JaegerTracing

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java

@@ -35,6 +35,7 @@
 
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.function.SupplierWithIOException;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
@@ -57,6 +58,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import io.opentracing.Scope;
+import io.opentracing.Span;
 import io.opentracing.util.GlobalTracer;
 import org.apache.ratis.thirdparty.io.grpc.ManagedChannel;
 import org.apache.ratis.thirdparty.io.grpc.Status;
@@ -265,14 +267,18 @@ public ContainerCommandResponseProto sendCommand(
   private XceiverClientReply sendCommandWithTraceIDAndRetry(
       ContainerCommandRequestProto request, List<CheckedBiFunction> validators)
       throws IOException {
-    try (Scope scope = GlobalTracer.get()
-        .buildSpan("XceiverClientGrpc." + request.getCmdType().name())
-        .startActive(true)) {
-      ContainerCommandRequestProto finalPayload =
-          ContainerCommandRequestProto.newBuilder(request)
-              .setTraceID(TracingUtil.exportCurrentSpan()).build();
-      return sendCommandWithRetry(finalPayload, validators);
-    }
+
+    String spanName = "XceiverClientGrpc." + request.getCmdType().name();
+
+    return TracingUtil.executeInNewSpan(spanName,
+        (SupplierWithIOException<XceiverClientReply>) () -> {
+
+          ContainerCommandRequestProto finalPayload =
+              ContainerCommandRequestProto.newBuilder(request)
+                  .setTraceID(TracingUtil.exportCurrentSpan()).build();
+          return sendCommandWithRetry(finalPayload, validators);
+
+        });
   }
 
   private XceiverClientReply sendCommandWithRetry(
@@ -387,9 +393,11 @@ private XceiverClientReply sendCommandWithRetry(
   public XceiverClientReply sendCommandAsync(
       ContainerCommandRequestProto request)
       throws IOException, ExecutionException, InterruptedException {
-    try (Scope scope = GlobalTracer.get()
-        .buildSpan("XceiverClientGrpc." + request.getCmdType().name())
-        .startActive(true)) {
+
+    Span span = GlobalTracer.get()
+        .buildSpan("XceiverClientGrpc." + request.getCmdType().name()).start();
+
+    try (Scope scope = GlobalTracer.get().activateSpan(span)) {
 
       ContainerCommandRequestProto finalPayload =
           ContainerCommandRequestProto.newBuilder(request)
@@ -405,6 +413,9 @@ public XceiverClientReply sendCommandAsync(
         asyncReply.getResponse().get();
       }
       return asyncReply;
+
+    } finally {
+      span.finish();
     }
   }
 

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java

@@ -51,8 +51,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import io.opentracing.Scope;
-import io.opentracing.util.GlobalTracer;
+import com.google.common.base.Supplier;
 import org.apache.ratis.client.RaftClient;
 import org.apache.ratis.grpc.GrpcTlsConfig;
 import org.apache.ratis.proto.RaftProtos;
@@ -208,24 +207,27 @@ public ConcurrentMap<UUID, Long> getCommitInfoMap() {
 
   private CompletableFuture<RaftClientReply> sendRequestAsync(
       ContainerCommandRequestProto request) {
-    try (Scope scope = GlobalTracer.get()
-        .buildSpan("XceiverClientRatis." + request.getCmdType().name())
-        .startActive(true)) {
-      final ContainerCommandRequestMessage message
-          = ContainerCommandRequestMessage.toMessage(
+    return TracingUtil.executeInNewSpan(
+        "XceiverClientRatis." + request.getCmdType().name(),
+        (Supplier<CompletableFuture<RaftClientReply>>) () -> {
+          final ContainerCommandRequestMessage message
+              = ContainerCommandRequestMessage.toMessage(
               request, TracingUtil.exportCurrentSpan());
-      if (HddsUtils.isReadOnly(request)) {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("sendCommandAsync ReadOnly {}", message);
-        }
-        return getClient().sendReadOnlyAsync(message);
-      } else {
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("sendCommandAsync {}", message);
+          if (HddsUtils.isReadOnly(request)) {
+            if (LOG.isDebugEnabled()) {
+              LOG.debug("sendCommandAsync ReadOnly {}", message);
+            }
+            return getClient().sendReadOnlyAsync(message);
+          } else {
+            if (LOG.isDebugEnabled()) {
+              LOG.debug("sendCommandAsync {}", message);
+            }
+            return getClient().sendAsync(message);
+          }
+
         }
-        return getClient().sendAsync(message);
-      }
-    }
+
+    );
   }
 
   // gets the minimum log index replicated to all servers

hadoop-hdds/common/pom.xml

@@ -177,7 +177,7 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <dependency>
       <groupId>io.opentracing</groupId>
       <artifactId>opentracing-util</artifactId>
-      <version>0.31.0</version>
+      <version>0.33.0</version>
     </dependency>
     <dependency>
       <groupId>org.yaml</groupId>

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/function/SupplierWithIOException.java

@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdds.function;
+
+import java.io.IOException;
+
+/**
+ * Functional interface like java.util.function.Supplier but with
+ * checked exception.
+ */
+@FunctionalInterface
+public interface SupplierWithIOException<T> {
+
+  /**
+   * Return the given output..
+   *
+   * @return the function result
+   */
+  T get() throws IOException;
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/GrpcServerInterceptor.java

@@ -18,6 +18,8 @@
 package org.apache.hadoop.hdds.tracing;
 
 import io.opentracing.Scope;
+import io.opentracing.Span;
+import io.opentracing.util.GlobalTracer;
 import org.apache.ratis.thirdparty.io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener;
 import org.apache.ratis.thirdparty.io.grpc.Metadata;
 import org.apache.ratis.thirdparty.io.grpc.ServerCall;
@@ -39,11 +41,14 @@ public <ReqT, RespT> Listener<ReqT> interceptCall(
         next.startCall(call, headers)) {
       @Override
       public void onMessage(ReqT message) {
-        try (Scope scope = TracingUtil
-            .importAndCreateScope(
+        Span span = TracingUtil
+            .importAndCreateSpan(
                 call.getMethodDescriptor().getFullMethodName(),
-                headers.get(GrpcClientInterceptor.TRACING_HEADER))) {
+                headers.get(GrpcClientInterceptor.TRACING_HEADER));
+        try (Scope scope = GlobalTracer.get().activateSpan(span)) {
           super.onMessage(message);
+        } finally {
+          span.finish();
         }
       }
     };

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TraceAllMethod.java

@@ -25,6 +25,7 @@
 import java.util.Map.Entry;
 
 import io.opentracing.Scope;
+import io.opentracing.Span;
 import io.opentracing.util.GlobalTracer;
 
 /**
@@ -64,9 +65,10 @@ public Object invoke(Object proxy, Method method, Object[] args)
         method.getName());
     }
 
-    try (Scope scope = GlobalTracer.get().buildSpan(
+    Span span = GlobalTracer.get().buildSpan(
         name + "." + method.getName())
-        .startActive(true)) {
+        .start();
+    try (Scope scope = GlobalTracer.get().activateSpan(span)) {
       try {
         return delegateMethod.invoke(delegate, args);
       } catch (Exception ex) {
@@ -75,6 +77,8 @@ public Object invoke(Object proxy, Method method, Object[] args)
         } else {
           throw ex;
         }
+      } finally {
+        span.finish();
       }
     }
   }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TracingUtil.java

@@ -17,7 +17,13 @@
  */
 package org.apache.hadoop.hdds.tracing;
 
+import java.io.IOException;
 import java.lang.reflect.Proxy;
+import java.util.function.Supplier;
+
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.function.SupplierWithIOException;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 
 import io.jaegertracing.Configuration;
 import io.jaegertracing.internal.JaegerTracer;
@@ -27,9 +33,6 @@
 import io.opentracing.Tracer;
 import io.opentracing.util.GlobalTracer;
 
-import org.apache.hadoop.hdds.conf.ConfigurationSource;
-import org.apache.hadoop.hdds.scm.ScmConfigKeys;
-
 /**
  * Utility class to collect all the tracing helper methods.
  */
@@ -86,11 +89,11 @@ public static String exportSpan(Span span) {
    *
    * @return OpenTracing scope.
    */
-  public static Scope importAndCreateScope(String name, String encodedParent) {
+  public static Span importAndCreateSpan(String name, String encodedParent) {
     Tracer tracer = GlobalTracer.get();
     return tracer.buildSpan(name)
         .asChildOf(extractParent(encodedParent, tracer))
-        .startActive(true);
+        .start();
   }
 
   private static SpanContext extractParent(String parent, Tracer tracer) {
@@ -130,8 +133,57 @@ public static <T> T createProxy(
   private static boolean isTracingEnabled(
       ConfigurationSource conf) {
     return conf.getBoolean(
-          ScmConfigKeys.HDDS_TRACING_ENABLED,
-          ScmConfigKeys.HDDS_TRACING_ENABLED_DEFAULT);
+        ScmConfigKeys.HDDS_TRACING_ENABLED,
+        ScmConfigKeys.HDDS_TRACING_ENABLED_DEFAULT);
   }
 
+  /**
+   * Execute a new function inside an activated span.
+   */
+  public static <R> R executeInNewSpan(String spanName,
+      SupplierWithIOException<R> supplier)
+      throws IOException {
+    Span span = GlobalTracer.get()
+        .buildSpan(spanName).start();
+    try (Scope scope = GlobalTracer.get().activateSpan(span)) {
+      return supplier.get();
+    } catch (Exception ex) {
+      span.setTag("failed", true);
+      throw ex;
+    } finally {
+      span.finish();
+    }
+  }
+
+  /**
+   * Execute a new function inside an activated span.
+   */
+  public static <R> R executeInNewSpan(String spanName,
+      Supplier<R> supplier) {
+    Span span = GlobalTracer.get()
+        .buildSpan(spanName).start();
+    try (Scope scope = GlobalTracer.get().activateSpan(span)) {
+      return supplier.get();
+    } catch (Exception ex) {
+      span.setTag("failed", true);
+      throw ex;
+    } finally {
+      span.finish();
+    }
+  }
+
+  /**
+   * Create an active span with auto-close at finish.
+   * <p>
+   * This is a simplified way to use span as there is no way to add any tag
+   * in case of Exceptions.
+   */
+  public static AutoCloseable createActivatedSpan(String spanName) {
+    Span span = GlobalTracer.get().buildSpan(spanName).start();
+    Scope scope = GlobalTracer.get().activateSpan(span);
+    return () -> {
+      scope.close();
+      span.finish();
+    };
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java

@@ -61,6 +61,8 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import io.opentracing.Scope;
+import io.opentracing.Span;
+import io.opentracing.util.GlobalTracer;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.malformedRequest;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.unsupportedRequest;
 import org.apache.ratis.thirdparty.com.google.protobuf.ProtocolMessageEnum;
@@ -157,10 +159,12 @@ public ContainerCommandResponseProto dispatch(
       ContainerCommandRequestProto msg, DispatcherContext dispatcherContext) {
     String spanName = "HddsDispatcher." + msg.getCmdType().name();
     long startTime = System.nanoTime();
-    try (Scope scope = TracingUtil
-        .importAndCreateScope(spanName, msg.getTraceID())) {
+    Span span = TracingUtil
+        .importAndCreateSpan(spanName, msg.getTraceID());
+    try (Scope scope = GlobalTracer.get().activateSpan(span)) {
       return dispatchRequest(msg, dispatcherContext);
     } finally {
+      span.finish();
       protocolMetrics
           .increment(msg.getCmdType(), System.nanoTime() - startTime);
     }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java

@@ -43,6 +43,8 @@
 
 import com.google.common.base.Preconditions;
 import io.opentracing.Scope;
+import io.opentracing.Span;
+import io.opentracing.util.GlobalTracer;
 import org.apache.ratis.thirdparty.io.grpc.BindableService;
 import org.apache.ratis.thirdparty.io.grpc.Server;
 import org.apache.ratis.thirdparty.io.grpc.ServerBuilder;
@@ -170,16 +172,19 @@ public void stop() {
   @Override
   public void submitRequest(ContainerCommandRequestProto request,
       HddsProtos.PipelineID pipelineID) throws IOException {
-    try (Scope scope = TracingUtil
-        .importAndCreateScope(
+    Span span = TracingUtil
+        .importAndCreateSpan(
             "XceiverServerGrpc." + request.getCmdType().name(),
-            request.getTraceID())) {
+            request.getTraceID());
+    try (Scope scope = GlobalTracer.get().activateSpan(span)) {
       ContainerProtos.ContainerCommandResponseProto response =
           storageContainer.dispatch(request, null);
       if (response.getResult() != ContainerProtos.Result.SUCCESS) {
         throw new StorageContainerException(response.getMessage(),
             response.getResult());
       }
+    } finally {
+      span.finish();
     }
   }