otel: plumb optional labels other than disconnect_error

Author: AgraVator

api/src/main/java/io/grpc/LoadBalancer.java

@@ -135,6 +135,12 @@ public abstract class LoadBalancer {
   public static final Attributes.Key<Boolean> IS_PETIOLE_POLICY =
       Attributes.Key.create("io.grpc.IS_PETIOLE_POLICY");
 
+  /**
+   * The name of the locality that this EquivalentAddressGroup is in.
+   */
+  public static final Attributes.Key<String> ATTR_LOCALITY_NAME =
+      Attributes.Key.create("io.grpc.lb.locality");
+
   /**
    * A picker that always returns an erring pick.
    *

core/src/main/java/io/grpc/internal/InternalSubchannel.java

@@ -49,7 +49,9 @@
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.MetricRecorder;
+import io.grpc.NameResolver;
 import io.grpc.Status;
+import io.grpc.SecurityLevel;
 import io.grpc.SynchronizationContext;
 import io.grpc.SynchronizationContext.ScheduledHandle;
 import java.net.SocketAddress;
@@ -163,9 +165,6 @@ protected void handleNotInUse() {
   private volatile Attributes connectedAddressAttributes;
   private final SubchannelMetrics subchannelMetrics;
   private final String target;
-  private final String backendService;
-  private final String locality;
-  private final String securityLevel;
 
   InternalSubchannel(LoadBalancer.CreateSubchannelArgs args, String authority, String userAgent,
                      BackoffPolicy.Provider backoffPolicyProvider,
@@ -175,7 +174,7 @@ protected void handleNotInUse() {
                      Callback callback, InternalChannelz channelz, CallTracer callsTracer,
                      ChannelTracer channelTracer, InternalLogId logId,
                      ChannelLogger channelLogger, List<ClientTransportFilter> transportFilters,
-                     String target, String backendService, String locality, String securityLevel,
+                     String target,
                      MetricRecorder metricRecorder) {
     List<EquivalentAddressGroup> addressGroups = args.getAddresses();
     Preconditions.checkNotNull(addressGroups, "addressGroups");
@@ -201,9 +200,6 @@ protected void handleNotInUse() {
     this.transportFilters = transportFilters;
     this.reconnectDisabled = args.getOption(LoadBalancer.DISABLE_SUBCHANNEL_RECONNECT_KEY);
     this.target = target;
-    this.backendService = backendService;
-    this.locality = locality;
-    this.securityLevel = securityLevel;
     this.subchannelMetrics = new SubchannelMetrics(metricRecorder);
   }
 
@@ -592,8 +588,13 @@ public Attributes filterTransport(Attributes attributes) {
     @Override
     public void transportReady() {
       channelLogger.log(ChannelLogLevel.INFO, "READY");
-      subchannelMetrics.recordConnectionAttemptSucceeded(
-          buildLabelSet(null, extractSecurityLevel()));
+      subchannelMetrics.recordConnectionAttemptSucceeded(buildLabelSet(
+          addressIndex.getCurrentEagAttributes().get(NameResolver.ATTR_BACKEND_SERVICE),
+          addressIndex.getCurrentEagAttributes().get(LoadBalancer.ATTR_LOCALITY_NAME),
+          null,
+          extractSecurityLevel(
+              addressIndex.getCurrentEagAttributes().get(GrpcAttributes.ATTR_SECURITY_LEVEL))
+      ));
       syncContext.execute(new Runnable() {
         @Override
         public void run() {
@@ -623,7 +624,11 @@ public void transportShutdown(final Status s) {
       channelLogger.log(
           ChannelLogLevel.INFO, "{0} SHUTDOWN with {1}", transport.getLogId(), printShortStatus(s));
       shutdownInitiated = true;
-      subchannelMetrics.recordConnectionAttemptFailed(buildLabelSet("Peer Pressure", null));
+      subchannelMetrics.recordConnectionAttemptFailed(buildLabelSet(
+          addressIndex.getCurrentEagAttributes().get(NameResolver.ATTR_BACKEND_SERVICE),
+          addressIndex.getCurrentEagAttributes().get(LoadBalancer.ATTR_LOCALITY_NAME),
+          null, null
+      ));
       syncContext.execute(new Runnable() {
         @Override
         public void run() {
@@ -664,8 +669,13 @@ public void transportTerminated() {
       for (ClientTransportFilter filter : transportFilters) {
         filter.transportTerminated(transport.getAttributes());
       }
-      subchannelMetrics.recordDisconnection(buildLabelSet("Peer Pressure",
-          null));
+      subchannelMetrics.recordDisconnection(buildLabelSet(
+          addressIndex.getCurrentEagAttributes().get(NameResolver.ATTR_BACKEND_SERVICE),
+          addressIndex.getCurrentEagAttributes().get(LoadBalancer.ATTR_LOCALITY_NAME),
+          "Peer Pressure",
+          extractSecurityLevel(
+              addressIndex.getCurrentEagAttributes().get(GrpcAttributes.ATTR_SECURITY_LEVEL))
+      ));
       syncContext.execute(new Runnable() {
         @Override
         public void run() {
@@ -677,8 +687,20 @@ public void run() {
       });
     }
 
-    private String extractSecurityLevel() {
-      return "Hold the door!";
+    private String extractSecurityLevel(SecurityLevel securityLevel) {
+      if (securityLevel == null) {
+        return "none";
+      }
+      switch (securityLevel) {
+        case NONE:
+          return "none";
+        case INTEGRITY:
+          return "integrity_only";
+        case PRIVACY_AND_INTEGRITY:
+          return "privacy_and_integrity";
+        default:
+          throw new IllegalArgumentException("Unknown SecurityLevel: " + securityLevel);
+      }
     }
   }
 
@@ -839,13 +861,14 @@ private String printShortStatus(Status status) {
     return buffer.toString();
   }
 
-  private OtelMetricsAttributes buildLabelSet(String disconnectError, String secLevel) {
+  private OtelMetricsAttributes buildLabelSet(String backendService, String locality,
+                                              String disconnectError, String securityLevel) {
     return new OtelMetricsAttributes(
         target,
         backendService,
         locality,
         disconnectError,
-        secLevel != null ? secLevel : securityLevel
+        securityLevel
     );
   }
 

core/src/main/java/io/grpc/internal/ManagedChannelImpl.java

@@ -415,7 +415,7 @@ void exitIdleMode() {
     LbHelperImpl lbHelper = new LbHelperImpl();
     lbHelper.lb = loadBalancerFactory.newLoadBalancer(lbHelper);
     // Delay setting lbHelper until fully initialized, since loadBalancerFactory is user code and
-    // may throw. We don't want to confuse our state, even if we will enter panic mode.
+    // may throw. We don't want to confuse our state, even if we enter panic mode.
     this.lbHelper = lbHelper;
 
     channelStateManager.gotoState(CONNECTING);
@@ -1466,9 +1466,6 @@ void onStateChange(InternalSubchannel is, ConnectivityStateInfo newState) {
           subchannelLogger,
           transportFilters,
           target,
-          "",
-          "",
-          "",
           lbHelper.getMetricRecorder());
       oobChannelTracer.reportEvent(new ChannelTrace.Event.Builder()
           .setDescription("Child Subchannel created")
@@ -1901,9 +1898,6 @@ void onNotInUse(InternalSubchannel is) {
           subchannelLogId,
           subchannelLogger,
           transportFilters, target,
-          "",
-          "",
-          "",
           lbHelper.getMetricRecorder());
 
       channelTracer.reportEvent(new ChannelTrace.Event.Builder()

core/src/main/java/io/grpc/internal/PickFirstLeafLoadBalancer.java

@@ -92,7 +92,7 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
       return Status.FAILED_PRECONDITION.withDescription("Already shut down");
     }
 
-    // Cache whether or not this is a petiole policy, which is based off of an address attribute
+    // Check weather or not this is a petiole policy, which is based off of an address attribute
     Boolean isPetiolePolicy = resolvedAddresses.getAttributes().get(IS_PETIOLE_POLICY);
     this.notAPetiolePolicy = isPetiolePolicy == null || !isPetiolePolicy;
 

core/src/test/java/io/grpc/internal/InternalSubchannelTest.java

@@ -1449,9 +1449,6 @@ private void createInternalSubchannel(boolean reconnectDisabled,
         new ChannelLoggerImpl(subchannelTracer, fakeClock.getTimeProvider()),
         Collections.emptyList(),
         "",
-        "",
-        "",
-        "",
         new MetricRecorder() {}
     );
   }

xds/src/main/java/io/grpc/xds/ClusterImplLoadBalancer.java

@@ -305,7 +305,7 @@ private List<EquivalentAddressGroup> withAdditionalAttributes(
 
     private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAttributes) {
       Locality locality = addressAttributes.get(XdsAttributes.ATTR_LOCALITY);
-      String localityName = addressAttributes.get(XdsAttributes.ATTR_LOCALITY_NAME);
+      String localityName = addressAttributes.get(LoadBalancer.ATTR_LOCALITY_NAME);
 
       // Endpoint addresses resolved by ClusterResolverLoadBalancer should always contain
       // attributes with its locality, including endpoints in LOGICAL_DNS clusters.

xds/src/main/java/io/grpc/xds/ClusterResolverLoadBalancer.java

@@ -428,7 +428,7 @@ public void run() {
                   Attributes attr =
                       endpoint.eag().getAttributes().toBuilder()
                           .set(XdsAttributes.ATTR_LOCALITY, locality)
-                          .set(XdsAttributes.ATTR_LOCALITY_NAME, localityName)
+                          .set(LoadBalancer.ATTR_LOCALITY_NAME, localityName)
                           .set(XdsAttributes.ATTR_LOCALITY_WEIGHT,
                               localityLbInfo.localityWeight())
                           .set(XdsAttributes.ATTR_SERVER_WEIGHT, weight)
@@ -679,7 +679,7 @@ public Status onResult2(final ResolutionResult resolutionResult) {
               String localityName = localityName(LOGICAL_DNS_CLUSTER_LOCALITY);
               Attributes attr = eag.getAttributes().toBuilder()
                       .set(XdsAttributes.ATTR_LOCALITY, LOGICAL_DNS_CLUSTER_LOCALITY)
-                      .set(XdsAttributes.ATTR_LOCALITY_NAME, localityName)
+                      .set(LoadBalancer.ATTR_LOCALITY_NAME, localityName)
                       .set(XdsAttributes.ATTR_ADDRESS_NAME, dnsHostName)
                       .build();
               eag = new EquivalentAddressGroup(eag.getAddresses(), attr);

xds/src/main/java/io/grpc/xds/WrrLocalityLoadBalancer.java

@@ -74,7 +74,7 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     Map<String, Integer> localityWeights = new HashMap<>();
     for (EquivalentAddressGroup eag : resolvedAddresses.getAddresses()) {
       Attributes eagAttrs = eag.getAttributes();
-      String locality = eagAttrs.get(XdsAttributes.ATTR_LOCALITY_NAME);
+      String locality = eagAttrs.get(LoadBalancer.ATTR_LOCALITY_NAME);
       Integer localityWeight = eagAttrs.get(XdsAttributes.ATTR_LOCALITY_WEIGHT);
 
       if (locality == null) {

xds/src/main/java/io/grpc/xds/XdsAttributes.java

@@ -81,13 +81,6 @@ final class XdsAttributes {
   static final Attributes.Key<Locality> ATTR_LOCALITY =
       Attributes.Key.create("io.grpc.xds.XdsAttributes.locality");
 
-  /**
-   * The name of the locality that this EquivalentAddressGroup is in.
-   */
-  @EquivalentAddressGroup.Attr
-  static final Attributes.Key<String> ATTR_LOCALITY_NAME =
-      Attributes.Key.create("io.grpc.xds.XdsAttributes.localityName");
-
   /**
    * Endpoint weight for load balancing purposes.
    */

xds/src/test/java/io/grpc/xds/ClusterImplLoadBalancerTest.java

@@ -1017,7 +1017,7 @@ public String toString() {
     Attributes.Builder attributes = Attributes.newBuilder()
         .set(XdsAttributes.ATTR_LOCALITY, locality)
         // Unique but arbitrary string
-        .set(XdsAttributes.ATTR_LOCALITY_NAME, locality.toString());
+        .set(LoadBalancer.ATTR_LOCALITY_NAME, locality.toString());
     if (authorityHostname != null) {
       attributes.set(XdsAttributes.ATTR_ADDRESS_NAME, authorityHostname);
     }