Revert "HBASE-26124 Backport HBASE-25373 "Remove HTrace completely in code base and try to make use of OpenTelemetry" to branch-2 (#3529)"

This reverts commit f049301.

Author: taklwu

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

@@ -19,9 +19,11 @@
 
 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;
@@ -102,6 +104,7 @@ protected static void startMiniDFSCluster(int servers) throws IOException {
     org.apache.log4j.Logger.getLogger(org.apache.hadoop.metrics2.impl.MetricsSystemImpl.class)
       .setLevel(org.apache.log4j.Level.ERROR);
 
+    TraceUtil.initTracer(conf);
     CLUSTER = new MiniDFSCluster.Builder(conf).numDataNodes(servers).build();
     CLUSTER.waitClusterUp();
   }

hbase-client/pom.xml

@@ -145,8 +145,8 @@
       <artifactId>zookeeper</artifactId>
     </dependency>
     <dependency>
-      <groupId>io.opentelemetry</groupId>
-      <artifactId>opentelemetry-api</artifactId>
+      <groupId>org.apache.htrace</groupId>
+      <artifactId>htrace-core4</artifactId>
     </dependency>
     <dependency>
       <groupId>org.jruby.jcodings</groupId>

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

@@ -46,8 +46,10 @@
 import org.apache.hadoop.hbase.client.backoff.ServerStatistics;
 import org.apache.hadoop.hbase.client.coprocessor.Batch;
 import org.apache.hadoop.hbase.exceptions.ClientExceptionsUtil;
+import org.apache.hadoop.hbase.trace.TraceUtil;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.htrace.core.Tracer;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -570,9 +572,13 @@ private Collection<? extends Runnable> getNewMultiActionRunnable(ServerName serv
       asyncProcess.incTaskCounters(multiAction.getRegions(), server);
       SingleServerRequestRunnable runnable = createSingleServerRequest(
               multiAction, numAttempt, server, callsInProgress);
+      Tracer tracer = Tracer.curThreadTracer();
 
-      // remove trace for runnable because HBASE-25373 and OpenTelemetry do not cover TraceRunnable
-      return Collections.singletonList(runnable);
+      if (tracer == null) {
+        return Collections.singletonList(runnable);
+      } else {
+        return Collections.singletonList(tracer.wrap(runnable, "AsyncProcess.sendMultiAction"));
+      }
     }
 
     // group the actions by the amount of delay
@@ -592,10 +598,12 @@ private Collection<? extends Runnable> getNewMultiActionRunnable(ServerName serv
     List<Runnable> toReturn = new ArrayList<>(actions.size());
     for (DelayingRunner runner : actions.values()) {
       asyncProcess.incTaskCounters(runner.getActions().getRegions(), server);
+      String traceText = "AsyncProcess.sendMultiAction";
       Runnable runnable = createSingleServerRequest(runner.getActions(), numAttempt, server, callsInProgress);
       // use a delay runner only if we need to sleep for some time
       if (runner.getSleepTime() > 0) {
         runner.setRunner(runnable);
+        traceText = "AsyncProcess.clientBackoff.sendMultiAction";
         runnable = runner;
         if (asyncProcess.connection.getConnectionMetrics() != null) {
           asyncProcess.connection.getConnectionMetrics()
@@ -606,7 +614,7 @@ private Collection<? extends Runnable> getNewMultiActionRunnable(ServerName serv
           asyncProcess.connection.getConnectionMetrics().incrNormalRunners();
         }
       }
-      // remove trace for runnable because HBASE-25373 and OpenTelemetry do not cover TraceRunnable
+      runnable = TraceUtil.wrap(runnable, traceText);
       toReturn.add(runnable);
 
     }

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

@@ -26,6 +26,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.hadoop.hbase.trace.TraceUtil;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -167,8 +168,7 @@ public ResultBoundedCompletionService(
 
   public void submit(RetryingCallable<V> task, int callTimeout, int id) {
     QueueingFuture<V> newFuture = new QueueingFuture<>(task, callTimeout, id);
-    // remove trace for runnable because HBASE-25373 and OpenTelemetry do not cover TraceRunnable
-    executor.execute(newFuture);
+    executor.execute(TraceUtil.wrap(newFuture, "ResultBoundedCompletionService.submit"));
     tasks[id] = newFuture;
   }
 

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcConnection.java

@@ -24,9 +24,6 @@
 import static org.apache.hadoop.hbase.ipc.IPCUtil.setCancelled;
 import static org.apache.hadoop.hbase.ipc.IPCUtil.write;
 
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.context.Context;
-import io.opentelemetry.context.Scope;
 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.DataInputStream;
@@ -65,6 +62,7 @@
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.htrace.core.TraceScope;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -595,12 +593,9 @@ private void negotiateCryptoAes(RPCProtos.CryptoCipherMeta cryptoCipherMeta)
   }
 
   private void tracedWriteRequest(Call call) throws IOException {
-    Span span = TraceUtil.getGlobalTracer().spanBuilder("RpcClientImpl.tracedWriteRequest")
-      .setParent(Context.current().with(call.span)).startSpan();
-    try (Scope scope = span.makeCurrent()) {
+    try (TraceScope ignored = TraceUtil.createTrace("RpcClientImpl.tracedWriteRequest",
+          call.span)) {
       writeRequest(call);
-    } finally {
-      span.end();
     }
   }
 

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/Call.java

@@ -17,21 +17,20 @@
  */
 package org.apache.hadoop.hbase.ipc;
 
-import io.opentelemetry.api.trace.Span;
 import java.io.IOException;
 import java.util.Optional;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
 import org.apache.hadoop.hbase.CellScanner;
 import org.apache.hadoop.hbase.client.MetricsConnection;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.Tracer;
 import org.apache.yetus.audience.InterfaceAudience;
-
 import org.apache.hbase.thirdparty.com.google.protobuf.Descriptors;
 import org.apache.hbase.thirdparty.com.google.protobuf.Message;
 import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
 import org.apache.hbase.thirdparty.io.netty.util.Timeout;
-
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 
 /** A call waiting for a value. */
@@ -74,7 +73,7 @@ protected Call(int id, final Descriptors.MethodDescriptor md, Message param,
     this.timeout = timeout;
     this.priority = priority;
     this.callback = callback;
-    this.span = Span.current();
+    this.span = Tracer.getCurrentSpan();
   }
 
   /**

hbase-common/pom.xml

@@ -192,8 +192,8 @@
     </dependency>
     <!-- tracing Dependencies -->
     <dependency>
-      <groupId>io.opentelemetry</groupId>
-      <artifactId>opentelemetry-api</artifactId>
+      <groupId>org.apache.htrace</groupId>
+      <artifactId>htrace-core4</artifactId>
     </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>

hbase-common/src/main/java/org/apache/hadoop/hbase/trace/HBaseHTraceConfiguration.java

@@ -0,0 +1,80 @@
+/**
+ * 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.hbase.trace;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.htrace.core.HTraceConfiguration;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@InterfaceAudience.Private
+public class HBaseHTraceConfiguration extends HTraceConfiguration {
+  private static final Logger LOG = LoggerFactory.getLogger(HBaseHTraceConfiguration.class);
+
+  public static final String KEY_PREFIX = "hbase.htrace.";
+
+  private Configuration conf;
+
+  private void handleDeprecation(String key) {
+    String oldKey = "hbase." + key;
+    String newKey = KEY_PREFIX + key;
+    String oldValue = conf.get(oldKey);
+    if (oldValue != null) {
+      LOG.warn("Warning: using deprecated configuration key " + oldKey +
+          ".  Please use " + newKey + " instead.");
+      String newValue = conf.get(newKey);
+      if (newValue == null) {
+        conf.set(newKey, oldValue);
+      } else {
+        LOG.warn("Conflicting values for " + newKey + " and " + oldKey +
+            ".  Using " + newValue);
+      }
+    }
+  }
+
+  public HBaseHTraceConfiguration(Configuration conf) {
+    this.conf = conf;
+    handleDeprecation("local-file-span-receiver.path");
+    handleDeprecation("local-file-span-receiver.capacity");
+    handleDeprecation("sampler.frequency");
+    handleDeprecation("sampler.fraction");
+    handleDeprecation("zipkin.collector-hostname");
+    handleDeprecation("zipkin.collector-port");
+    handleDeprecation("zipkin.num-threads");
+    handleDeprecation("zipkin.traced-service-hostname");
+    handleDeprecation("zipkin.traced-service-port");
+  }
+
+  @Override
+  public String get(String key) {
+    return conf.get(KEY_PREFIX + key);
+  }
+
+  @Override
+  public String get(String key, String defaultValue) {
+    return conf.get(KEY_PREFIX + key,defaultValue);
+
+  }
+
+  @Override
+  public boolean getBoolean(String key, boolean defaultValue) {
+    return conf.getBoolean(KEY_PREFIX + key, defaultValue);
+  }
+}

hbase-common/src/main/java/org/apache/hadoop/hbase/trace/SpanReceiverHost.java

@@ -0,0 +1,120 @@
+/**
+ * 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.hbase.trace;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashSet;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.htrace.core.SpanReceiver;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class provides functions for reading the names of SpanReceivers from
+ * hbase-site.xml, adding those SpanReceivers to the Tracer, and closing those
+ * SpanReceivers when appropriate.
+ */
+@InterfaceAudience.Private
+public class SpanReceiverHost {
+  public static final String SPAN_RECEIVERS_CONF_KEY = "hbase.trace.spanreceiver.classes";
+  private static final Logger LOG = LoggerFactory.getLogger(SpanReceiverHost.class);
+  private Collection<SpanReceiver> receivers;
+  private Configuration conf;
+  private boolean closed = false;
+
+  private enum SingletonHolder {
+    INSTANCE;
+    final transient Object lock = new Object();
+    transient SpanReceiverHost host = null;
+  }
+
+  public static SpanReceiverHost getInstance(Configuration conf) {
+    synchronized (SingletonHolder.INSTANCE.lock) {
+      if (SingletonHolder.INSTANCE.host != null) {
+        return SingletonHolder.INSTANCE.host;
+      }
+
+      SpanReceiverHost host = new SpanReceiverHost(conf);
+      host.loadSpanReceivers();
+      SingletonHolder.INSTANCE.host = host;
+      return SingletonHolder.INSTANCE.host;
+    }
+
+  }
+
+  public static Configuration getConfiguration(){
+    synchronized (SingletonHolder.INSTANCE.lock) {
+      if (SingletonHolder.INSTANCE.host == null || SingletonHolder.INSTANCE.host.conf == null) {
+        return null;
+      }
+
+      return SingletonHolder.INSTANCE.host.conf;
+    }
+  }
+
+  SpanReceiverHost(Configuration conf) {
+    receivers = new HashSet<>();
+    this.conf = conf;
+  }
+
+  /**
+   * Reads the names of classes specified in the {@code hbase.trace.spanreceiver.classes} property
+   * and instantiates and registers them with the Tracer.
+   */
+  public void loadSpanReceivers() {
+    String[] receiverNames = conf.getStrings(SPAN_RECEIVERS_CONF_KEY);
+    if (receiverNames == null || receiverNames.length == 0) {
+      return;
+    }
+
+    SpanReceiver.Builder builder = new SpanReceiver.Builder(new HBaseHTraceConfiguration(conf));
+    for (String className : receiverNames) {
+      className = className.trim();
+
+      SpanReceiver receiver = builder.className(className).build();
+      if (receiver != null) {
+        receivers.add(receiver);
+        LOG.info("SpanReceiver {} was loaded successfully.", className);
+      }
+    }
+    for (SpanReceiver rcvr : receivers) {
+      TraceUtil.addReceiver(rcvr);
+    }
+  }
+
+  /**
+   * Calls close() on all SpanReceivers created by this SpanReceiverHost.
+   */
+  public synchronized void closeReceivers() {
+    if (closed) {
+      return;
+    }
+
+    closed = true;
+    for (SpanReceiver rcvr : receivers) {
+      try {
+        rcvr.close();
+      } catch (IOException e) {
+        LOG.warn("Unable to close SpanReceiver correctly: " + e.getMessage(), e);
+      }
+    }
+  }
+}