INFOPLAT-13349: feat(beholder): track beholder.export.* metrics per export via custom gRPC stats handler

pkg/beholder/client.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"io"
 
-	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
@@ -110,7 +109,7 @@ type otlploggrpcFactory func(options ...otlploggrpc.Option) (sdklog.Exporter, er
 
 // NewGRPCClient creates a GRPC based beholder Client. Use NewClient to create a client from a Config which will pick
 // the best client type from the Config.
-func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, error) {
+func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (_ *Client, err error) {
 	baseResource, err := newOtelResource(cfg)
 	if err != nil {
 		return nil, err
@@ -132,29 +131,58 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 		return nil, err
 	}
 
+	traceStatsHandler := newExportStatsHandler(signalTraces, cfg.AuthPublicKeyHex)
+
 	// Tracer
-	tracerProvider, err := newTracerProvider(cfg, baseResource, auth, creds)
+	tracerProvider, meteredTraces, err := newTracerProvider(cfg, baseResource, auth, creds, traceStatsHandler)
 	if err != nil {
 		return nil, err
 	}
+	// If any later step fails, shut the providers down so their gRPC
+	// connections and the metric periodic-reader goroutine are not leaked.
+	defer func() {
+		if err != nil {
+			_ = tracerProvider.Shutdown(context.Background())
+		}
+	}()
 	tracer := tracerProvider.Tracer(defaultPackageName)
 
+	logStatsHandler := newExportStatsHandler(signalLogs, cfg.AuthPublicKeyHex)
+	metricStatsHandler := newExportStatsHandler(signalMetrics, cfg.AuthPublicKeyHex)
+
 	// Meter
-	meterProvider, err := newMeterProvider(cfg, baseResource, auth, creds)
+	meterProvider, meteredMetrics, err := newMeterProvider(cfg, baseResource, auth, creds, metricStatsHandler)
 	if err != nil {
 		return nil, err
 	}
+	defer func() {
+		if err != nil {
+			_ = meterProvider.Shutdown(context.Background())
+		}
+	}()
 	meter := meterProvider.Meter(defaultPackageName)
 
-	// Shared log exporter for both logger and message emitter
-	logOpts, err := newLoggerOpts(cfg, auth, creds, meterProvider, tracerProvider)
+	// Shared export instruments beholder.export.* metrics
+	expMetrics, err := newExportMetrics(meter)
 	if err != nil {
 		return nil, err
 	}
-	sharedLogExporter, err := otlploggrpcNew(logOpts...)
+	logStatsHandler.attachMetrics(expMetrics)
+	metricStatsHandler.attachMetrics(expMetrics)
+	traceStatsHandler.attachMetrics(expMetrics)
+	meteredMetrics.attachMetrics(expMetrics, cfg.AuthPublicKeyHex)
+	meteredTraces.attachMetrics(expMetrics, cfg.AuthPublicKeyHex)
+
+	// Shared log exporter for both logger and message emitter.
+	logOpts, err := newLoggerOpts(cfg, auth, creds, logStatsHandler)
 	if err != nil {
 		return nil, err
 	}
+	rawLogExporter, err := otlploggrpcNew(logOpts...)
+	if err != nil {
+		return nil, err
+	}
+	sharedLogExporter := newMeteredLogExporter(rawLogExporter, expMetrics, cfg.AuthPublicKeyHex)
 
 	// Logger
 	var loggerProvider *sdklog.LoggerProvider
@@ -352,7 +380,7 @@ func newOtelResource(cfg Config) (resource *sdkresource.Resource, err error) {
 	}
 
 	// Add csa public key resource attribute
-	csaPublicKeyHex := "not-configured"
+	csaPublicKeyHex := csaPublicKeyNotConfigured
 	if len(cfg.AuthPublicKeyHex) > 0 {
 		csaPublicKeyHex = cfg.AuthPublicKeyHex
 	}
@@ -448,23 +476,29 @@ type shutdowner interface {
 	Shutdown(ctx context.Context) error
 }
 
-func newTracerProvider(config Config, resource *sdkresource.Resource, auth Auth, creds credentials.TransportCredentials) (*sdktrace.TracerProvider, error) {
+func newTracerProvider(config Config, resource *sdkresource.Resource, auth Auth, creds credentials.TransportCredentials, statsHandler *exportStatsHandler) (*sdktrace.TracerProvider, *meteredTraceExporter, error) {
 	ctx := context.Background()
 
+	dialOpts := []grpc.DialOption{
+		grpc.WithStatsHandler(statsHandler),
+	}
+
 	exporterOpts := []otlptracegrpc.Option{
 		otlptracegrpc.WithTLSCredentials(creds),
 		otlptracegrpc.WithEndpoint(config.OtelExporterGRPCEndpoint),
 	}
 	switch {
 	// Rotating auth
 	case auth != nil:
-		exporterOpts = append(exporterOpts, otlptracegrpc.WithDialOption(authDialOpt(auth)))
+		dialOpts = append(dialOpts, authDialOpt(auth))
 	// Static auth
 	case len(config.AuthHeaders) > 0:
 		exporterOpts = append(exporterOpts, otlptracegrpc.WithHeaders(config.AuthHeaders))
 	// No auth
 	default:
 	}
+
+	exporterOpts = append(exporterOpts, otlptracegrpc.WithDialOption(dialOpts...))
 	switch compressor := config.TraceCompressor; compressor {
 	case "none":
 	case "":
@@ -484,14 +518,16 @@ func newTracerProvider(config Config, resource *sdkresource.Resource, auth Auth,
 	// note: context is used internally
 	exporter, err := otlptracegrpc.New(ctx, exporterOpts...)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
+	metered := newMeteredTraceExporter(exporter)
+
 	batcherOpts := []sdktrace.BatchSpanProcessorOption{}
 	if config.TraceBatchTimeout > 0 {
 		batcherOpts = append(batcherOpts, sdktrace.WithBatchTimeout(config.TraceBatchTimeout)) // Default is 5s
 	}
 	opts := []sdktrace.TracerProviderOption{
-		sdktrace.WithBatcher(exporter, batcherOpts...),
+		sdktrace.WithBatcher(metered, batcherOpts...),
 		sdktrace.WithResource(resource),
 		sdktrace.WithSampler(
 			sdktrace.ParentBased(
@@ -502,15 +538,20 @@ func newTracerProvider(config Config, resource *sdkresource.Resource, auth Auth,
 	if config.TraceSpanExporter != nil {
 		opts = append(opts, sdktrace.WithBatcher(config.TraceSpanExporter))
 	}
-	return sdktrace.NewTracerProvider(opts...), nil
+	return sdktrace.NewTracerProvider(opts...), metered, nil
 }
 
-func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, creds credentials.TransportCredentials) (*sdkmetric.MeterProvider, error) {
+func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, creds credentials.TransportCredentials, statsHandler *exportStatsHandler) (*sdkmetric.MeterProvider, *meteredMetricExporter, error) {
 	ctx := context.Background()
 	opts := []otlpmetricgrpc.Option{
 		otlpmetricgrpc.WithTLSCredentials(creds),
 		otlpmetricgrpc.WithEndpoint(cfg.OtelExporterGRPCEndpoint),
 	}
+
+	dialOpts := []grpc.DialOption{
+		grpc.WithStatsHandler(statsHandler),
+	}
+
 	switch compressor := cfg.MetricCompressor; compressor {
 	case "none":
 	case "":
@@ -522,13 +563,14 @@ func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, cre
 	switch {
 	// Rotating auth
 	case auth != nil:
-		opts = append(opts, otlpmetricgrpc.WithDialOption(authDialOpt(auth)))
+		dialOpts = append(dialOpts, authDialOpt(auth))
 	// Static auth
 	case len(cfg.AuthHeaders) > 0:
 		opts = append(opts, otlpmetricgrpc.WithHeaders(cfg.AuthHeaders))
 	// No auth
 	default:
 	}
+	opts = append(opts, otlpmetricgrpc.WithDialOption(dialOpts...))
 
 	if cfg.MetricRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK
@@ -542,31 +584,30 @@ func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, cre
 	// note: context is unused internally
 	exporter, err := otlpmetricgrpc.New(ctx, opts...)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
+	metered := newMeteredMetricExporter(exporter)
+
 	readerOpts := []sdkmetric.PeriodicReaderOption{
 		sdkmetric.WithInterval(cfg.MetricReaderInterval), // Default is 10s
 	}
 	for _, p := range cfg.MetricProducers {
 		readerOpts = append(readerOpts, sdkmetric.WithProducer(p))
 	}
+
 	mpOpts := append(cfg.metricOptions(),
-		sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exporter, readerOpts...)),
+		sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metered, readerOpts...)),
 		sdkmetric.WithResource(resource),
 	)
-	return sdkmetric.NewMeterProvider(mpOpts...), nil
+	return sdkmetric.NewMeterProvider(mpOpts...), metered, nil
 }
 
 // newLoggerOpts creates options for a logger exporter
-func newLoggerOpts(cfg Config, auth Auth, creds credentials.TransportCredentials, meter *sdkmetric.MeterProvider, tracer *sdktrace.TracerProvider) ([]otlploggrpc.Option, error) {
-	otelOpts := []otelgrpc.Option{
-		otelgrpc.WithMeterProvider(meter),
-		otelgrpc.WithTracerProvider(tracer),
-	}
+func newLoggerOpts(cfg Config, auth Auth, creds credentials.TransportCredentials, statsHandler *exportStatsHandler) ([]otlploggrpc.Option, error) {
 
 	dialOpts := []grpc.DialOption{
-		grpc.WithStatsHandler(otelgrpc.NewClientHandler(otelOpts...)),
+		grpc.WithStatsHandler(statsHandler),
 	}
 
 	opts := []otlploggrpc.Option{

pkg/beholder/meter_provider_test.go

@@ -2,14 +2,17 @@ package beholder
 
 import (
 	"context"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder/metricviews"
 )
@@ -89,3 +92,114 @@ func TestConfig_metricViews_emptyDenylistOmitsCatchAll(t *testing.T) {
 	require.Len(t, views, len(metricviews.Default(nil)))
 	require.NotEmpty(t, views)
 }
+
+// capturingMetricExporter records the ResourceMetrics handed to Export so a test
+// can inspect what a MeterProvider's reader actually collected.
+type capturingMetricExporter struct {
+	mu     sync.Mutex
+	rm     metricdata.ResourceMetrics
+	called bool
+}
+
+func (c *capturingMetricExporter) Temporality(k sdkmetric.InstrumentKind) metricdata.Temporality {
+	return sdkmetric.DefaultTemporalitySelector(k)
+}
+
+func (c *capturingMetricExporter) Aggregation(k sdkmetric.InstrumentKind) sdkmetric.Aggregation {
+	return sdkmetric.DefaultAggregationSelector(k)
+}
+
+func (c *capturingMetricExporter) Export(_ context.Context, rm *metricdata.ResourceMetrics) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.rm, c.called = *rm, true
+	return nil
+}
+
+func (c *capturingMetricExporter) ForceFlush(context.Context) error { return nil }
+func (c *capturingMetricExporter) Shutdown(context.Context) error   { return nil }
+
+func (c *capturingMetricExporter) collected() (metricdata.ResourceMetrics, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.rm, c.called
+}
+
+// TestNewMeterProvider_AppliesCardinalityLimit is the regression guard for the
+// gRPC meter provider bypassing Config.metricOptions. Asserting on
+// metricOptions alone cannot catch that, since the bug is the call site not
+// using it.
+func TestNewMeterProvider_AppliesCardinalityLimit(t *testing.T) {
+	t.Parallel()
+
+	const (
+		uniqueAttributes = 10
+		limit            = 5
+	)
+
+	cfg := TestDefaultConfig()
+	cfg.MetricCardinalityLimit = limit
+	// Isolate the assertion from metricviews.Default, same as
+	// TestConfig_metricOptions_cardinalityLimit above.
+	cfg.metricViewsDisabled = true
+	// Long interval so only the explicit ForceFlush below triggers a collection.
+	cfg.MetricReaderInterval = time.Hour
+
+	resource, err := newOtelResource(cfg)
+	require.NoError(t, err)
+
+	mp, metered, err := newMeterProvider(cfg, resource, nil, insecure.NewCredentials(), newExportStatsHandler("metrics", cfg.AuthPublicKeyHex))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = mp.Shutdown(context.Background()) })
+
+	// Swap the real OTLP exporter for one we can read. Shut the original down
+	// separately, since mp.Shutdown will now reach the capturing exporter.
+	capture := &capturingMetricExporter{}
+	original := metered.Exporter
+	metered.Exporter = capture
+	t.Cleanup(func() { _ = original.Shutdown(context.Background()) })
+
+	counter, err := mp.Meter("test").Int64Counter("overflow_test_total")
+	require.NoError(t, err)
+	for i := range uniqueAttributes {
+		counter.Add(context.Background(), 1, metric.WithAttributes(attribute.Int("key", i)))
+	}
+
+	require.NoError(t, mp.ForceFlush(context.Background()))
+
+	rm, called := capture.collected()
+	require.True(t, called, "ForceFlush should have exported a batch")
+
+	var sum metricdata.Sum[int64]
+	var found bool
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name == "overflow_test_total" {
+				sum, found = m.Data.(metricdata.Sum[int64]), true
+			}
+		}
+	}
+	require.True(t, found, "expected overflow_test_total in the collected batch")
+	assert.Len(t, sum.DataPoints, limit,
+		"newMeterProvider must honour MetricCardinalityLimit via Config.metricOptions")
+}
+
+func TestNewTracerProvider_WiresMeteredTraceExporter(t *testing.T) {
+	cfg := TestDefaultConfig()
+	resource, err := newOtelResource(cfg)
+	require.NoError(t, err)
+
+	tp, metered, err := newTracerProvider(cfg, resource, nil, insecure.NewCredentials(), newExportStatsHandler("traces", cfg.AuthPublicKeyHex))
+	require.NoError(t, err)
+	require.NotNil(t, tp)
+	require.NotNil(t, metered, "the trace exporter must be wrapped so beholder.export.duration is emitted for traces")
+
+	inst, collect := newTestMetrics(t)
+	metered.attachMetrics(inst, cfg.AuthPublicKeyHex)
+	require.NoError(t, metered.ExportSpans(context.Background(), nil))
+
+	_, ok := histForSignal(t, collect(), "traces", false)
+	assert.True(t, ok, "expected a traces duration datapoint")
+
+	require.NoError(t, tp.Shutdown(context.Background()))
+}