xds: add the missing xds.authority metric (#12018)

This completes the [XDS client metrics](https://github.com/grpc/proposal/blob/master/A78-grpc-metrics-wrr-pf-xds.md#xdsclient) by adding the remaining grpc.xds.authority metric.

Author: AgraVator

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);
  }