HADOOP-15566 Opentelemetry changes using java agent

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

@@ -375,6 +375,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

@@ -428,3 +428,14 @@ 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
+# TODO: javaagent is asummed to be stored in common/lib this should be moved to seperate directory
+###
+# export OPENTELEMETRY_JAVAAGENT_PATH="$(find $HADOOP_HOME/share/hadoop/common/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/ipc/Server.java

@@ -2783,15 +2783,15 @@ private void processRpcRequest(RpcRequestHeaderProto header,
       Span span = null;
       if (header.hasTraceInfo()) {
         RPCTraceInfoProto traceInfoProto = header.getTraceInfo();
-        if (traceInfoProto.hasSpanContext()) {
+        if (traceInfoProto.getSpanContextMap() != 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.getSpanContextMap());
             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,54 @@
  */
 package org.apache.hadoop.tracing;
 
+import io.opentelemetry.context.Scope;
+
 import java.io.Closeable;
 
 public class Span implements Closeable {
-
+  private io.opentelemetry.api.trace.Span span = null;
   public Span() {
   }
 
+  public Span(io.opentelemetry.api.trace.Span span){
+    this.span = span;
+  }
+
   public Span addKVAnnotation(String key, String value) {
+    if(span != null){
+      span.setAttribute(key, value);
+    }
     return this;
   }
 
   public Span addTimelineAnnotation(String msg) {
+    if(span != null){
+      span.addEvent(msg);
+    }
     return this;
   }
 
   public SpanContext getContext() {
+    if(span != null){
+      return  new SpanContext(span.getSpanContext());
+    }
     return null;
   }
 
   public void finish() {
+    close();
   }
 
   public void close() {
+    if(span != null){
+      span.end();
+    }
+  }
+
+  public Scope makeCurrent() {
+    if(span != null){
+      return span.makeCurrent();
+    }
+    return null;
   }
 }

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

@@ -18,15 +18,75 @@
 package org.apache.hadoop.tracing;
 
 import java.io.Closeable;
+import java.util.HashMap;
+import java.util.Map;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.TraceFlags;
+import io.opentelemetry.api.trace.TraceState;
+import io.opentelemetry.api.trace.TraceStateBuilder;
+import org.apache.hadoop.thirdparty.protobuf.ByteString;
+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 spanContext = null;
+  public SpanContext(io.opentelemetry.api.trace.SpanContext spanContext) {
+    this.spanContext = spanContext;
   }
 
   public void close() {
+
+  }
+
+  public Map<String, String> getKVSpanContext(){
+    if(spanContext != null){
+      //TODO: may we should move this to Proto
+      Map<String, String> kvMap = new HashMap<>();
+      kvMap.put(TRACE_ID, spanContext.getTraceId());
+      kvMap.put(SPAN_ID, spanContext.getSpanId());
+      kvMap.put(TRACE_FLAGS, spanContext.getTraceFlags().asHex());
+      kvMap.putAll(spanContext.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);
+      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 " + kvMap != null? kvMap.toString() : "", e);
+      return null;
+    }
+
+
+  }
+
+  public io.opentelemetry.api.trace.SpanContext getSpanContext() {
+      return spanContext;
   }
 }

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 scope;
   public TraceScope(Span span) {
     this.span = span;
+    if(span != null){
+      this.scope = 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(scope != null){
+      scope.close();
+    }
     if (span != null) {
       span.close();
     }

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

@@ -20,28 +20,36 @@
 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.io.*;
+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,39 +47,43 @@ 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.getSpanContext());
+    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;
+    static Tracer globalTracer = new Tracer();
     private String name;
 
     public Builder(final String name) {
@@ -88,10 +94,7 @@ public Builder conf(TraceConfiguration conf) {
       return this;
     }
 
-    public Tracer build() {
-      if (globalTracer == null) {
-        globalTracer = new Tracer(name);
-      }
+    public synchronized Tracer build() {
       return globalTracer;
     }
   }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ProtoUtil.java

@@ -182,8 +182,8 @@ public static RpcRequestHeaderProto makeRpcRequestHeader(RPC.RpcKind rpcKind,
     Span span = Tracer.getCurrentSpan();
     if (span != null) {
       RPCTraceInfoProto.Builder traceInfoProtoBuilder =
-          RPCTraceInfoProto.newBuilder().setSpanContext(
-              TraceUtils.spanContextToByteString(span.getContext()));
+          RPCTraceInfoProto.newBuilder().putAllSpanContext(
+              TraceUtils.spanContextToMap(span.getContext()));
       result.setTraceInfo(traceInfoProtoBuilder);
     }
 

hadoop-common-project/hadoop-common/src/main/proto/RpcHeader.proto

@@ -63,7 +63,7 @@ enum RpcKindProto {
 message RPCTraceInfoProto {
     optional int64 traceId = 1; // parentIdHigh
     optional int64 parentId = 2; // parentIdLow
-    optional bytes spanContext = 3; // Trace SpanContext
+    map<string, string> spanContext = 3; // Trace SpanContext
 }
 
 /**