HADOOP-15566 Opentelemetry changes using java agent

hadoop-common-project/hadoop-common/pom.xml

@@ -424,6 +424,11 @@
       <artifactId>lz4-java</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-api</artifactId>
+      <version>${opentelemetry.version}</version>
+    </dependency>
   </dependencies>
 
   <build>

hadoop-common-project/hadoop-common/src/main/conf/hadoop-env.sh

@@ -432,3 +432,13 @@ export HADOOP_OS_TYPE=${HADOOP_OS_TYPE:-$(uname -s)}
 # By default, Hadoop uses jsvc which needs to know to launch a
 # server jvm.
 # export HADOOP_REGISTRYDNS_SECURE_EXTRA_OPTS="-jvm server"
+
+###
+# Opentelemetry Trace Exporters Configuration
+###
+# export OPENTELEMETRY_JAVAAGENT_PATH="$(find $HADOOP_HOME/share/hadoop/tools/lib/ -name opentelemetry-javaagent*)"
+# export HADOOP_TRACE_OPTS="-javaagent:$OPENTELEMETRY_JAVAAGENT_PATH -Dotel.traces.exporter=jaeger -Dotel.metrics.exporter=none"
+# export HDFS_NAMENODE_OPTS="$HDFS_NAMENODE_OPTS $HADOOP_TRACE_OPTS -Dotel.resource.attributes=service.name=hdfs-namenode"
+# export HDFS_DATANODE_OPTS="$HDFS_DATANODE_OPTS $HADOOP_TRACE_OPTS -Dotel.resource.attributes=service.name=hdfs-datanode"
+# export HDFS_SECONDARYNAMENODE_OPTS="$HDFS_SECONDARYNAMENODE_OPTS $HADOOP_TRACE_OPTS -Dotel.resource.attributes=service.name=hdfs-secondarynamenode"
+# export HADOOP_CLIENT_OPTS="$HADOOP_CLIENT_OPTS $HADOOP_TRACE_OPTS -Dotel.resource.attributes=service.name=hadoop-client"

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FsShell.java

@@ -301,7 +301,7 @@ private TableListing createOptionTableListing() {
   public int run(String[] argv) {
     // initialize FsShell
     init();
-    Tracer tracer = new Tracer.Builder("FsShell").
+    Tracer tracer = new Tracer.Builder().
         conf(TraceUtils.wrapHadoopConf(SHELL_HTRACE_PREFIX, getConf())).
         build();
     int exitCode = -1;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FsTracer.java

@@ -38,7 +38,7 @@ public final class FsTracer {
 
   public static synchronized Tracer get(Configuration conf) {
     if (instance == null) {
-      instance = new Tracer.Builder("FSClient").
+      instance = new Tracer.Builder().
           conf(TraceUtils.wrapHadoopConf(CommonConfigurationKeys.
               FS_CLIENT_HTRACE_PREFIX, conf)).
           build();

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -2901,15 +2901,15 @@ private void processRpcRequest(RpcRequestHeaderProto header,
       Span span = null;
       if (header.hasTraceInfo()) {
         RPCTraceInfoProto traceInfoProto = header.getTraceInfo();
-        if (traceInfoProto.hasSpanContext()) {
+        if (traceInfoProto.getOpenSpanContextMap() != null) {
           if (tracer == null) {
             setTracer(Tracer.curThreadTracer());
           }
           if (tracer != null) {
             // If the incoming RPC included tracing info, always continue the
             // trace
-            SpanContext spanCtx = TraceUtils.byteStringToSpanContext(
-                traceInfoProto.getSpanContext());
+            SpanContext spanCtx = TraceUtils.mapToSpanContext(
+                traceInfoProto.getOpenSpanContextMap());
             if (spanCtx != null) {
               span = tracer.newSpan(
                   RpcClientUtil.toTraceName(rpcRequest.toString()), spanCtx);

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/tracing/Span.java

@@ -17,28 +17,62 @@
  */
 package org.apache.hadoop.tracing;
 
+import io.opentelemetry.context.Scope;
+
 import java.io.Closeable;
 
+/***
+ * This class is a wrapper class on top of opentelemetry Span class
+ * avoiding direct dependency on opentelemetry API.
+ */
 public class Span implements Closeable {
-
+  private io.opentelemetry.api.trace.Span openSpan;
   public Span() {
   }
 
+  public Span(io.opentelemetry.api.trace.Span openSpan){
+    this.openSpan = openSpan;
+  }
+
   public Span addKVAnnotation(String key, String value) {
+    if(openSpan != null){
+      openSpan.setAttribute(key, value);
+    }
     return this;
   }
 
   public Span addTimelineAnnotation(String msg) {
+    if(openSpan != null){
+      openSpan.addEvent(msg);
+    }
     return this;
   }
 
   public SpanContext getContext() {
+    if(openSpan != null){
+      return  new SpanContext(openSpan.getSpanContext());
+    }
     return null;
   }
 
   public void finish() {
+    close();
   }
 
   public void close() {
+    if(openSpan != null){
+      openSpan.end();
+    }
+  }
+
+  /***
+   * This method activates the current span on the current thread
+   * @return the scope for the current span
+   */
+  public Scope makeCurrent() {
+    if(openSpan != null){
+      return openSpan.makeCurrent();
+    }
+    return null;
   }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/tracing/SpanContext.java

@@ -18,15 +18,76 @@
 package org.apache.hadoop.tracing;
 
 import java.io.Closeable;
+import java.util.HashMap;
+import java.util.Map;
+
+import io.opentelemetry.api.trace.TraceFlags;
+import io.opentelemetry.api.trace.TraceState;
+import io.opentelemetry.api.trace.TraceStateBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Wrapper class for SpanContext to avoid using OpenTracing/OpenTelemetry
  * SpanContext class directly for better separation.
  */
-public class SpanContext implements Closeable {
-  public SpanContext() {
+public class SpanContext implements Closeable  {
+  public static final Logger LOG = LoggerFactory.getLogger(SpanContext.class.getName());
+  private static final String TRACE_ID = "TRACE_ID";
+  private static final String SPAN_ID = "SPAN_ID";
+  private static final String TRACE_FLAGS = "TRACE_FLAGS";
+
+
+  private io.opentelemetry.api.trace.SpanContext openSpanContext;
+  public SpanContext(io.opentelemetry.api.trace.SpanContext openSpanContext) {
+    this.openSpanContext = openSpanContext;
   }
 
   public void close() {
+
+  }
+
+  public Map<String, String> getKVSpanContext(){
+    if(openSpanContext != null){
+      //TODO: may we should move this to Proto
+      Map<String, String> kvMap = new HashMap<>();
+      kvMap.put(TRACE_ID, openSpanContext.getTraceId());
+      kvMap.put(SPAN_ID, openSpanContext.getSpanId());
+      kvMap.put(TRACE_FLAGS, openSpanContext.getTraceFlags().asHex());
+      kvMap.putAll(openSpanContext.getTraceState().asMap());
+      return kvMap;
+    }
+    return null;
+  }
+
+  static SpanContext buildFromKVMap(Map<String, String> kvMap){
+    try{
+      String traceId = kvMap.get(TRACE_ID);
+      String spanId = kvMap.get(SPAN_ID);
+      String traceFlagsHex = kvMap.get(TRACE_FLAGS);
+      if(traceId == null || spanId == null || traceFlagsHex == null){
+        return null;
+      }
+      TraceFlags traceFlags = TraceFlags.fromHex(traceFlagsHex, 0);
+      TraceStateBuilder traceStateBuilder = TraceState.builder();
+      for(Map.Entry<String, String> keyValue: kvMap.entrySet()){
+        if(keyValue.getKey().equals(TRACE_ID) || keyValue.getKey().equals(SPAN_ID) || keyValue.getKey().equals(TRACE_FLAGS)){
+          continue;
+        }
+        traceStateBuilder.put(keyValue.getKey(), keyValue.getValue());
+      }
+      TraceState traceState = traceStateBuilder.build();
+      io.opentelemetry.api.trace.SpanContext spanContext = io.opentelemetry.api.trace.SpanContext.createFromRemoteParent(traceId, spanId, traceFlags, traceState );
+      return new SpanContext(spanContext);
+    } catch (Exception e){
+      LOG.error("Error in processing remote context :", e);
+      return null;
+    }
+
+
+  }
+
+  public io.opentelemetry.api.trace.SpanContext getOpenSpanContext() {
+      return openSpanContext;
   }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/tracing/TraceScope.java

@@ -17,23 +17,32 @@
  */
 package org.apache.hadoop.tracing;
 
+import io.opentelemetry.context.Scope;
+
 import java.io.Closeable;
 
 public class TraceScope implements Closeable {
   Span span;
-
+  Scope openScope;
   public TraceScope(Span span) {
     this.span = span;
+    if(span != null){
+      this.openScope = span.makeCurrent();
+    }
   }
 
   // Add tag to the span
   public void addKVAnnotation(String key, String value) {
+    span.addKVAnnotation(key, value);
   }
 
   public void addKVAnnotation(String key, Number value) {
+    span.addKVAnnotation(key, value.toString());
   }
 
+
   public void addTimelineAnnotation(String msg) {
+    span.addTimelineAnnotation(msg);
   }
 
   public Span span() {
@@ -51,6 +60,9 @@ public void detach() {
   }
 
   public void close() {
+    if(openScope != null){
+      openScope.close();
+    }
     if (span != null) {
       span.close();
     }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/tracing/TraceUtils.java

@@ -19,29 +19,35 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.thirdparty.protobuf.ByteString;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
 
 /**
  * This class provides utility functions for tracing.
  */
 @InterfaceAudience.Private
 public class TraceUtils {
+  public static final Logger LOG = LoggerFactory.getLogger(TraceUtils.class.getName());
   static final String DEFAULT_HADOOP_TRACE_PREFIX = "hadoop.htrace.";
 
-  public static TraceConfiguration wrapHadoopConf(final String prefix,
-      final Configuration conf) {
+  public static TraceConfiguration wrapHadoopConf(final String prefix, final Configuration conf) {
     return null;
   }
 
   public static Tracer createAndRegisterTracer(String name) {
     return null;
   }
 
-  public static SpanContext byteStringToSpanContext(ByteString byteString) {
-    return null;
+  public static SpanContext mapToSpanContext(Map<String, String> kvMap) {
+    return SpanContext.buildFromKVMap(kvMap);
   }
 
-  public static ByteString spanContextToByteString(SpanContext context) {
-    return null;
+  public static Map<String, String> spanContextToMap(SpanContext context) {
+    if (context == null) {
+      return null;
+    }
+    return context.getKVSpanContext();
   }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/tracing/Tracer.java

@@ -17,22 +17,24 @@
  */
 package org.apache.hadoop.tracing;
 
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.context.Context;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 /**
  * No-Op Tracer (for now) to remove HTrace without changing too many files.
  */
 public class Tracer {
+  public static final Logger LOG = LoggerFactory.getLogger(Tracer.class.getName());
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.contrib.hadoop";
   // Singleton
-  private static final Tracer globalTracer = null;
-  private final NullTraceScope nullTraceScope;
-  private final String name;
+  private static Tracer globalTracer = null;
+  io.opentelemetry.api.trace.Tracer OTelTracer = GlobalOpenTelemetry.getTracer(INSTRUMENTATION_NAME);
 
   public final static String SPAN_RECEIVER_CLASSES_KEY =
       "span.receiver.classes";
 
-  public Tracer(String name) {
-    this.name = name;
-    nullTraceScope = NullTraceScope.INSTANCE;
-  }
+  private Tracer() {}
 
   // Keeping this function at the moment for HTrace compatiblity,
   // in fact all threads share a single global tracer for OpenTracing.
@@ -45,53 +47,52 @@ public static Tracer curThreadTracer() {
    * @return org.apache.hadoop.tracing.Span
    */
   public static Span getCurrentSpan() {
-    return null;
+    io.opentelemetry.api.trace.Span span = io.opentelemetry.api.trace.Span.current();
+    return span.getSpanContext().isValid()? new Span(span): null;
   }
 
   public TraceScope newScope(String description) {
-    return nullTraceScope;
+    Span span = new Span(OTelTracer.spanBuilder(description).startSpan());
+    return new TraceScope(span);
   }
 
   public Span newSpan(String description, SpanContext spanCtx) {
-    return new Span();
+    io.opentelemetry.api.trace.Span parentSpan = io.opentelemetry.api.trace.Span.wrap(spanCtx.getOpenSpanContext());
+    io.opentelemetry.api.trace.Span span = OTelTracer.spanBuilder(description).setParent(Context.current().with(parentSpan)).startSpan();
+    return new Span(span);
   }
 
   public TraceScope newScope(String description, SpanContext spanCtx) {
-    return nullTraceScope;
+    if(spanCtx == null){
+      return new TraceScope(new Span(io.opentelemetry.api.trace.Span.getInvalid()));
+    }
+    return new TraceScope(newSpan(description, spanCtx));
   }
 
   public TraceScope newScope(String description, SpanContext spanCtx,
       boolean finishSpanOnClose) {
-    return nullTraceScope;
+    return new TraceScope(newSpan(description, spanCtx));
   }
 
   public TraceScope activateSpan(Span span) {
-    return nullTraceScope;
+    return new TraceScope(span);
   }
 
   public void close() {
   }
 
-  public String getName() {
-    return name;
-  }
-
+
   public static class Builder {
-    static Tracer globalTracer;
-    private String name;
+    static Tracer globalTracer = new Tracer();
 
-    public Builder(final String name) {
-      this.name = name;
+    public Builder() {
     }
 
     public Builder conf(TraceConfiguration conf) {
       return this;
     }
 
-    public Tracer build() {
-      if (globalTracer == null) {
-        globalTracer = new Tracer(name);
-      }
+    public synchronized Tracer build() {
       return globalTracer;
     }
   }