HBASE-22120 Replace HTrace with OpenTelemetry

bin/hbase

@@ -480,6 +480,11 @@ add_jdk11_deps_to_classpath() {
   done
 }
 
+enable_trace() {
+  agent_jar=$(find lib/trace -type f -name "opentelemetry-javaagent-*")
+  HBASE_OPTS="$HBASE_OPTS -javaagent:$agent_jar $HBASE_TRACE_OPTS"
+}
+
 #Add the development env class path stuff
 if $in_dev_env; then
   add_maven_deps_to_classpath "cached_classpath.txt"
@@ -772,6 +777,11 @@ elif [ "${DEBUG}" = "true" ]; then
   echo "JDK11 jars skipped from classpath."
 fi
 
+if [[ -n "${HBASE_TRACE_OPTS}" ]]; then
+  echo "Attach opentelemetry agent to enable trace"
+  enable_trace
+fi
+
 # Have JVM dump heap if we run out of memory.  Files will be 'launch directory'
 # and are named like the following: java_pid21612.hprof. Apparently it doesn't
 # 'cost' to have this flag enabled. Its a 1.6 flag only. See:

conf/hbase-env.cmd

@@ -81,3 +81,8 @@ set HBASE_OPTS=%HBASE_OPTS% "-XX:+UseConcMarkSweepGC" "-Djava.net.preferIPv4Stac
 
 @rem Tell HBase whether it should manage it's own instance of ZooKeeper or not.
 @rem set HBASE_MANAGES_ZK=true
+
+@rem Uncomment to enable trace, you can change the options to use other exporters such as jaeger or
+@rem zipkin. See https://github.com/open-telemetry/opentelemetry-java-instrumentation on how to
+@rem configure exporters and other components through system properties.
+@rem set HBASE_TRACE_OPTS="-Dotel.resource.attributes=service.name=HBase -Dotel.traces.exporter=logging otel.metrics.exporter=none"

conf/hbase-env.sh

@@ -142,3 +142,8 @@
 # Override text processing tools for use by these launch scripts.
 # export GREP="${GREP-grep}"
 # export SED="${SED-sed}"
+
+# Uncomment to enable trace, you can change the options to use other exporters such as jaeger or
+# zipkin. See https://github.com/open-telemetry/opentelemetry-java-instrumentation on how to
+# configure exporters and other components through system properties.
+# export HBASE_TRACE_OPTS="-Dotel.resource.attributes=service.name=HBase -Dotel.traces.exporter=logging otel.metrics.exporter=none"

hbase-assembly/pom.xml

@@ -347,5 +347,11 @@
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-1.2-api</artifactId>
     </dependency>
+    <!-- Include OpenTelemetry agent -->
+    <dependency>
+      <groupId>io.opentelemetry.javaagent</groupId>
+      <artifactId>opentelemetry-javaagent</artifactId>
+      <classifier>all</classifier>
+    </dependency>
   </dependencies>
 </project>

hbase-assembly/src/main/assembly/hadoop-three-compat.xml

@@ -111,6 +111,7 @@
               <exclude>org.apache.yetus:audience-annotations</exclude>
               <exclude>org.slf4j:*</exclude>
               <exclude>org.apache.logging.log4j:*</exclude>
+              <exclude>io.opentelemetry.javaagent:*</exclude>
             </excludes>
           </dependencySet>
         </dependencySets>
@@ -221,7 +222,7 @@
       <outputDirectory>lib/jdk11</outputDirectory>
       <useTransitiveDependencies>true</useTransitiveDependencies>
       <includes>
-          <include>com.sun.activation:javax.activation</include>
+        <include>com.sun.activation:javax.activation</include>
         <!-- The following artifacts are transitive dependencies of com.sun.xml.ws:jaxws-ri:pom
              They are needed to be included in lib/jdk11 to be added to classpath during
              Java 11 runtime
@@ -253,6 +254,12 @@
         <include>jakarta.jws:jakarta.jws-api</include>
       </includes>
     </dependencySet>
+    <dependencySet>
+      <outputDirectory>lib/trace</outputDirectory>
+      <includes>
+        <include>io.opentelemetry.javaagent:*</include>
+      </includes>
+    </dependencySet>
   </dependencySets>
 
 </assembly>

hbase-asyncfs/src/test/java/org/apache/hadoop/hbase/io/asyncfs/AsyncFSTestBase.java

@@ -19,11 +19,9 @@
 
 import java.io.File;
 import java.io.IOException;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
-import org.apache.hadoop.hbase.trace.TraceUtil;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -99,7 +97,6 @@ protected static void startMiniDFSCluster(int servers) throws IOException {
 
     Configuration conf = UTIL.getConfiguration();
 
-    TraceUtil.initTracer(conf);
     CLUSTER = new MiniDFSCluster.Builder(conf).numDataNodes(servers).build();
     CLUSTER.waitClusterUp();
   }

hbase-client/pom.xml

@@ -133,8 +133,8 @@
       <artifactId>zookeeper</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.apache.htrace</groupId>
-      <artifactId>htrace-core4</artifactId>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-api</artifactId>
     </dependency>
     <dependency>
       <groupId>org.jruby.jcodings</groupId>
@@ -144,6 +144,16 @@
       <groupId>org.jruby.joni</groupId>
       <artifactId>joni</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk-testing</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>jcl-over-slf4j</artifactId>

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnection.java

@@ -64,8 +64,8 @@ public interface AsyncConnection extends Closeable {
   /**
    * Retrieve an {@link AsyncTable} implementation for accessing a table.
    * <p>
-   * The returned instance will use default configs. Use {@link #getTableBuilder(TableName)} if
-   * you want to customize some configs.
+   * The returned instance will use default configs. Use {@link #getTableBuilder(TableName)} if you
+   * want to customize some configs.
    * <p>
    * This method no longer checks table existence. An exception will be thrown if the table does not
    * exist only when the first operation is attempted.

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java

@@ -27,6 +27,7 @@
 import static org.apache.hadoop.hbase.client.NonceGenerator.CLIENT_NONCES_ENABLED_KEY;
 import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
 
+import io.opentelemetry.api.trace.Span;
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.util.Optional;
@@ -38,6 +39,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import org.apache.commons.io.IOUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.AuthUtil;
@@ -51,6 +53,7 @@
 import org.apache.hadoop.hbase.ipc.RpcClientFactory;
 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
 import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.trace.TraceUtil;
 import org.apache.hadoop.hbase.util.ConcurrentMapUtils;
 import org.apache.hadoop.hbase.util.Threads;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -121,7 +124,7 @@ class AsyncConnectionImpl implements AsyncConnection {
   private volatile ConnectionOverAsyncConnection conn;
 
   public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, String clusterId,
-      SocketAddress localAddress, User user) {
+    SocketAddress localAddress, User user) {
     this.conf = conf;
     this.user = user;
 
@@ -135,8 +138,8 @@ public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, Stri
     } else {
       this.metrics = Optional.empty();
     }
-    this.rpcClient = RpcClientFactory.createClient(
-        conf, clusterId, localAddress, metrics.orElse(null));
+    this.rpcClient =
+      RpcClientFactory.createClient(conf, clusterId, localAddress, metrics.orElse(null));
     this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
     this.rpcTimeout =
       (int) Math.min(Integer.MAX_VALUE, TimeUnit.NANOSECONDS.toMillis(connConf.getRpcTimeoutNs()));
@@ -159,14 +162,13 @@ public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, Stri
         LOG.warn("{} is true, but {} is not set", STATUS_PUBLISHED, STATUS_LISTENER_CLASS);
       } else {
         try {
-          listener = new ClusterStatusListener(
-            new ClusterStatusListener.DeadServerHandler() {
-              @Override
-              public void newDead(ServerName sn) {
-                locator.clearCache(sn);
-                rpcClient.cancelConnections(sn);
-              }
-            }, conf, listenerClass);
+          listener = new ClusterStatusListener(new ClusterStatusListener.DeadServerHandler() {
+            @Override
+            public void newDead(ServerName sn) {
+              locator.clearCache(sn);
+              rpcClient.cancelConnections(sn);
+            }
+          }, conf, listenerClass);
         } catch (IOException e) {
           LOG.warn("Failed create of ClusterStatusListener, not a critical, ignoring...", e);
         }
@@ -206,28 +208,30 @@ public boolean isClosed() {
 
   @Override
   public void close() {
-    if (!closed.compareAndSet(false, true)) {
-      return;
-    }
-    LOG.info("Connection has been closed by {}.", Thread.currentThread().getName());
-    if(LOG.isDebugEnabled()){
-      logCallStack(Thread.currentThread().getStackTrace());
-    }
-    IOUtils.closeQuietly(clusterStatusListener,
-      e -> LOG.warn("failed to close clusterStatusListener", e));
-    IOUtils.closeQuietly(rpcClient, e -> LOG.warn("failed to close rpcClient", e));
-    IOUtils.closeQuietly(registry, e -> LOG.warn("failed to close registry", e));
-    synchronized (this) {
-      if (choreService != null) {
-        choreService.shutdown();
-        choreService = null;
+    TraceUtil.trace(() -> {
+      if (!closed.compareAndSet(false, true)) {
+        return;
       }
-    }
-    metrics.ifPresent(MetricsConnection::shutdown);
-    ConnectionOverAsyncConnection c = this.conn;
-    if (c != null) {
-      c.closePool();
-    }
+      LOG.info("Connection has been closed by {}.", Thread.currentThread().getName());
+      if (LOG.isDebugEnabled()) {
+        logCallStack(Thread.currentThread().getStackTrace());
+      }
+      IOUtils.closeQuietly(clusterStatusListener,
+        e -> LOG.warn("failed to close clusterStatusListener", e));
+      IOUtils.closeQuietly(rpcClient, e -> LOG.warn("failed to close rpcClient", e));
+      IOUtils.closeQuietly(registry, e -> LOG.warn("failed to close registry", e));
+      synchronized (this) {
+        if (choreService != null) {
+          choreService.shutdown();
+          choreService = null;
+        }
+      }
+      metrics.ifPresent(MetricsConnection::shutdown);
+      ConnectionOverAsyncConnection c = this.conn;
+      if (c != null) {
+        c.closePool();
+      }
+    }, "AsyncConnection.close");
   }
 
   private void logCallStack(StackTraceElement[] stackTraceElements) {
@@ -341,7 +345,7 @@ public AsyncTable<AdvancedScanResultConsumer> build() {
 
   @Override
   public AsyncTableBuilder<ScanResultConsumer> getTableBuilder(TableName tableName,
-      ExecutorService pool) {
+    ExecutorService pool) {
     return new AsyncTableBuilderBase<ScanResultConsumer>(tableName, connConf) {
 
       @Override
@@ -382,7 +386,7 @@ public AsyncBufferedMutatorBuilder getBufferedMutatorBuilder(TableName tableName
 
   @Override
   public AsyncBufferedMutatorBuilder getBufferedMutatorBuilder(TableName tableName,
-      ExecutorService pool) {
+    ExecutorService pool) {
     return new AsyncBufferedMutatorBuilderImpl(connConf, getTableBuilder(tableName, pool),
       RETRY_TIMER);
   }
@@ -404,30 +408,39 @@ public Connection toConnection() {
     return c;
   }
 
+  private Hbck getHbckInternal(ServerName masterServer) {
+    Span.current().setAttribute(TraceUtil.SERVER_NAME_KEY, masterServer.getServerName());
+    // we will not create a new connection when creating a new protobuf stub, and for hbck there
+    // will be no performance consideration, so for simplification we will create a new stub every
+    // time instead of caching the stub here.
+    return new HBaseHbck(MasterProtos.HbckService.newBlockingStub(
+      rpcClient.createBlockingRpcChannel(masterServer, user, rpcTimeout)), rpcControllerFactory);
+  }
+
   @Override
   public CompletableFuture<Hbck> getHbck() {
-    CompletableFuture<Hbck> future = new CompletableFuture<>();
-    addListener(registry.getActiveMaster(), (sn, error) -> {
-      if (error != null) {
-        future.completeExceptionally(error);
-      } else {
-        try {
-          future.complete(getHbck(sn));
-        } catch (IOException e) {
-          future.completeExceptionally(e);
+    return TraceUtil.tracedFuture(() -> {
+      CompletableFuture<Hbck> future = new CompletableFuture<>();
+      addListener(registry.getActiveMaster(), (sn, error) -> {
+        if (error != null) {
+          future.completeExceptionally(error);
+        } else {
+          future.complete(getHbckInternal(sn));
         }
-      }
-    });
-    return future;
+      });
+      return future;
+    }, "AsyncConnection.getHbck");
   }
 
   @Override
-  public Hbck getHbck(ServerName masterServer) throws IOException {
-    // we will not create a new connection when creating a new protobuf stub, and for hbck there
-    // will be no performance consideration, so for simplification we will create a new stub every
-    // time instead of caching the stub here.
-    return new HBaseHbck(MasterProtos.HbckService.newBlockingStub(
-      rpcClient.createBlockingRpcChannel(masterServer, user, rpcTimeout)), rpcControllerFactory);
+  public Hbck getHbck(ServerName masterServer) {
+    return TraceUtil.trace(new Supplier<Hbck>() {
+
+      @Override
+      public Hbck get() {
+        return getHbckInternal(masterServer);
+      }
+    }, "AsyncConnection.getHbck");
   }
 
   Optional<MetricsConnection> getConnectionMetrics() {