fix(spanner): use AggregationDefault for spanner grpc metrics

This prevents the OTel SDK from dropping the configured AttributeFilter when
processing asynchronous gauge metrics (such as grpc.subchannel.open_connections).
Forcing AggregationSum on gauge instruments caused the View to be rejected,
exporting distinct metric points with full cardinality. The exporter then blindly
stripped the labels, resulting in identical TimeSeries collisions and metric drops
in Cloud Monitoring due to duplicate timestamps.

By using AggregationDefault, the OTel SDK correctly interprets the AttributeFilter
for all metrics and natively aggregates gauges together before export.

Author: sakthivelmanii

spanner/metric_monitoring_exporter_test.go

@@ -21,6 +21,7 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"testing"
 	"time"
 
 	"cloud.google.com/go/monitoring/apiv3/v2/monitoringpb"
@@ -29,6 +30,13 @@ import (
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/emptypb"
+
+	"cloud.google.com/go/spanner/internal"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"google.golang.org/api/option"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 type MetricsTestServer struct {
@@ -124,3 +132,94 @@ func NewMetricTestServer() (*MetricsTestServer, error) {
 
 	return testServer, nil
 }
+
+func TestBuiltinMetrics_GaugeAggregation(t *testing.T) {
+	ctx := context.Background()
+
+	// Setup mock monitoring server
+	monitoringServer, err := NewMetricTestServer()
+	if err != nil {
+		t.Fatalf("Error setting up metrics test server")
+	}
+	defer monitoringServer.Shutdown()
+	go monitoringServer.Serve()
+
+	// Override exporter options to use the mock
+	origCreateExporterOptions := createExporterOptions
+	createExporterOptions = func(opts ...option.ClientOption) []option.ClientOption {
+		return []option.ClientOption{
+			option.WithEndpoint(monitoringServer.Endpoint), // Connect to mock
+			option.WithoutAuthentication(),
+			option.WithGRPCDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())),
+		}
+	}
+	defer func() {
+		createExporterOptions = origCreateExporterOptions
+	}()
+
+	// Build the meter provider using the built-in configs
+	clientAttributes := []attribute.KeyValue{
+		attribute.String("client_uid", "test-uid"),
+	}
+
+	mpOptions, exporter, err := builtInMeterProviderOptions("test-project", "", clientAttributes)
+	if err != nil {
+		t.Fatalf("builtInMeterProviderOptions failed: %v", err)
+	}
+
+	mp := sdkmetric.NewMeterProvider(mpOptions...)
+	defer mp.Shutdown(ctx)
+
+	// Obtain the specific meter bound to the grpc metric meter name configured in views
+	meter := mp.Meter("grpc-go", metric.WithInstrumentationVersion(internal.Version))
+
+	// Create the gauge (UpDownCounter in grpc-go)
+	openConns, err := meter.Int64UpDownCounter("grpc.subchannel.open_connections")
+	if err != nil {
+		t.Fatalf("Failed to create counter: %v", err)
+	}
+
+	// Record values with DIFFERENT target attributes (which should be DROPPED by the AttributeFilter)
+	openConns.Add(ctx, 1, metric.WithAttributes(attribute.String("grpc.target", "target-1")))
+	openConns.Add(ctx, 2, metric.WithAttributes(attribute.String("grpc.target", "target-2")))
+
+	// Force flush to the mock server
+	err = mp.ForceFlush(ctx)
+	if err != nil {
+		t.Fatalf("ForceFlush failed: %v", err)
+	}
+	// Give the mock server a moment to receive the RPC
+	time.Sleep(100 * time.Millisecond)
+
+	reqs := monitoringServer.CreateServiceTimeSeriesRequests()
+	if len(reqs) == 0 {
+		t.Fatalf("No CreateTimeSeriesRequests received")
+	}
+
+	// Check if all exported metrics only have one data point for grpc.subchannel.open_connections
+	var gaugePoints int64
+	var foundTimeSeries int
+
+	for _, req := range reqs {
+		for _, ts := range req.TimeSeries {
+			if ts.Metric.Type == "spanner.googleapis.com/internal/client/grpc/subchannel/open_connections" {
+				foundTimeSeries++
+				if len(ts.Points) > 0 {
+					gaugePoints += ts.Points[0].Value.GetInt64Value()
+				}
+			}
+		}
+	}
+
+	if foundTimeSeries != 1 {
+		t.Errorf("Expected exactly ONE TimeSeries for open_connections due to attribute filtering making them identical, got %d", foundTimeSeries)
+	}
+
+	// Since both Adds are recorded and their distinct attributes dropped, they are aggregated into 3.
+	if gaugePoints != 3 {
+		t.Errorf("Expected sum of open connections to be 3, got %d", gaugePoints)
+	}
+
+	// Ensure that after flush we cleanly close exporter
+	exporter.stop()
+}

spanner/metrics.go

@@ -362,7 +362,7 @@ func builtInMeterProviderOptions(project, compression string, clientAttributes [
 				Name: m,
 			},
 			sdkmetric.Stream{
-				Aggregation: sdkmetric.AggregationSum{},
+				Aggregation: sdkmetric.AggregationDefault{},
 				AttributeFilter: func(kv attribute.KeyValue) bool {
 					if _, ok := allowedMetricLabels[string(kv.Key)]; ok {
 						return true