xds: Add counter and gauge metrics

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

@@ -18,8 +18,6 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import io.grpc.Internal;
 import io.grpc.LongCounterMetricInstrument;
 import io.grpc.LongGaugeMetricInstrument;
 import io.grpc.MetricInstrumentRegistry;
@@ -28,7 +26,6 @@
 import io.grpc.MetricRecorder.BatchRecorder;
 import io.grpc.MetricRecorder.Registration;
 import io.grpc.xds.client.XdsClient;
-import io.grpc.xds.client.XdsClient.ResourceCallback;
 import io.grpc.xds.client.XdsClient.ResourceMetadata;
 import io.grpc.xds.client.XdsClient.ResourceMetadata.ResourceMetadataStatus;
 import io.grpc.xds.client.XdsClient.ServerConnectionCallback;
@@ -37,9 +34,9 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.Locale;
 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;
@@ -49,7 +46,6 @@
 /**
  * XdsClientMetricReporter implementation.
  */
-@Internal
 final class XdsClientMetricReporterImpl implements XdsClientMetricReporter {
 
   private static final Logger logger = Logger.getLogger(
@@ -139,34 +135,29 @@
   void reportCallbackMetrics(BatchRecorder recorder, XdsClient xdsClient) {
     MetricReporterCallback callback = new MetricReporterCallback(recorder, target);
     try {
-      SettableFuture<Void> reportServerConnectionsCompleted =
-          xdsClient.reportServerConnections(callback);
+      Future<Void> reportServerConnectionsCompleted = xdsClient.reportServerConnections(callback);
 
       ListenableFuture<Map<XdsResourceType<?>, Map<String, ResourceMetadata>>>
           getResourceMetadataCompleted = xdsClient.getSubscribedResourcesMetadataSnapshot();
 
       Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType =
           getResourceMetadataCompleted.get(10, TimeUnit.SECONDS);
 
-      SettableFuture<Void> reportResourceCountsCompleted = computeAndReportResourceCounts(
-          metadataByType, callback);
+      computeAndReportResourceCounts(metadataByType, callback);
 
       // Normally this shouldn't take long, but adding a timeout to avoid indefinite blocking
-      Void unused1 = reportServerConnectionsCompleted.get(5, TimeUnit.SECONDS);
-      Void unused2 = reportResourceCountsCompleted.get(5, TimeUnit.SECONDS);
+      Void unused = reportServerConnectionsCompleted.get(5, TimeUnit.SECONDS);
     } catch (ExecutionException | TimeoutException | InterruptedException e) {
       if (e instanceof InterruptedException) {
         Thread.currentThread().interrupt(); // re-set the current thread's interruption state
       }
       logger.log(Level.WARNING, "Failed to report gauge metrics", e);
     }
   }
 
-  private SettableFuture<Void> computeAndReportResourceCounts(
+  private void computeAndReportResourceCounts(
       Map<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByType,
       MetricReporterCallback callback) {
-    SettableFuture<Void> future = SettableFuture.create();
-
     for (Map.Entry<XdsResourceType<?>, Map<String, ResourceMetadata>> metadataByTypeEntry :
         metadataByType.entrySet()) {
       XdsResourceType<?> type = metadataByTypeEntry.getKey();
@@ -180,20 +171,26 @@
       resourceCountsByState.forEach((cacheState, count) ->
           callback.reportResourceCountGauge(count, cacheState, type.typeUrl()));
     }
-    future.set(null);
-    return future;
   }
 
   private static String cacheStateFromResourceStatus(ResourceMetadataStatus metadataStatus,
       boolean isResourceCached) {
-    String status = metadataStatus.toString().toLowerCase(Locale.ROOT);
-    return metadataStatus == ResourceMetadataStatus.NACKED && isResourceCached
-        ? status + "_but_cached" : status;
+    switch (metadataStatus) {
+      case REQUESTED:
+        return "requested";
+      case DOES_NOT_EXIST:
+        return "does_not_exist";
+      case ACKED:
+        return "acked";
+      case NACKED:
+        return isResourceCached ? "nacked_but_cached" : "nacked";
+      default:
+        return "unknown";
+    }
   }
 
   @VisibleForTesting
-  static final class MetricReporterCallback implements ResourceCallback,
-      ServerConnectionCallback {
+  static final class MetricReporterCallback implements ServerConnectionCallback {
     private final BatchRecorder recorder;
     private final String target;
 
@@ -203,7 +200,6 @@
     }
 
     // TODO(dnvindhya): include the "authority" label once xds.authority is available.
-    @Override
     public void reportResourceCountGauge(long resourceCount, String cacheState,
         String resourceType) {
       recorder.recordLongGauge(RESOURCES_GAUGE, resourceCount,

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

@@ -24,7 +24,6 @@
 import com.google.common.net.UrlEscapers;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.SettableFuture;
 import com.google.protobuf.Any;
 import io.grpc.ExperimentalApi;
 import io.grpc.Status;
@@ -37,6 +36,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
 import javax.annotation.Nullable;
 
@@ -386,25 +386,21 @@
     throw new UnsupportedOperationException();
   }
 
-  /** 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);
-  }
-
   /** Callback used to report a gauge metric value for server connections. */
   public interface ServerConnectionCallback {
     void reportServerConnectionGauge(boolean isConnected, String xdsServer);
   }
 
   /**
-   * Reports whether xDS client has a "working" ADS stream to xDS server.
-   * The definition of a working stream is defined in gRFC A78.
-   * @see <a href="https://github.com/grpc/proposal/blob/master/A78-grpc-metrics-wrr-pf-xds.md#xdsclient">
-   *   A78-grpc-metrics-wrr-pf-xds.md</a>
+   * Reports whether xDS client has a "working" ADS stream to xDS server. The definition of a
+   * working stream is defined in gRFC A78.
+   *
+   * @see <a
+   *     href="https://github.com/grpc/proposal/blob/master/A78-grpc-metrics-wrr-pf-xds.md#xdsclient">
+   *     A78-grpc-metrics-wrr-pf-xds.md</a>
    */
-  public SettableFuture<Void> reportServerConnections(ServerConnectionCallback callback) {
+  public Future<Void> reportServerConnections(ServerConnectionCallback callback) {
     throw new UnsupportedOperationException();
   }
 
   static final class ProcessingTracker {

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

@@ -51,6 +51,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
@@ -529,7 +530,7 @@ private <T extends ResourceUpdate> void handleResourceUpdate(
   }
 
   @Override
-  public SettableFuture<Void> reportServerConnections(ServerConnectionCallback callback) {
+  public Future<Void> reportServerConnections(ServerConnectionCallback callback) {
     SettableFuture<Void> future = SettableFuture.create();
     syncContext.execute(() -> {
       serverCpClientMap.forEach((serverInfo, controlPlaneClient) ->