feat: add the missing xds.authority label

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

@@ -90,7 +90,7 @@ final class XdsClientMetricReporterImpl implements XdsClientMetricReporter {
         Arrays.asList("grpc.target", "grpc.xds.server"), Collections.emptyList(), false);
     RESOURCES_GAUGE = metricInstrumentRegistry.registerLongGauge("grpc.xds_client.resources",
         "EXPERIMENTAL.  Number of xDS resources.", "{resource}",
-        Arrays.asList("grpc.target", "grpc.xds.cache_state",
+        Arrays.asList("grpc.target", "grpc.xds.authority", "grpc.xds.cache_state",
             "grpc.xds.resource_type"), Collections.emptyList(), false);
   }
 
@@ -161,15 +161,32 @@ private void computeAndReportResourceCounts(
     for (Map.Entry<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByTypeEntry :
         metadataByType.entrySet()) {
       XdsResourceType<?> type = metadataByTypeEntry.getKey();
+      Map<String, ResourceMetadata> resources = metadataByTypeEntry.getValue();
 
-      Map<String, Long> resourceCountsByState = new HashMap<>();
-      for (ResourceMetadata metadata : metadataByTypeEntry.getValue().values()) {
+      Map<String, Map<String, Long>> resourceCountsByAuthorityAndState = new HashMap<>();
+      for (Map.Entry<String, ResourceMetadata> resourceEntry : resources.entrySet()) {
+        String resourceName = resourceEntry.getKey();
+        ResourceMetadata metadata = resourceEntry.getValue();
+        String authority = XdsClient.getAuthorityFromResourceName(resourceName);
         String cacheState = cacheStateFromResourceStatus(metadata.getStatus(), metadata.isCached());
-        resourceCountsByState.compute(cacheState, (k, v) -> (v == null) ? 1 : v + 1);
+        resourceCountsByAuthorityAndState
+            .computeIfAbsent(authority, k -> new HashMap<>())
+            .merge(cacheState, 1L, Long::sum);
       }
 
-      resourceCountsByState.forEach((cacheState, count) ->
-          callback.reportResourceCountGauge(count, cacheState, type.typeUrl()));
+      // Report metrics
+      for (Map.Entry<String, Map<String, Long>> authorityEntry
+            : resourceCountsByAuthorityAndState.entrySet()) {
+        String authority = authorityEntry.getKey();
+        Map<String, Long> stateCounts = authorityEntry.getValue();
+
+        for (Map.Entry<String, Long> stateEntry : stateCounts.entrySet()) {
+          String cacheState = stateEntry.getKey();
+          Long count = stateEntry.getValue();
+
+          callback.reportResourceCountGauge(count, authority, cacheState, type.typeUrl());
+        }
+      }
     }
   }
 
@@ -199,11 +216,12 @@ static final class MetricReporterCallback implements ServerConnectionCallback {
       this.target = target;
     }
 
-    // TODO(dnvindhya): include the "authority" label once xds.authority is available.
-    void reportResourceCountGauge(long resourceCount, String cacheState,
+    void reportResourceCountGauge(long resourceCount, String authority, String cacheState,
         String resourceType) {
+      // authority = #old, for non-xdstp resource names
       recorder.recordLongGauge(RESOURCES_GAUGE, resourceCount,
-          Arrays.asList(target, cacheState, resourceType), Collections.emptyList());
+          Arrays.asList(target, authority == null ? "#old" : authority, cacheState, resourceType),
+          Collections.emptyList());
     }
 
     @Override

xds/src/main/java/io/grpc/xds/client/XdsClient.java

@@ -118,6 +118,23 @@ public static String percentEncodePath(String input) {
     return Joiner.on('/').join(encodedSegs);
   }
 
+  /**
+   * Returns the authority from the resource name.
+   */
+  public static String getAuthorityFromResourceName(String resourceNames) {
+    String authority;
+    if (resourceNames.startsWith(XDSTP_SCHEME)) {
+      URI uri = URI.create(resourceNames);
+      authority = uri.getAuthority();
+      if (authority == null) {
+        authority = "";
+      }
+    } else {
+      authority = null;
+    }
+    return authority;
+  }
+
   public interface ResourceUpdate {}
 
   /**

xds/src/main/java/io/grpc/xds/client/XdsClientImpl.java

@@ -18,7 +18,6 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
-import static io.grpc.xds.client.Bootstrapper.XDSTP_SCHEME;
 import static io.grpc.xds.client.XdsResourceType.ParsedResource;
 import static io.grpc.xds.client.XdsResourceType.ValidatedResourceUpdate;
 
@@ -43,7 +42,6 @@
 import io.grpc.xds.client.XdsClient.ResourceStore;
 import io.grpc.xds.client.XdsLogger.XdsLogLevel;
 import java.io.IOException;
-import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -530,21 +528,6 @@ public Map<ServerInfo, LoadReportClient> getServerLrsClientMap() {
     return ImmutableMap.copyOf(serverLrsClientMap);
   }
 
-  private String getAuthority(String resource) {
-    String authority;
-    if (resource.startsWith(XDSTP_SCHEME)) {
-      URI uri = URI.create(resource);
-      authority = uri.getAuthority();
-      if (authority == null) {
-        authority = "";
-      }
-    } else {
-      authority = null;
-    }
-
-    return authority;
-  }
-
   @Nullable
   private ImmutableList<ServerInfo> getServerInfos(String authority) {
     if (authority != null) {
@@ -698,7 +681,7 @@ private final class ResourceSubscriber<T extends ResourceUpdate> {
       syncContext.throwIfNotInThisSynchronizationContext();
       this.type = type;
       this.resource = resource;
-      this.authority = getAuthority(resource);
+      this.authority = getAuthorityFromResourceName(resource);
       if (getServerInfos(authority) == null) {
         this.errorDescription = "Wrong configuration: xds server does not exist for resource "
             + resource;

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

@@ -76,6 +76,7 @@
 public class XdsClientMetricReporterImplTest {
 
   private static final String target = "test-target";
+  private static final String authority = "test-authority";
   private static final String server = "trafficdirector.googleapis.com";
   private static final String resourceTypeUrl =
       "resourceTypeUrl.googleapis.com/envoy.config.cluster.v3.Cluster";
@@ -101,7 +102,6 @@ public void setUp() {
 
   @Test
   public void reportResourceUpdates() {
-    // TODO(dnvindhya): add the "authority" label once available.
     reporter.reportResourceUpdates(10, 5, server, resourceTypeUrl);
     verify(mockMetricRecorder).addLongCounter(
         eqMetricInstrumentName("grpc.xds_client.resource_updates_valid"), eq((long) 10),
@@ -175,8 +175,8 @@ public void setXdsClient_reportCallbackMetrics_resourceCountsFails() {
   public void metricGauges() {
     SettableFuture<Void> future = SettableFuture.create();
     future.set(null);
-    when(mockXdsClient.getSubscribedResourcesMetadataSnapshot()).thenReturn(Futures.immediateFuture(
-        ImmutableMap.of()));
+    when(mockXdsClient.getSubscribedResourcesMetadataSnapshot())
+        .thenReturn(Futures.immediateFuture(ImmutableMap.of()));
     when(mockXdsClient.reportServerConnections(any(ServerConnectionCallback.class)))
         .thenReturn(future);
     reporter.setXdsClient(mockXdsClient);
@@ -199,13 +199,15 @@ public void metricGauges() {
 
     // Verify that reportResourceCounts and reportServerConnections were called
     // with the captured callback
-    callback.reportResourceCountGauge(10, "acked", resourceTypeUrl);
+    callback.reportResourceCountGauge(10, "MrPotatoHead",
+        "acked", resourceTypeUrl);
     inOrder.verify(mockBatchRecorder)
         .recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"), eq(10L), any(),
             any());
     callback.reportServerConnectionGauge(true, "xdsServer");
     inOrder.verify(mockBatchRecorder)
-        .recordLongGauge(eqMetricInstrumentName("grpc.xds_client.connected"), eq(1L), any(), any());
+        .recordLongGauge(eqMetricInstrumentName("grpc.xds_client.connected"),
+            eq(1L), any(), any());
 
     inOrder.verifyNoMoreInteractions();
   }
@@ -222,10 +224,10 @@ public void metricReporterCallback() {
         eq(Lists.newArrayList()));
 
     String cacheState = "requested";
-    callback.reportResourceCountGauge(10, cacheState, resourceTypeUrl);
+    callback.reportResourceCountGauge(10, authority, cacheState, resourceTypeUrl);
     verify(mockBatchRecorder, times(1)).recordLongGauge(
         eqMetricInstrumentName("grpc.xds_client.resources"), eq(10L),
-        eq(Arrays.asList(target, cacheState, resourceTypeUrl)),
+        eq(Arrays.asList(target, authority, cacheState, resourceTypeUrl)),
         eq(Collections.emptyList()));
   }
 
@@ -236,32 +238,31 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
     XdsResourceType<?> routeConfigResource = XdsRouteConfigureResource.getInstance();
     XdsResourceType<?> clusterResource = XdsClusterResource.getInstance();
 
-    Any rawListener =
-        Any.pack(Listener.newBuilder().setName("listener.googleapis.com").build());
+    Any rawListener = Any.pack(Listener.newBuilder().setName("listener.googleapis.com").build());
     long nanosLastUpdate = 1577923199_606042047L;
 
     Map<String, ResourceMetadata> ldsResourceMetadataMap = new HashMap<>();
-    ldsResourceMetadataMap.put("resource1",
+    ldsResourceMetadataMap.put("xdstp://authority1",
         ResourceMetadata.newResourceMetadataRequested());
-    ResourceMetadata ackedLdsResource = ResourceMetadata.newResourceMetadataAcked(rawListener, "42",
-        nanosLastUpdate);
+    ResourceMetadata ackedLdsResource =
+        ResourceMetadata.newResourceMetadataAcked(rawListener, "42", nanosLastUpdate);
     ldsResourceMetadataMap.put("resource2", ackedLdsResource);
     ldsResourceMetadataMap.put("resource3",
         ResourceMetadata.newResourceMetadataAcked(rawListener, "43", nanosLastUpdate));
-    ldsResourceMetadataMap.put("resource4",
-        ResourceMetadata.newResourceMetadataNacked(ackedLdsResource, "44", nanosLastUpdate,
-            "nacked after previous ack", true));
+    ldsResourceMetadataMap.put("xdstp:/no_authority",
+        ResourceMetadata.newResourceMetadataNacked(ackedLdsResource, "44",
+            nanosLastUpdate, "nacked after previous ack", true));
 
     Map<String, ResourceMetadata> rdsResourceMetadataMap = new HashMap<>();
     ResourceMetadata requestedRdsResourceMetadata = ResourceMetadata.newResourceMetadataRequested();
-    rdsResourceMetadataMap.put("resource5",
+    rdsResourceMetadataMap.put("xdstp://authority5",
         ResourceMetadata.newResourceMetadataNacked(requestedRdsResourceMetadata, "24",
             nanosLastUpdate, "nacked after request", false));
-    rdsResourceMetadataMap.put("resource6",
+    rdsResourceMetadataMap.put("xdstp://authority6",
         ResourceMetadata.newResourceMetadataDoesNotExist());
 
     Map<String, ResourceMetadata> cdsResourceMetadataMap = new HashMap<>();
-    cdsResourceMetadataMap.put("resource7", ResourceMetadata.newResourceMetadataUnknown());
+    cdsResourceMetadataMap.put("xdstp://authority7", ResourceMetadata.newResourceMetadataUnknown());
 
     metadataByType.put(listenerResource, ldsResourceMetadataMap);
     metadataByType.put(routeConfigResource, rdsResourceMetadataMap);
@@ -281,24 +282,34 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
 
     // LDS resource requested
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(1L), eq(Arrays.asList(target, "requested", listenerResource.typeUrl())), any());
+        eq(1L),
+        eq(Arrays.asList(target, "authority1", "requested", listenerResource.typeUrl())), any());
     // LDS resources acked
+    // authority = #old, for non-xdstp resource names
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(2L), eq(Arrays.asList(target, "acked", listenerResource.typeUrl())), any());
+        eq(2L),
+        eq(Arrays.asList(target, "#old", "acked", listenerResource.typeUrl())), any());
     // LDS resource nacked but cached
+    // "" for missing authority in the resource name
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(1L), eq(Arrays.asList(target, "nacked_but_cached", listenerResource.typeUrl())), any());
+        eq(1L),
+        eq(Arrays.asList(target, "", "nacked_but_cached", listenerResource.typeUrl())), any());
 
     // RDS resource nacked
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(1L), eq(Arrays.asList(target, "nacked", routeConfigResource.typeUrl())), any());
+        eq(1L),
+        eq(Arrays.asList(target, "authority5", "nacked", routeConfigResource.typeUrl())), any());
     // RDS resource does not exist
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(1L), eq(Arrays.asList(target, "does_not_exist", routeConfigResource.typeUrl())), any());
+        eq(1L),
+        eq(Arrays.asList(target, "authority6", "does_not_exist", routeConfigResource.typeUrl())),
+        any());
 
     // CDS resource unknown
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(1L), eq(Arrays.asList(target, "unknown", clusterResource.typeUrl())), any());
+        eq(1L),
+        eq(Arrays.asList(target, "authority7", "unknown", clusterResource.typeUrl())),
+        any());
     verifyNoMoreInteractions(mockBatchRecorder);
   }