HBASE-25535 Set span kind to CLIENT in AbstractRpcClient (#2907)

Signed-off-by: Guanghao Zhang <[email protected]>

Author: Apache9

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

@@ -395,7 +395,7 @@ private void onCallFinished(Call call, HBaseRpcController hrc, Address addr,
   private Call callMethod(final Descriptors.MethodDescriptor md, final HBaseRpcController hrc,
     final Message param, Message returnType, final User ticket, final Address addr,
     final RpcCallback<Message> callback) {
-    Span span = TraceUtil.createSpan("RpcClient.callMethod")
+    Span span = TraceUtil.createClientSpan("RpcClient.callMethod")
       .setAttribute(TraceUtil.RPC_SERVICE_KEY, md.getService().getName())
       .setAttribute(TraceUtil.RPC_METHOD_KEY, md.getName())
       .setAttribute(TraceUtil.REMOTE_HOST_KEY, addr.getHostName())

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

@@ -105,6 +105,13 @@ public static Span createRemoteSpan(String name, Context ctx) {
     return getGlobalTracer().spanBuilder(name).setParent(ctx).setSpanKind(Kind.SERVER).startSpan();
   }
 
+  /**
+   * Create a span with {@link Kind#CLIENT}.
+   */
+  public static Span createClientSpan(String name) {
+    return createSpan(name, Kind.CLIENT);
+  }
+
   /**
    * Trace an asynchronous operation for a table.
    */

hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java

@@ -34,6 +34,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.internal.verification.VerificationModeFactory.times;
 
+import io.opentelemetry.api.trace.Span.Kind;
 import io.opentelemetry.api.trace.StatusCode;
 import io.opentelemetry.sdk.testing.junit4.OpenTelemetryRule;
 import io.opentelemetry.sdk.trace.data.SpanData;
@@ -455,14 +456,22 @@ private SpanData waitSpan(String name) {
     return traceRule.getSpans().stream().filter(s -> s.getName().equals(name)).findFirst().get();
   }
 
-  private void assertRpcAttribute(SpanData data, String methodName, InetSocketAddress addr) {
+  private void assertRpcAttribute(SpanData data, String methodName, InetSocketAddress addr,
+    Kind kind) {
     assertEquals(SERVICE.getDescriptorForType().getName(),
       data.getAttributes().get(TraceUtil.RPC_SERVICE_KEY));
     assertEquals(methodName, data.getAttributes().get(TraceUtil.RPC_METHOD_KEY));
     if (addr != null) {
       assertEquals(addr.getHostName(), data.getAttributes().get(TraceUtil.REMOTE_HOST_KEY));
       assertEquals(addr.getPort(), data.getAttributes().get(TraceUtil.REMOTE_PORT_KEY).intValue());
     }
+    assertEquals(kind, data.getKind());
+  }
+
+  private void assertRemoteSpan() {
+    SpanData data = waitSpan("RpcServer.process");
+    assertTrue(data.getParentSpanContext().isRemote());
+    assertEquals(Kind.SERVER, data.getKind());
   }
 
   @Test
@@ -474,8 +483,10 @@ public void testTracing() throws IOException, ServiceException {
       rpcServer.start();
       BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
       stub.pause(null, PauseRequestProto.newBuilder().setMs(100).build());
-      assertRpcAttribute(waitSpan("RpcClient.callMethod"), "pause", rpcServer.getListenerAddress());
-      assertRpcAttribute(waitSpan("RpcServer.callMethod"), "pause", null);
+      assertRpcAttribute(waitSpan("RpcClient.callMethod"), "pause", rpcServer.getListenerAddress(),
+        Kind.CLIENT);
+      assertRpcAttribute(waitSpan("RpcServer.callMethod"), "pause", null, Kind.INTERNAL);
+      assertRemoteSpan();
       assertSameTraceId();
       for (SpanData data : traceRule.getSpans()) {
         assertThat(
@@ -487,8 +498,10 @@ public void testTracing() throws IOException, ServiceException {
       traceRule.clearSpans();
       assertThrows(ServiceException.class,
         () -> stub.error(null, EmptyRequestProto.getDefaultInstance()));
-      assertRpcAttribute(waitSpan("RpcClient.callMethod"), "error", rpcServer.getListenerAddress());
-      assertRpcAttribute(waitSpan("RpcServer.callMethod"), "error", null);
+      assertRpcAttribute(waitSpan("RpcClient.callMethod"), "error", rpcServer.getListenerAddress(),
+        Kind.CLIENT);
+      assertRpcAttribute(waitSpan("RpcServer.callMethod"), "error", null, Kind.INTERNAL);
+      assertRemoteSpan();
       assertSameTraceId();
       for (SpanData data : traceRule.getSpans()) {
         assertEquals(StatusCode.ERROR, data.getStatus().getStatusCode());