fix spanner metrics regression

go.mod

@@ -124,6 +124,8 @@ require (
 
 require (
 	github.com/caio/go-tdigest/v4 v4.0.1
+	go.opentelemetry.io/otel/exporters/prometheus v0.57.0
+	go.opentelemetry.io/otel/sdk/metric v1.35.0
 	k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738
 )
 
@@ -390,7 +392,6 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.27.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.22.0 // indirect
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
-	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/automaxprocs v1.6.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect

go.sum

@@ -2526,6 +2526,8 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.27.0 h1:qFffA
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.27.0/go.mod h1:MOiCmryaYtc+V0Ei+Tx9o5S1ZjA7kzLucuVuyzBZloQ=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.22.0 h1:FyjCyI9jVEfqhUh2MoSkmolPjfh5fp2hnV0b0irxH4Q=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.22.0/go.mod h1:hYwym2nDEeZfG/motx0p7L7J1N1vyzIThemQsb4g2qY=
+go.opentelemetry.io/otel/exporters/prometheus v0.57.0 h1:AHh/lAP1BHrY5gBwk8ncc25FXWm/gmmY3BX258z5nuk=
+go.opentelemetry.io/otel/exporters/prometheus v0.57.0/go.mod h1:QpFWz1QxqevfjwzYdbMb4Y1NnlJvqSGwyuU0B4iuc9c=
 go.opentelemetry.io/otel/metric v1.19.0/go.mod h1:L5rUsV9kM1IxCj1MmSdS+JQAcVm319EUrDVLrt7jqt8=
 go.opentelemetry.io/otel/metric v1.21.0/go.mod h1:o1p3CA8nNHW8j5yuQLdc1eeqEaPfzug24uvsyIEJRWM=
 go.opentelemetry.io/otel/metric v1.22.0/go.mod h1:evJGjVpZv0mQ5QBRJoBF64yMuOf4xCWdXjK8pzFvliY=

internal/datastore/spanner/spanner.go

@@ -21,6 +21,10 @@
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	otelprom "go.opentelemetry.io/otel/exporters/prometheus"
+	"go.opentelemetry.io/otel/sdk/metric"
+	otelres "go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/api/option"
 	"google.golang.org/grpc"
@@ -98,6 +102,7 @@
 
 	tableSizesStatsTable string
 	filterMaximumIDCount uint16
+	meterProvider        *metric.MeterProvider
 }
 
 // NewSpannerDatastore returns a datastore backed by cloud spanner
@@ -122,9 +127,23 @@
 		log.Info().Str("spanner-emulator-host", os.Getenv("SPANNER_EMULATOR_HOST")).Msg("running against spanner emulator")
 	}
 
+	var meterProvider *metric.MeterProvider
 	if config.datastoreMetricsOption == DatastoreMetricsOptionOpenTelemetry {
 		log.Info().Msg("enabling OpenTelemetry metrics for Spanner datastore")
 		spanner.EnableOpenTelemetryMetrics()
+
+		res, err := otelres.Merge(otelres.Default(),
+			otelres.NewWithAttributes(semconv.SchemaURL,
+				semconv.ServiceName("spicedb"),
+			))
+		if err != nil {
+			return nil, fmt.Errorf("failed to create otel metrics resource: %w", err)
+		}
+
+		meterProvider, err = getMeterProviderWithPromExporter(res)
+		if err != nil {
+			return nil, fmt.Errorf("failed to enable Spanner prometheus metrics: %w", err)
+		}
 	}
 
 	if config.datastoreMetricsOption == DatastoreMetricsOptionLegacyPrometheus {
@@ -133,23 +152,21 @@
 		if err != nil {
 			return nil, fmt.Errorf("failed to enable spanner session metrics: %w", err)
 		}
+
 		err = spanner.EnableGfeLatencyAndHeaderMissingCountViews() // nolint: staticcheck
 		if err != nil {
 			return nil, fmt.Errorf("failed to enable spanner GFE metrics: %w", err)
 		}
-	}
 
-	// Register Spanner client gRPC metrics (include round-trip latency, received/sent bytes...)
-	if err := view.Register(ocgrpc.DefaultClientViews...); err != nil {
-		return nil, fmt.Errorf("failed to enable gRPC metrics for Spanner client: %w", err)
-	}
+		// Register Spanner client gRPC metrics (include round-trip latency, received/sent bytes...)
+		if err := view.Register(ocgrpc.DefaultClientViews...); err != nil {
+			return nil, fmt.Errorf("failed to enable gRPC metrics for Spanner client: %w", err)
+		}
 
-	_, err = ocprom.NewExporter(ocprom.Options{
-		Namespace:  "spicedb",
-		Registerer: prometheus.DefaultRegisterer,
-	})
-	if err != nil {
-		return nil, fmt.Errorf("failed to enable spanner GFE latency stats: %w", err)
+		_, err = ocprom.NewExporter(ocprom.Options{
+			Namespace:  "spicedb",
+			Registerer: prometheus.DefaultRegisterer,
+		})
 	}
 
 	cfg := spanner.DefaultSessionPoolConfig
@@ -175,8 +192,9 @@
 		context.Background(),
 		database,
 		spanner.ClientConfig{
-			SessionPoolConfig:    cfg,
-			DisableNativeMetrics: config.datastoreMetricsOption != DatastoreMetricsOptionNative,
+			SessionPoolConfig:          cfg,
+			DisableNativeMetrics:       config.datastoreMetricsOption != DatastoreMetricsOptionNative,
+			OpenTelemetryMeterProvider: meterProvider,
 		},
 		spannerOpts...,
 	)
@@ -243,6 +261,7 @@
 		tableSizesStatsTable:                    tableSizesStatsTable,
 		filterMaximumIDCount:                    config.filterMaximumIDCount,
 		schema:                                  *schema,
+		meterProvider:                           meterProvider,
 	}
 	// Optimized revision and revision checking use a stale read for the
 	// current timestamp.
@@ -253,6 +272,20 @@
 	return ds, nil
 }
 
+func getMeterProviderWithPromExporter(res *otelres.Resource) (*metric.MeterProvider, error) {
+	exporter, err := otelprom.New()
+	if err != nil {
+		return nil, err
+	}
+
+	meterProvider := metric.NewMeterProvider(
+		metric.WithResource(res),
+		metric.WithReader(exporter),
+	)
+
+	return meterProvider, nil
+}
+
 type traceableRTX struct {
 	delegate readTX
 }
@@ -410,6 +443,11 @@
 
 func (sd *spannerDatastore) Close() error {
 	sd.client.Close()
+
+	if sd.meterProvider != nil {
+		return sd.meterProvider.ForceFlush(context.TODO())
+	}
+
 	return nil
 }