Inject value from XdsClientMetricReporterImpl and addressed review comments

Author: DNVindhya

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

@@ -52,6 +52,8 @@ final class SharedXdsClientPoolProvider implements XdsClientPoolFactory {
   private static final boolean LOG_XDS_NODE_ID = Boolean.parseBoolean(
       System.getenv("GRPC_LOG_XDS_NODE_ID"));
   private static final Logger log = Logger.getLogger(XdsClientImpl.class.getName());
+  private static final ExponentialBackoffPolicy.Provider BACKOFF_POLICY_PROVIDER =
+      new ExponentialBackoffPolicy.Provider();
 
   private final Bootstrapper bootstrapper;
   private final Object lock = new Object();
@@ -121,8 +123,6 @@ private static class SharedXdsClientPoolProviderHolder {
   @VisibleForTesting
   protected class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
 
-    private final ExponentialBackoffPolicy.Provider backoffPolicyProvider =
-        new ExponentialBackoffPolicy.Provider();
     private final BootstrapInfo bootstrapInfo;
     private final String target; // The target associated with the xDS client.
     private final XdsClientMetricReporterImpl xdsClientMetricReporter;
@@ -139,7 +139,7 @@ protected class RefCountedXdsClientObjectPool implements ObjectPool<XdsClient> {
         MetricRecorder metricRecorder) {
       this.bootstrapInfo = checkNotNull(bootstrapInfo);
       this.target = target;
-      this.xdsClientMetricReporter = new XdsClientMetricReporterImpl(metricRecorder);
+      this.xdsClientMetricReporter = new XdsClientMetricReporterImpl(metricRecorder, target);
     }
 
     @Override
@@ -154,12 +154,11 @@ public XdsClient getObject() {
               DEFAULT_XDS_TRANSPORT_FACTORY,
               bootstrapInfo,
               scheduler,
-              backoffPolicyProvider,
+              BACKOFF_POLICY_PROVIDER,
               GrpcUtil.STOPWATCH_SUPPLIER,
               TimeProvider.SYSTEM_TIME_PROVIDER,
               MessagePrinter.INSTANCE,
               new TlsContextManagerImpl(bootstrapInfo),
-              getTarget(),
               xdsClientMetricReporter);
           xdsClientMetricReporter.setXdsClient(xdsClient);
         }

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

@@ -52,6 +52,7 @@ public class XdsClientMetricReporterImpl implements XdsClientMetricReporter {
   private static final LongGaugeMetricInstrument RESOURCES_GAUGE;
 
   private final MetricRecorder metricRecorder;
+  private final String target;
   @Nullable
   private Registration gaugeRegistration = null;
   @Nullable
@@ -90,21 +91,22 @@ public class XdsClientMetricReporterImpl implements XdsClientMetricReporter {
             "grpc.xds.resource_type"), Collections.emptyList(), false);
   }
 
-  XdsClientMetricReporterImpl(MetricRecorder metricRecorder) {
+  XdsClientMetricReporterImpl(MetricRecorder metricRecorder, String target) {
     this.metricRecorder = metricRecorder;
+    this.target = target;
   }
 
   @Override
   public void reportResourceUpdates(long validResourceCount, long invalidResourceCount,
-      String target, String xdsServer, String resourceType) {
+      String xdsServer, String resourceType) {
     metricRecorder.addLongCounter(RESOURCE_UPDATES_VALID_COUNTER, validResourceCount,
         Arrays.asList(target, xdsServer, resourceType), Collections.emptyList());
     metricRecorder.addLongCounter(RESOURCE_UPDATES_INVALID_COUNTER, invalidResourceCount,
         Arrays.asList(target, xdsServer, resourceType), Collections.emptyList());
   }
 
   @Override
-  public void reportServerFailure(long serverFailure, String target, String xdsServer) {
+  public void reportServerFailure(long serverFailure, String xdsServer) {
     metricRecorder.addLongCounter(SERVER_FAILURE_COUNTER, serverFailure,
         Arrays.asList(target, xdsServer), Collections.emptyList());
   }
@@ -127,7 +129,7 @@ void close() {
   }
 
   void reportCallbackMetrics(BatchRecorder recorder) {
-    MetricReporterCallback callback = new MetricReporterCallback(recorder);
+    MetricReporterCallback callback = new MetricReporterCallback(recorder, target);
     try {
       SettableFuture<Void> reportResourceCountsCompleted = this.xdsClient.reportResourceCounts(
           callback);
@@ -148,21 +150,23 @@ void reportCallbackMetrics(BatchRecorder recorder) {
   static final class MetricReporterCallback implements ResourceCallback,
       ServerConnectionCallback {
     private final BatchRecorder recorder;
+    private final String target;
 
-    MetricReporterCallback(BatchRecorder recorder) {
+    MetricReporterCallback(BatchRecorder recorder, String target) {
       this.recorder = recorder;
+      this.target = target;
     }
 
-    // TODO(@dnvindhya): include the "authority" label once xds.authority is available.
+    // TODO(dnvindhya): include the "authority" label once xds.authority is available.
     @Override
-    public void reportResourceCountGauge(long resourceCount, String cacheState, String resourceType,
-        String target) {
+    public void reportResourceCountGauge(long resourceCount, String cacheState,
+        String resourceType) {
       recorder.recordLongGauge(RESOURCES_GAUGE, resourceCount,
           Arrays.asList(target, cacheState, resourceType), Collections.emptyList());
     }
 
     @Override
-    public void reportServerConnectionGauge(boolean isConnected, String target, String xdsServer) {
+    public void reportServerConnectionGauge(boolean isConnected, String xdsServer) {
       recorder.recordLongGauge(CONNECTED_GAUGE, isConnected ? 1 : 0,
           Arrays.asList(target, xdsServer), Collections.emptyList());
     }

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

@@ -126,9 +126,9 @@ final class XdsNameResolver extends NameResolver {
   private final ConcurrentMap<String, ClusterRefState> clusterRefs = new ConcurrentHashMap<>();
   private final ConfigSelector configSelector = new ConfigSelector();
   private final long randomChannelId;
+  private final MetricRecorder metricRecorder;
 
   private volatile RoutingConfig routingConfig = RoutingConfig.empty;
-  private volatile MetricRecorder metricRecorder;
   private Listener2 listener;
   private ObjectPool<XdsClient> xdsClientPool;
   private XdsClient xdsClient;

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

@@ -172,6 +172,8 @@ public void run() {
 
   private void internalStart() {
     try {
+      // TODO(dnvindhya): Add "#server" as "grpc.target" attribute value for
+      // xDS enabled servers.
       xdsClientPool = xdsClientPoolFactory.getOrCreate("", new MetricRecorder() {});
     } catch (Exception e) {
       StatusException statusException = Status.UNAVAILABLE.withDescription(

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

@@ -381,14 +381,13 @@ public Map<Bootstrapper.ServerInfo, LoadReportClient> getServerLrsClientMap() {
 
   /** Callback used to report gauge metric value for resources. */
   public interface ResourceCallback {
-    // TODO(@dnvindhya): include the "authority" label once xds.authority is available.
-    void reportResourceCountGauge(long resourceCount, String cacheState, String resourceType,
-        String target);
+    // TODO(dnvindhya): include the "authority" label once xds.authority is available.
+    void reportResourceCountGauge(long resourceCount, String cacheState, String resourceType);
   }
 
   /** Callback used to report a gauge metric value for server connections. */
   public interface ServerConnectionCallback {
-    void reportServerConnectionGauge(boolean isConnected, String target, String xdsServer);
+    void reportServerConnectionGauge(boolean isConnected, String xdsServer);
   }
 
   /**

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

@@ -101,7 +101,6 @@ public void uncaughtException(Thread t, Throwable e) {
   private final XdsLogger logger;
   private volatile boolean isShutdown;
   private final MessagePrettyPrinter messagePrinter;
-  private final String target;
   private final XdsClientMetricReporter metricReporter;
 
   public XdsClientImpl(
@@ -113,7 +112,6 @@ public XdsClientImpl(
       TimeProvider timeProvider,
       MessagePrettyPrinter messagePrinter,
       Object securityConfig,
-      String target,
       XdsClientMetricReporter metricReporter) {
     this.xdsTransportFactory = xdsTransportFactory;
     this.bootstrapInfo = bootstrapInfo;
@@ -123,7 +121,6 @@ public XdsClientImpl(
     this.timeProvider = timeProvider;
     this.messagePrinter = messagePrinter;
     this.securityConfig = securityConfig;
-    this.target = target;
     this.metricReporter = metricReporter;
     logId = InternalLogId.allocate("xds-client", null);
     logger = XdsLogger.withLogId(logId);
@@ -148,7 +145,7 @@ private void handleStreamClosed(Status error, ServerInfo serverInfo) {
     syncContext.throwIfNotInThisSynchronizationContext();
     cleanUpResourceTimers();
     if (!error.isOk()) {
-      metricReporter.reportServerFailure(1L, target, serverInfo.target());
+      metricReporter.reportServerFailure(1L, serverInfo.target());
       for (Map<String, ResourceSubscriber<? extends ResourceUpdate>> subscriberMap :
           resourceSubscribers.values()) {
         for (ResourceSubscriber<? extends ResourceUpdate> subscriber : subscriberMap.values()) {
@@ -474,7 +471,7 @@ private <T extends ResourceUpdate> void handleResourceUpdate(
     Map<String, ParsedResource<T>> parsedResources = result.parsedResources;
     Set<String> invalidResources = result.invalidResources;
     metricReporter.reportResourceUpdates(Long.valueOf(parsedResources.size()),
-        Long.valueOf(invalidResources.size()), target,
+        Long.valueOf(invalidResources.size()),
         args.getServerInfo().target(), xdsResourceType.typeUrl());
 
     List<String> errors = result.errors;
@@ -539,7 +536,7 @@ public SettableFuture<Void> reportServerConnections(ServerConnectionCallback cal
     syncContext.execute(() -> {
       serverCpClientMap.forEach((serverInfo, controlPlaneClient) ->
           callback.reportServerConnectionGauge(
-              controlPlaneClient.hasWorkingAdsStream(), target, serverInfo.target()));
+              controlPlaneClient.hasWorkingAdsStream(), serverInfo.target()));
       future.set(null);
     });
     return future;
@@ -549,9 +546,17 @@ public SettableFuture<Void> reportServerConnections(ServerConnectionCallback cal
   public SettableFuture<Void> reportResourceCounts(ResourceCallback callback) {
     SettableFuture<Void> future = SettableFuture.create();
     syncContext.execute(() -> {
-      Map<XdsResourceType<?>, Map<String, Long>> resourceCountsByType =
-          getResourceCountsByType();
-      reportResourceCountsToCallback(callback, resourceCountsByType);
+      for (XdsResourceType<? extends ResourceUpdate> resourceType : resourceSubscribers.keySet()) {
+        Map<String, Long> resourceCountsByState = new HashMap<>();
+        for (ResourceSubscriber<? extends ResourceUpdate> subscriber :
+            resourceSubscribers.get(resourceType).values()) {
+          String cacheState = cacheStateFromResourceStatus(subscriber.metadata,
+              subscriber.data != null);
+          resourceCountsByState.compute(cacheState, (k, v) -> (v == null) ? 1 : v + 1);
+        }
+        resourceCountsByState.forEach((cacheState, count) ->
+            callback.reportResourceCountGauge(count, cacheState, resourceType.typeUrl()));
+      }
       future.set(null);
     });
     return future;
@@ -563,36 +568,6 @@ private String cacheStateFromResourceStatus(ResourceMetadata metadata, boolean i
         ? status + "_but_cached" : status;
   }
 
-  /** Calculates number of resources by ResourceType and ResourceSubscriber.metadata.status. */
-  Map<XdsResourceType<?>, Map<String, Long>> getResourceCountsByType() {
-    Map<XdsResourceType<?>, Map<String, Long>> resourceCountsByType = new HashMap<>();
-    for (XdsResourceType<? extends ResourceUpdate> resourceType : resourceSubscribers.keySet()) {
-      Map<String, Long> resourceCountsByState = new HashMap<>();
-      for (ResourceSubscriber<? extends ResourceUpdate> subscriber :
-          resourceSubscribers.get(resourceType).values()) {
-        String cacheState = cacheStateFromResourceStatus(subscriber.metadata,
-            subscriber.data != null);
-        resourceCountsByState.compute(cacheState, (k, v) -> (v == null) ? 1 : v + 1);
-      }
-      resourceCountsByType.put(resourceType, resourceCountsByState);
-    }
-    return resourceCountsByType;
-  }
-
-  /** Reports resource counts using the provided ResourceCallback. */
-  void reportResourceCountsToCallback(ResourceCallback callback,
-      Map<XdsResourceType<?>, Map<String, Long>> resourceCountsByType) {
-    for (Map.Entry<XdsResourceType<?>, Map<String, Long>> entry :
-        resourceCountsByType.entrySet()) {
-      XdsResourceType<?> resourceType = entry.getKey();
-      Map<String, Long> resourceCountsByState = entry.getValue();
-      // TODO(@dnvindhya): include the "authority" label once authority is available here.
-      resourceCountsByState.forEach((cacheState, count) ->
-          callback.reportResourceCountGauge(count, cacheState, resourceType.typeUrl(), target));
-    }
-  }
-
-
   /** Tracks a single subscribed resource. */
   private final class ResourceSubscriber<T extends ResourceUpdate> {
     @Nullable private final ServerInfo serverInfo;

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

@@ -29,22 +29,20 @@ public interface XdsClientMetricReporter {
    *
    * @param validResourceCount Number of resources that were valid.
    * @param invalidResourceCount Number of resources that were invalid.
-   * @param target Target of the gRPC channel.
    * @param xdsServer Target URI of the xDS server with which the XdsClient is communicating.
    * @param resourceType Type of XDS resource (e.g., "envoy.config.listener.v3.Listener").
    */
   default void reportResourceUpdates(long validResourceCount, long invalidResourceCount,
-      String target, String xdsServer, String resourceType) {
+      String xdsServer, String resourceType) {
   }
 
   /**
    * Reports number of xDS servers going from healthy to unhealthy.
    *
    * @param serverFailure Number of xDS server failures.
-   * @param target Target of the gRPC channel.
    * @param xdsServer Target URI of the xDS server with which the XdsClient is communicating.
    */
-  default void reportServerFailure(long serverFailure, String target, String xdsServer) {
+  default void reportServerFailure(long serverFailure, String xdsServer) {
   }
 
 }