Retrieve dubbo server.address/server.port according to latest SemConv

instrumentation/apache-dubbo-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apachedubbo/v2_7/DubboInstrumentationModule.java

@@ -33,6 +33,9 @@ public void registerHelperResources(HelperResourceBuilder helperResourceBuilder)
     helperResourceBuilder.register(
         "META-INF/services/org.apache.dubbo.rpc.Filter",
         "apache-dubbo-2.7/META-INF/services/org.apache.dubbo.rpc.Filter");
+    helperResourceBuilder.register(
+        "META-INF/services/org.apache.dubbo.rpc.cluster.Cluster",
+        "apache-dubbo-2.7/META-INF/services/org.apache.dubbo.rpc.cluster.Cluster");
   }
 
   @Override
@@ -50,6 +53,10 @@ public void injectClasses(ClassInjector injector) {
         .proxyBuilder(
             "io.opentelemetry.javaagent.instrumentation.apachedubbo.v2_7.OpenTelemetryServerFilter")
         .inject(InjectionMode.CLASS_ONLY);
+    injector
+        .proxyBuilder(
+            "io.opentelemetry.javaagent.instrumentation.apachedubbo.v2_7.RegistryCapturingClusterWrapperProxy")
+        .inject(InjectionMode.CLASS_ONLY);
   }
 
   @Override

instrumentation/apache-dubbo-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/apachedubbo/v2_7/RegistryCapturingClusterWrapperProxy.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apachedubbo.v2_7;
+
+import io.opentelemetry.instrumentation.apachedubbo.v2_7.internal.RegistryCapturingClusterWrapper;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.cluster.Cluster;
+import org.apache.dubbo.rpc.cluster.Directory;
+
+public final class RegistryCapturingClusterWrapperProxy implements Cluster {
+
+  private final RegistryCapturingClusterWrapper delegate;
+
+  @SuppressWarnings("unused")
+  public RegistryCapturingClusterWrapperProxy(Cluster cluster) {
+    this.delegate = new RegistryCapturingClusterWrapper(cluster);
+  }
+
+  @Override
+  public <T> Invoker<T> join(Directory<T> directory) {
+    return delegate.join(directory);
+  }
+
+  @SuppressWarnings("unused")
+  public <T> Invoker<T> join(Directory<T> directory, boolean buildFilterChain) {
+    return delegate.join(directory, buildFilterChain);
+  }
+}

instrumentation/apache-dubbo-2.7/javaagent/src/main/resources/apache-dubbo-2.7/META-INF/services/org.apache.dubbo.rpc.cluster.Cluster

@@ -0,0 +1 @@
+io.opentelemetry.javaagent.instrumentation.apachedubbo.v2_7.RegistryCapturingClusterWrapperProxy

instrumentation/apache-dubbo-2.7/javaagent/src/testDubbo/java/io/opentelemetry/javaagent/instrumentation/apachedubbo/v2_7/DubboAgentRegistryTest.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.apachedubbo.v2_7;
+
+import io.opentelemetry.instrumentation.apachedubbo.v2_7.AbstractDubboRegistryTest;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class DubboAgentRegistryTest extends AbstractDubboRegistryTest {
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @Override
+  protected InstrumentationExtension testing() {
+    return testing;
+  }
+}

instrumentation/apache-dubbo-2.7/library-autoconfigure/src/main/java/io/opentelemetry/instrumentation/apachedubbo/v2_7/DubboRequest.java

@@ -26,7 +26,7 @@ static DubboRequest create(RpcInvocation invocation, RpcContext context) {
         context.getLocalAddress());
   }
 
-  abstract RpcInvocation invocation();
+  public abstract RpcInvocation invocation();
 
   public abstract RpcContext context();
 

instrumentation/apache-dubbo-2.7/library-autoconfigure/src/main/java/io/opentelemetry/instrumentation/apachedubbo/v2_7/internal/DubboClientNetworkAttributesGetter.java

@@ -5,6 +5,9 @@
 
 package io.opentelemetry.instrumentation.apachedubbo.v2_7.internal;
 
+import static io.opentelemetry.instrumentation.apachedubbo.v2_7.internal.DubboRegistryUtil.buildServiceTarget;
+import static io.opentelemetry.instrumentation.apachedubbo.v2_7.internal.DubboRegistryUtil.extractRegistryAddress;
+
 import io.opentelemetry.instrumentation.apachedubbo.v2_7.DubboRequest;
 import io.opentelemetry.instrumentation.api.semconv.network.NetworkAttributesGetter;
 import io.opentelemetry.instrumentation.api.semconv.network.ServerAttributesGetter;
@@ -22,11 +25,19 @@ public final class DubboClientNetworkAttributesGetter
   @Nullable
   @Override
   public String getServerAddress(DubboRequest request) {
+    String registryAddress = extractRegistryAddress(request.invocation());
+    if (registryAddress != null) {
+      return registryAddress + "/" + buildServiceTarget(request.url());
+    }
     return request.url().getHost();
   }
 
+  @Nullable
   @Override
   public Integer getServerPort(DubboRequest request) {
+    if (extractRegistryAddress(request.invocation()) != null) {
+      return null;
+    }
     return request.url().getPort();
   }
 

instrumentation/apache-dubbo-2.7/library-autoconfigure/src/main/java/io/opentelemetry/instrumentation/apachedubbo/v2_7/internal/DubboRegistryUtil.java

@@ -0,0 +1,207 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.apachedubbo.v2_7.internal;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.cluster.Directory;
+
+/**
+ * This class is internal and is hence not for public use. Its APIs are unstable and can change at
+ * any time.
+ */
+public final class DubboRegistryUtil {
+
+  private static final MethodHandles.Lookup LOOKUP = MethodHandles.lookup();
+
+  private static final Map<Class<?>, Optional<MethodHandle>> DIRECTORY_ACCESSOR_CACHE =
+      new ConcurrentHashMap<>();
+  private static final Map<Class<?>, Optional<MethodHandle>> REGISTRY_ACCESSOR_CACHE =
+      new ConcurrentHashMap<>();
+  private static final Map<Class<?>, Optional<MethodHandle>> URL_ACCESSOR_CACHE =
+      new ConcurrentHashMap<>();
+
+  private static final ThreadLocal<String> CAPTURED_REGISTRY_ADDRESS = new ThreadLocal<>();
+
+  /**
+   * Used by {@link RegistryCapturingInvoker} while the cluster delegate runs (including into
+   * consumer protocol filters).
+   */
+  static void pushCapturedRegistryAddress(String address) {
+    CAPTURED_REGISTRY_ADDRESS.set(address);
+  }
+
+  public static void clearCapturedRegistryAddress() {
+    CAPTURED_REGISTRY_ADDRESS.remove();
+  }
+
+  @Nullable
+  public static String extractRegistryAddress(RpcInvocation invocation) {
+    String captured = CAPTURED_REGISTRY_ADDRESS.get();
+    if (captured != null) {
+      return captured;
+    }
+
+    Invoker<?> invoker = invocation.getInvoker();
+    if (invoker == null) {
+      return null;
+    }
+
+    Directory<?> directory = getDirectory(invoker);
+    if (directory != null) {
+      String address = tryExtractRegistryAddressFromDirectory(directory);
+      if (address != null) {
+        return address;
+      }
+    }
+
+    return null;
+  }
+
+  /**
+   * Resolves {@code protocol://host:port} from a registry-backed directory (for example {@code
+   * RegistryDirectory}), using {@code getRegistry()} when present and otherwise the {@code
+   * registry} field. Called once per consumer refer when {@link RegistryCapturingClusterWrapper}
+   * wraps the cluster invoker.
+   */
+  @Nullable
+  public static String tryExtractRegistryAddressFromDirectory(Directory<?> directory) {
+    return extractRegistryAddressFromDirectory(directory);
+  }
+
+  public static String buildServiceTarget(URL url) {
+    String interfaceName = url.getServiceInterface();
+    if (interfaceName == null || interfaceName.isEmpty()) {
+      interfaceName = url.getPath();
+    }
+    if (interfaceName == null || interfaceName.isEmpty()) {
+      return "";
+    }
+
+    String version = url.getParameter("version");
+    String group = url.getParameter("group");
+    boolean hasVersion = version != null && !version.isEmpty();
+    boolean hasGroup = group != null && !group.isEmpty();
+
+    if (!hasVersion && !hasGroup) {
+      return interfaceName;
+    }
+
+    StringBuilder sb = new StringBuilder(interfaceName);
+    sb.append(':');
+    if (hasVersion) {
+      sb.append(version);
+    }
+    if (hasGroup) {
+      sb.append(':').append(group);
+    }
+    return sb.toString();
+  }
+
+  @Nullable
+  private static Directory<?> getDirectory(Invoker<?> invoker) {
+    MethodHandle mh =
+        findAccessor(invoker.getClass(), "getDirectory", "directory", DIRECTORY_ACCESSOR_CACHE);
+    if (mh == null) {
+      return null;
+    }
+    try {
+      Object obj = mh.invoke(invoker);
+      return obj instanceof Directory ? (Directory<?>) obj : null;
+    } catch (Throwable t) {
+      return null;
+    }
+  }
+
+  @Nullable
+  private static String extractRegistryAddressFromDirectory(Directory<?> directory) {
+    MethodHandle getRegistry =
+        findAccessor(directory.getClass(), "getRegistry", "registry", REGISTRY_ACCESSOR_CACHE);
+    if (getRegistry == null) {
+      return null;
+    }
+    try {
+      Object registry = getRegistry.invoke(directory);
+      if (registry == null) {
+        return null;
+      }
+      MethodHandle getUrl = findAccessor(registry.getClass(), "getUrl", null, URL_ACCESSOR_CACHE);
+      if (getUrl == null) {
+        return null;
+      }
+      Object urlObj = getUrl.invoke(registry);
+      if (!(urlObj instanceof URL)) {
+        return null;
+      }
+      URL url = (URL) urlObj;
+      return url.getProtocol() + "://" + url.getAddress();
+    } catch (Throwable t) {
+      return null;
+    }
+  }
+
+  @Nullable
+  private static MethodHandle findAccessor(
+      Class<?> clazz,
+      String methodName,
+      @Nullable String fieldName,
+      Map<Class<?>, Optional<MethodHandle>> cache) {
+    return cache
+        .computeIfAbsent(
+            clazz,
+            cls -> {
+              MethodHandle mh = resolveMethod(cls, methodName);
+              if (mh != null) {
+                return Optional.of(mh);
+              }
+              if (fieldName != null) {
+                mh = resolveField(cls, fieldName);
+                if (mh != null) {
+                  return Optional.of(mh);
+                }
+              }
+              return Optional.empty();
+            })
+        .orElse(null);
+  }
+
+  @Nullable
+  private static MethodHandle resolveMethod(Class<?> clazz, String name) {
+    try {
+      Method m = clazz.getMethod(name);
+      m.setAccessible(true);
+      return LOOKUP.unreflect(m);
+    } catch (NoSuchMethodException | IllegalAccessException e) {
+      // ignore
+    }
+    return null;
+  }
+
+  @Nullable
+  private static MethodHandle resolveField(Class<?> clazz, String name) {
+    for (Class<?> c = clazz; c != null && c != Object.class; c = c.getSuperclass()) {
+      try {
+        Field f = c.getDeclaredField(name);
+        f.setAccessible(true);
+        return LOOKUP.unreflectGetter(f);
+      } catch (NoSuchFieldException | IllegalAccessException e) {
+        // ignore
+      }
+    }
+    return null;
+  }
+
+  private DubboRegistryUtil() {}
+}