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(authority, count, cacheState, type.typeUrl());
+        }
+      }
     }
   }
 
@@ -199,11 +216,11 @@ 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(String authority, long resourceCount, String cacheState,
         String resourceType) {
       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

@@ -378,6 +378,23 @@ public Map<Bootstrapper.ServerInfo, LoadReportClient> getServerLrsClientMap() {
     throw new UnsupportedOperationException();
   }
 
+  /**
+   * 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;
+  }
+
   /** Callback used to report a gauge metric value for server connections. */
   public interface ServerConnectionCallback {
     void reportServerConnectionGauge(boolean isConnected, String xdsServer);

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),
@@ -199,7 +199,7 @@ public void metricGauges() {
 
     // Verify that reportResourceCounts and reportServerConnections were called
     // with the captured callback
-    callback.reportResourceCountGauge(10, "acked", resourceTypeUrl);
+    callback.reportResourceCountGauge("PotatoHead", 10, "acked", resourceTypeUrl);
     inOrder.verify(mockBatchRecorder)
         .recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"), eq(10L), any(),
             any());
@@ -222,10 +222,10 @@ public void metricReporterCallback() {
         eq(Lists.newArrayList()));
 
     String cacheState = "requested";
-    callback.reportResourceCountGauge(10, cacheState, resourceTypeUrl);
+    callback.reportResourceCountGauge(authority, 10, 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()));
   }
 
@@ -237,31 +237,31 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
     XdsResourceType<?> clusterResource = XdsClusterResource.getInstance();
 
     Any rawListener =
-        Any.pack(Listener.newBuilder().setName("listener.googleapis.com").build());
+            Any.pack(Listener.newBuilder().setName("listener.googleapis.com").build());
     long nanosLastUpdate = 1577923199_606042047L;
 
     Map<String, ResourceMetadata> ldsResourceMetadataMap = new HashMap<>();
-    ldsResourceMetadataMap.put("resource1",
-        ResourceMetadata.newResourceMetadataRequested());
+    ldsResourceMetadataMap.put("xdstp://authority1",
+            ResourceMetadata.newResourceMetadataRequested());
     ResourceMetadata ackedLdsResource = ResourceMetadata.newResourceMetadataAcked(rawListener, "42",
-        nanosLastUpdate);
+            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));
+            ResourceMetadata.newResourceMetadataAcked(rawListener, "43", nanosLastUpdate));
+    ldsResourceMetadataMap.put("xdstp:/need_this",
+            ResourceMetadata.newResourceMetadataNacked(ackedLdsResource, "44", nanosLastUpdate,
+                    "nacked after previous ack", true));
 
     Map<String, ResourceMetadata> rdsResourceMetadataMap = new HashMap<>();
     ResourceMetadata requestedRdsResourceMetadata = ResourceMetadata.newResourceMetadataRequested();
-    rdsResourceMetadataMap.put("resource5",
-        ResourceMetadata.newResourceMetadataNacked(requestedRdsResourceMetadata, "24",
-            nanosLastUpdate, "nacked after request", false));
-    rdsResourceMetadataMap.put("resource6",
-        ResourceMetadata.newResourceMetadataDoesNotExist());
+    rdsResourceMetadataMap.put("xdstp://authority5",
+            ResourceMetadata.newResourceMetadataNacked(requestedRdsResourceMetadata, "24",
+                    nanosLastUpdate, "nacked after request", false));
+    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);
@@ -270,35 +270,42 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
     SettableFuture<Void> reportServerConnectionsCompleted = SettableFuture.create();
     reportServerConnectionsCompleted.set(null);
     when(mockXdsClient.reportServerConnections(any(MetricReporterCallback.class)))
-        .thenReturn(reportServerConnectionsCompleted);
+            .thenReturn(reportServerConnectionsCompleted);
 
     ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
-        getResourceMetadataCompleted = Futures.immediateFuture(metadataByType);
+            getResourceMetadataCompleted = Futures.immediateFuture(metadataByType);
     when(mockXdsClient.getSubscribedResourcesMetadataSnapshot())
-        .thenReturn(getResourceMetadataCompleted);
+            .thenReturn(getResourceMetadataCompleted);
+
 
     reporter.reportCallbackMetrics(mockBatchRecorder, mockXdsClient);
 
     // 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
     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
     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);
   }