1. xds: resolves data race

2. xds: considers authority as a separate dimension

Author: AgraVator

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

@@ -34,13 +34,15 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
 /**
@@ -143,13 +145,15 @@ void reportCallbackMetrics(BatchRecorder recorder, XdsClient xdsClient) {
       Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType =
           getResourceMetadataCompleted.get(10, TimeUnit.SECONDS);
 
-      ListenableFuture<Map<XdsResourceType<?>, Map<String, String>>>
-              getResourceAuthorityCompleted = xdsClient.getSubscribedResourcesAuthoritySnapshot();
+      List<String> resourceNames = metadataByType.values()
+          .stream()
+          .flatMap(innerMap -> innerMap.keySet().stream())
+          .collect(Collectors.toList());
 
-      Map<XdsResourceType<?>, Map<String, String>> authorityByType =
-              getResourceAuthorityCompleted.get(10, TimeUnit.SECONDS);
+      Map<String, String> resourceNameToAuthority =
+          xdsClient.getResourceNameToAuthorityMap(resourceNames);
 
-      computeAndReportResourceCounts(metadataByType, authorityByType, callback);
+      computeAndReportResourceCounts(metadataByType, resourceNameToAuthority, callback);
 
       // Normally this shouldn't take long, but adding a timeout to avoid indefinite blocking
       Void unused = reportServerConnectionsCompleted.get(5, TimeUnit.SECONDS);
@@ -163,27 +167,37 @@ void reportCallbackMetrics(BatchRecorder recorder, XdsClient xdsClient) {
 
   private void computeAndReportResourceCounts(
       Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType,
-      Map<XdsResourceType<?>, Map<String, String>> authorityByType,
+      Map<String, String> resourceNameToAuthority,
       MetricReporterCallback callback) {
     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<>();
-      Map<String, String> authorityByState = new HashMap<>();
-      for (Map.Entry<String, ResourceMetadata> metadataByName :
-              metadataByTypeEntry.getValue().entrySet()) {
-        String resourceName = metadataByName.getKey();
-        ResourceMetadata metadata = metadataByName.getValue();
+      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 = resourceNameToAuthority.getOrDefault(resourceName, "");
         String cacheState = cacheStateFromResourceStatus(metadata.getStatus(), metadata.isCached());
-        resourceCountsByState.compute(cacheState, (k, v) -> (v == null) ? 1 : v + 1);
-        authorityByState.put(cacheState, authorityByType.get(type).get(resourceName));
+        resourceCountsByAuthorityAndState
+            .computeIfAbsent(authority, k -> new HashMap<>())
+            .merge(cacheState, 1L, Long::sum);
       }
 
-      resourceCountsByState.forEach((cacheState, count) -> {
-        callback.reportResourceCountGauge(authorityByState.get(cacheState),
-            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());
+        }
+      }
     }
   }
 
@@ -213,12 +227,11 @@ static final class MetricReporterCallback implements ServerConnectionCallback {
       this.target = target;
     }
 
-    // TODO(dnvindhya): include the "authority" label once xds.authority is available.
     void reportResourceCountGauge(String authority, long resourceCount, String cacheState,
         String resourceType) {
       recorder.recordLongGauge(RESOURCES_GAUGE, resourceCount,
-          Arrays.asList(target, authority == null ? "#old" : authority,
-                  cacheState, resourceType), Collections.emptyList());
+          Arrays.asList(target, authority.isEmpty() ? "#old" : authority,
+          cacheState, resourceType), Collections.emptyList());
     }
 
     @Override

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

@@ -319,19 +319,6 @@ public Object getSecurityConfig() {
     throw new UnsupportedOperationException();
   }
 
-  /**
-   * Returns a {@link ListenableFuture} to the snapshot of the subscribed resources as
-   * they are at the moment of the call.
-   *
-   * <p>The snapshot is a map from the "resource type" to
-   * a map ("resource name": "authority").
-   */
-  // Must be synchronized.
-  public ListenableFuture<Map<XdsResourceType<?>, Map<String, String>>>
-      getSubscribedResourcesAuthoritySnapshot() {
-    throw new UnsupportedOperationException();
-  }
-
   /**
    * Registers a data watcher for the given Xds resource.
    */
@@ -391,6 +378,13 @@ public Map<Bootstrapper.ServerInfo, LoadReportClient> getServerLrsClientMap() {
     throw new UnsupportedOperationException();
   }
 
+  /**
+   * Returns a map of resource names to the authority.
+   */
+  public Map<String, String> getResourceNameToAuthorityMap(List<String> resourceNames) {
+    throw new UnsupportedOperationException();
+  }
+
   /** 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

@@ -57,6 +57,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
@@ -242,33 +243,6 @@ public void run() {
     return future;
   }
 
-  // As XdsClient APIs becomes resource agnostic, subscribed resource types are dynamic.
-  // ResourceTypes that do not have subscribers does not show up in the snapshot keys.
-  @Override
-  public ListenableFuture<Map<XdsResourceType<?>, Map<String, String>>>
-      getSubscribedResourcesAuthoritySnapshot() {
-    final SettableFuture<Map<XdsResourceType<?>, Map<String, String>>> future =
-            SettableFuture.create();
-    syncContext.execute(new Runnable() {
-      @Override
-      public void run() {
-        // A map from a "resource type" to a map ("resource name": "authority")
-        ImmutableMap.Builder<XdsResourceType<?>, Map<String, String>> authoritySnapshot =
-                ImmutableMap.builder();
-        for (XdsResourceType<?> resourceType : resourceSubscribers.keySet()) {
-          ImmutableMap.Builder<String, String> authorityMap = ImmutableMap.builder();
-          for (Map.Entry<String, ResourceSubscriber<? extends ResourceUpdate>> resourceEntry
-                  : resourceSubscribers.get(resourceType).entrySet()) {
-            authorityMap.put(resourceEntry.getKey(), resourceEntry.getValue().authority);
-          }
-          authoritySnapshot.put(resourceType, authorityMap.buildOrThrow());
-        }
-        future.set(authoritySnapshot.buildOrThrow());
-      }
-    });
-    return future;
-  }
-
   @Override
   public Object getSecurityConfig() {
     return securityConfig;
@@ -572,6 +546,17 @@ private String getAuthority(String resource) {
     return authority;
   }
 
+  @Override
+  public Map<String, String> getResourceNameToAuthorityMap(List<String> resourceNames) {
+    if (resourceNames == null || resourceNames.isEmpty()) {
+      return Collections.emptyMap();
+    }
+    return resourceNames.stream()
+        .collect(Collectors.toMap(
+            Function.identity(),
+            this::getAuthority));
+  }
+
   @Nullable
   private ImmutableList<ServerInfo> getServerInfos(String authority) {
     if (authority != null) {

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

@@ -19,6 +19,7 @@
 import static com.google.common.truth.Truth.assertThat;
 import static org.mockito.AdditionalAnswers.delegatesTo;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.inOrder;
@@ -129,8 +130,6 @@ public void setXdsClient_reportMetrics() throws Exception {
     future.set(null);
     when(mockXdsClient.getSubscribedResourcesMetadataSnapshot()).thenReturn(Futures.immediateFuture(
         ImmutableMap.of()));
-    when(mockXdsClient.getSubscribedResourcesAuthoritySnapshot())
-        .thenReturn(Futures.immediateFuture(ImmutableMap.of()));
     when(mockXdsClient.reportServerConnections(any(ServerConnectionCallback.class)))
         .thenReturn(future);
     reporter.setXdsClient(mockXdsClient);
@@ -152,8 +151,6 @@ public void setXdsClient_reportCallbackMetrics_resourceCountsFails() {
     future.set(null);
     when(mockXdsClient.getSubscribedResourcesMetadataSnapshot()).thenReturn(Futures.immediateFuture(
         ImmutableMap.of()));
-    when(mockXdsClient.getSubscribedResourcesAuthoritySnapshot())
-            .thenReturn(Futures.immediateFuture(ImmutableMap.of()));
 
     // Create a future that will throw an exception
     SettableFuture<Void> serverConnectionsFeature = SettableFuture.create();
@@ -181,8 +178,6 @@ public void metricGauges() {
     future.set(null);
     when(mockXdsClient.getSubscribedResourcesMetadataSnapshot()).thenReturn(Futures.immediateFuture(
         ImmutableMap.of()));
-    when(mockXdsClient.getSubscribedResourcesAuthoritySnapshot())
-        .thenReturn(Futures.immediateFuture(ImmutableMap.of()));
     when(mockXdsClient.reportServerConnections(any(ServerConnectionCallback.class)))
         .thenReturn(future);
     reporter.setXdsClient(mockXdsClient);
@@ -238,7 +233,7 @@ public void metricReporterCallback() {
   @Test
   public void reportCallbackMetrics_computeAndReportResourceCounts() {
     Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType = new HashMap<>();
-    Map<XdsResourceType<?>, Map<String, String>> authorityByType = new HashMap<>();
+    Map<String, String> resourceNameByAuthority = new HashMap<>();
     XdsResourceType<?> listenerResource = XdsListenerResource.getInstance();
     XdsResourceType<?> routeConfigResource = XdsRouteConfigureResource.getInstance();
     XdsResourceType<?> clusterResource = XdsClusterResource.getInstance();
@@ -248,44 +243,37 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
     long nanosLastUpdate = 1577923199_606042047L;
 
     Map<String, ResourceMetadata> ldsResourceMetadataMap = new HashMap<>();
-    Map<String, String> ldsAuthorityMap = new HashMap<>();
     ldsResourceMetadataMap.put("resource1",
         ResourceMetadata.newResourceMetadataRequested());
-    ldsAuthorityMap.put("resource1", "authority1");
+    resourceNameByAuthority.put("resource1", "authority1");
     ResourceMetadata ackedLdsResource = ResourceMetadata.newResourceMetadataAcked(rawListener, "42",
         nanosLastUpdate);
     ldsResourceMetadataMap.put("resource2", ackedLdsResource);
-    ldsAuthorityMap.put("resource2", "authority2");
+    resourceNameByAuthority.put("resource2", "authority2");
     ldsResourceMetadataMap.put("resource3",
         ResourceMetadata.newResourceMetadataAcked(rawListener, "43", nanosLastUpdate));
-    ldsAuthorityMap.put("resource3", "authority3");
+    resourceNameByAuthority.put("resource3", "authority2");
     ldsResourceMetadataMap.put("resource4",
         ResourceMetadata.newResourceMetadataNacked(ackedLdsResource, "44", nanosLastUpdate,
             "nacked after previous ack", true));
-    ldsAuthorityMap.put("resource4", "authority4");
+    resourceNameByAuthority.put("resource4", "authority4");
 
     Map<String, ResourceMetadata> rdsResourceMetadataMap = new HashMap<>();
-    Map<String, String> rdsAuthorityMap = new HashMap<>();
     ResourceMetadata requestedRdsResourceMetadata = ResourceMetadata.newResourceMetadataRequested();
     rdsResourceMetadataMap.put("resource5",
         ResourceMetadata.newResourceMetadataNacked(requestedRdsResourceMetadata, "24",
             nanosLastUpdate, "nacked after request", false));
-    rdsAuthorityMap.put("resource5", "authority5");
+    resourceNameByAuthority.put("resource5", "authority5");
     rdsResourceMetadataMap.put("resource6",
         ResourceMetadata.newResourceMetadataDoesNotExist());
-    rdsAuthorityMap.put("resource6", "authority6");
+    resourceNameByAuthority.put("resource6", "authority6");
 
     Map<String, ResourceMetadata> cdsResourceMetadataMap = new HashMap<>();
-    Map<String, String> cdsAuthorityMap = new HashMap<>();
     cdsResourceMetadataMap.put("resource7", ResourceMetadata.newResourceMetadataUnknown());
-    cdsAuthorityMap.put("resource7", "authority7");
 
     metadataByType.put(listenerResource, ldsResourceMetadataMap);
-    authorityByType.put(listenerResource, ldsAuthorityMap);
     metadataByType.put(routeConfigResource, rdsResourceMetadataMap);
-    authorityByType.put(routeConfigResource, rdsAuthorityMap);
     metadataByType.put(clusterResource, cdsResourceMetadataMap);
-    authorityByType.put(clusterResource, cdsAuthorityMap);
 
     SettableFuture<Void> reportServerConnectionsCompleted = SettableFuture.create();
     reportServerConnectionsCompleted.set(null);
@@ -297,10 +285,8 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
     when(mockXdsClient.getSubscribedResourcesMetadataSnapshot())
         .thenReturn(getResourceMetadataCompleted);
 
-    ListenableFuture<Map<XdsResourceType<?>, Map<String, String>>>
-            getResourceAuthorityCompleted = Futures.immediateFuture(authorityByType);
-    when(mockXdsClient.getSubscribedResourcesAuthoritySnapshot())
-            .thenReturn(getResourceAuthorityCompleted);
+    when(mockXdsClient.getResourceNameToAuthorityMap(anyList()))
+            .thenReturn(resourceNameByAuthority);
 
     reporter.reportCallbackMetrics(mockBatchRecorder, mockXdsClient);
 
@@ -310,7 +296,7 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
             "requested", listenerResource.typeUrl())), any());
     // LDS resources acked
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(2L), eq(Arrays.asList(target, "authority3",
+        eq(2L), eq(Arrays.asList(target, "authority2",
             "acked", listenerResource.typeUrl())), any());
     // LDS resource nacked but cached
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
@@ -328,18 +314,16 @@ public void reportCallbackMetrics_computeAndReportResourceCounts() {
 
     // CDS resource unknown
     verify(mockBatchRecorder).recordLongGauge(eqMetricInstrumentName("grpc.xds_client.resources"),
-        eq(1L), eq(Arrays.asList(target, "authority7",
+        eq(1L), eq(Arrays.asList(target, "#old",
             "unknown", clusterResource.typeUrl())), any());
     verifyNoMoreInteractions(mockBatchRecorder);
   }
 
   @Test
   public void reportCallbackMetrics_computeAndReportResourceCounts_emptyResources() {
     Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType = new HashMap<>();
-    Map<XdsResourceType<?>, Map<String, String>> authorityByType = new HashMap<>();
     XdsResourceType<?> listenerResource = XdsListenerResource.getInstance();
     metadataByType.put(listenerResource, Collections.emptyMap());
-    authorityByType.put(listenerResource, Collections.emptyMap());
 
     SettableFuture<Void> reportServerConnectionsCompleted = SettableFuture.create();
     reportServerConnectionsCompleted.set(null);
@@ -351,11 +335,6 @@ public void reportCallbackMetrics_computeAndReportResourceCounts_emptyResources(
     when(mockXdsClient.getSubscribedResourcesMetadataSnapshot())
         .thenReturn(getResourceMetadataCompleted);
 
-    ListenableFuture<Map<XdsResourceType<?>, Map<String, String>>>
-            getAuthorityCompleted = Futures.immediateFuture(authorityByType);
-    when(mockXdsClient.getSubscribedResourcesAuthoritySnapshot())
-            .thenReturn(getAuthorityCompleted);
-
     reporter.reportCallbackMetrics(mockBatchRecorder, mockXdsClient);
 
     // Verify that reportResourceCountGauge is never called