Merge branch 'main' into mux-rw-it

Author: harshachinta

google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporter.java

@@ -39,6 +39,7 @@
 import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.data.PointData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.io.IOException;
 import java.time.Duration;
@@ -49,6 +50,7 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 /**
@@ -66,7 +68,7 @@ class SpannerCloudMonitoringExporter implements MetricExporter {
   // https://cloud.google.com/monitoring/quotas#custom_metrics_quotas.
   private static final int EXPORT_BATCH_SIZE_LIMIT = 200;
   private final AtomicBoolean spannerExportFailureLogged = new AtomicBoolean(false);
-  private CompletableResultCode lastExportCode;
+  private final AtomicBoolean lastExportSkippedData = new AtomicBoolean(false);
   private final MetricServiceClient client;
   private final String spannerProjectId;
 
@@ -101,44 +103,49 @@ static SpannerCloudMonitoringExporter create(
   }
 
   @Override
-  public CompletableResultCode export(Collection<MetricData> collection) {
+  public CompletableResultCode export(@Nonnull Collection<MetricData> collection) {
     if (client.isShutdown()) {
       logger.log(Level.WARNING, "Exporter is shut down");
       return CompletableResultCode.ofFailure();
     }
 
-    this.lastExportCode = exportSpannerClientMetrics(collection);
-    return lastExportCode;
+    return exportSpannerClientMetrics(collection);
   }
 
   /** Export client built in metrics */
   private CompletableResultCode exportSpannerClientMetrics(Collection<MetricData> collection) {
-    // Filter spanner metrics
+    // Filter spanner metrics. Only include metrics that contain a project and instance ID.
     List<MetricData> spannerMetricData =
         collection.stream()
             .filter(md -> SPANNER_METRICS.contains(md.getName()))
             .collect(Collectors.toList());
 
-    // Skips exporting if there's none
-    if (spannerMetricData.isEmpty()) {
-      return CompletableResultCode.ofSuccess();
-    }
-
-    // Verifies metrics project id is the same as the spanner project id set on this client
-    if (!spannerMetricData.stream()
+    // Log warnings for metrics that will be skipped.
+    boolean mustFilter = false;
+    if (spannerMetricData.stream()
         .flatMap(metricData -> metricData.getData().getPoints().stream())
-        .allMatch(
-            pd -> spannerProjectId.equals(SpannerCloudMonitoringExporterUtils.getProjectId(pd)))) {
-      logger.log(Level.WARNING, "Metric data has a different projectId. Skipping export.");
-      return CompletableResultCode.ofFailure();
+        .anyMatch(this::shouldSkipPointDataDueToProjectId)) {
+      logger.log(
+          Level.WARNING, "Some metric data contain a different projectId. These will be skipped.");
+      mustFilter = true;
     }
-
-    // Verifies if metrics data has missing instance id.
     if (spannerMetricData.stream()
         .flatMap(metricData -> metricData.getData().getPoints().stream())
-        .anyMatch(pd -> SpannerCloudMonitoringExporterUtils.getInstanceId(pd) == null)) {
-      logger.log(Level.WARNING, "Metric data has missing instanceId. Skipping export.");
-      return CompletableResultCode.ofFailure();
+        .anyMatch(this::shouldSkipPointDataDueToMissingInstanceId)) {
+      logger.log(Level.WARNING, "Some metric data miss instanceId. These will be skipped.");
+      mustFilter = true;
+    }
+    if (mustFilter) {
+      spannerMetricData =
+          spannerMetricData.stream()
+              .filter(this::shouldSkipMetricData)
+              .collect(Collectors.toList());
+    }
+    lastExportSkippedData.set(mustFilter);
+
+    // Skips exporting if there's none
+    if (spannerMetricData.isEmpty()) {
+      return CompletableResultCode.ofSuccess();
     }
 
     List<TimeSeries> spannerTimeSeries;
@@ -190,6 +197,26 @@ public void onSuccess(List<Empty> empty) {
     return spannerExportCode;
   }
 
+  private boolean shouldSkipMetricData(MetricData metricData) {
+    return metricData.getData().getPoints().stream()
+        .anyMatch(
+            pd ->
+                shouldSkipPointDataDueToProjectId(pd)
+                    || shouldSkipPointDataDueToMissingInstanceId(pd));
+  }
+
+  private boolean shouldSkipPointDataDueToProjectId(PointData pointData) {
+    return !spannerProjectId.equals(SpannerCloudMonitoringExporterUtils.getProjectId(pointData));
+  }
+
+  private boolean shouldSkipPointDataDueToMissingInstanceId(PointData pointData) {
+    return SpannerCloudMonitoringExporterUtils.getInstanceId(pointData) == null;
+  }
+
+  boolean lastExportSkippedData() {
+    return this.lastExportSkippedData.get();
+  }
+
   private ApiFuture<List<Empty>> exportTimeSeriesInBatch(
       ProjectName projectName, List<TimeSeries> timeSeries) {
     List<ApiFuture<Empty>> batchResults = new ArrayList<>();
@@ -233,7 +260,7 @@ public CompletableResultCode shutdown() {
    * metric over time.
    */
   @Override
-  public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
+  public AggregationTemporality getAggregationTemporality(@Nonnull InstrumentType instrumentType) {
     return AggregationTemporality.CUMULATIVE;
   }
 }

google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/HeaderInterceptor.java

@@ -28,7 +28,6 @@
 import com.google.cloud.spanner.CompositeTracer;
 import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.SpannerRpcMetrics;
-import com.google.common.base.Supplier;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import com.google.spanner.admin.database.v1.DatabaseName;
@@ -57,6 +56,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.regex.Matcher;
@@ -120,14 +120,14 @@ public void start(Listener<RespT> responseListener, Metadata headers) {
               getMetricAttributes(key, method.getFullMethodName(), databaseName);
           Map<String, String> builtInMetricsAttributes =
               getBuiltInMetricAttributes(key, databaseName);
+          addBuiltInMetricAttributes(compositeTracer, builtInMetricsAttributes);
           super.start(
               new SimpleForwardingClientCallListener<RespT>(responseListener) {
                 @Override
                 public void onHeaders(Metadata metadata) {
                   Boolean isDirectPathUsed =
                       isDirectPathUsed(getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR));
-                  addBuiltInMetricAttributes(
-                      compositeTracer, builtInMetricsAttributes, isDirectPathUsed);
+                  addDirectPathUsedAttribute(compositeTracer, isDirectPathUsed);
                   processHeader(metadata, tagContext, attributes, span);
                   super.onHeaders(metadata);
                 }
@@ -241,16 +241,17 @@ private Map<String, String> getBuiltInMetricAttributes(String key, DatabaseName
   }
 
   private void addBuiltInMetricAttributes(
-      CompositeTracer compositeTracer,
-      Map<String, String> builtInMetricsAttributes,
-      Boolean isDirectPathUsed) {
+      CompositeTracer compositeTracer, Map<String, String> builtInMetricsAttributes) {
     if (compositeTracer != null) {
-      // Direct Path used attribute
-      Map<String, String> attributes = new HashMap<>(builtInMetricsAttributes);
-      attributes.put(
-          BuiltInMetricsConstant.DIRECT_PATH_USED_KEY.getKey(), Boolean.toString(isDirectPathUsed));
+      compositeTracer.addAttributes(builtInMetricsAttributes);
+    }
+  }
 
-      compositeTracer.addAttributes(attributes);
+  private void addDirectPathUsedAttribute(
+      CompositeTracer compositeTracer, Boolean isDirectPathUsed) {
+    if (compositeTracer != null) {
+      compositeTracer.addAttributes(
+          BuiltInMetricsConstant.DIRECT_PATH_USED_KEY.getKey(), Boolean.toString(isDirectPathUsed));
     }
   }