autoexport: Add a metric producer registry and ability to set producers

Signed-off-by: Goutham <[email protected]>

Author: gouthamve

exporters/autoexport/go.mod

@@ -5,6 +5,7 @@ go 1.21
 require (
 	github.com/prometheus/client_golang v1.19.0
 	github.com/stretchr/testify v1.9.0
+	go.opentelemetry.io/contrib/bridges/prometheus v0.50.0
 	go.opentelemetry.io/otel v1.25.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.25.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.25.0

exporters/autoexport/go.sum

@@ -33,6 +33,8 @@ github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDN
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+go.opentelemetry.io/contrib/bridges/prometheus v0.50.0 h1:akXN45Sg2oS2NOb2xBL0LKeq/oSyEIvc8CC/7XLaB+4=
+go.opentelemetry.io/contrib/bridges/prometheus v0.50.0/go.mod h1:uoFuIBjQ9kWtUv4KbRNq0ExS9BQoWxHrr63JWX/EMb8=
 go.opentelemetry.io/otel v1.25.0 h1:gldB5FfhRl7OJQbUHt/8s0a7cE8fbsPAtdpRaApKy4k=
 go.opentelemetry.io/otel v1.25.0/go.mod h1:Wa2ds5NOXEMkCmUou1WA7ZBfLTHWIsp034OVD7AO+Vg=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.25.0 h1:hDKnobznDpcdTlNzO0S/owRB8tyVr1OoeZZhDoqY+Cs=

exporters/autoexport/metrics.go

@@ -15,6 +15,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
+	prometheusbridge "go.opentelemetry.io/contrib/bridges/prometheus"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
@@ -71,10 +72,24 @@ func RegisterMetricReader(name string, factory func(context.Context) (metric.Rea
 	must(metricsSignal.registry.store(name, factory))
 }
 
+func RegisterMetricProducer(name string, factory func(context.Context) (metric.Producer, error)) {
+	must(metricsProducers.registry.store(name, factory))
+}
+
 var metricsSignal = newSignal[metric.Reader]("OTEL_METRICS_EXPORTER")
+var metricsProducers = newProducerRegistry("OTEL_METRICS_PRODUCERS")
 
 func init() {
 	RegisterMetricReader("otlp", func(ctx context.Context) (metric.Reader, error) {
+		producer, err := metricsProducers.create(ctx)
+		if err != nil {
+			return nil, err
+		}
+		readerOpts := []metric.PeriodicReaderOption{}
+		if producer != nil {
+			readerOpts = append(readerOpts, metric.WithProducer(producer))
+		}
+
 		proto := os.Getenv(otelExporterOTLPProtoEnvKey)
 		if proto == "" {
 			proto = "http/protobuf"
@@ -86,23 +101,32 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
-			return metric.NewPeriodicReader(r), nil
+			return metric.NewPeriodicReader(r, readerOpts...), nil
 		case "http/protobuf":
 			r, err := otlpmetrichttp.New(ctx)
 			if err != nil {
 				return nil, err
 			}
-			return metric.NewPeriodicReader(r), nil
+			return metric.NewPeriodicReader(r, readerOpts...), nil
 		default:
 			return nil, errInvalidOTLPProtocol
 		}
 	})
 	RegisterMetricReader("console", func(ctx context.Context) (metric.Reader, error) {
+		producer, err := metricsProducers.create(ctx)
+		if err != nil {
+			return nil, err
+		}
+		readerOpts := []metric.PeriodicReaderOption{}
+		if producer != nil {
+			readerOpts = append(readerOpts, metric.WithProducer(producer))
+		}
+
 		r, err := stdoutmetric.New()
 		if err != nil {
 			return nil, err
 		}
-		return metric.NewPeriodicReader(r), nil
+		return metric.NewPeriodicReader(r, readerOpts...), nil
 	})
 	RegisterMetricReader("none", func(ctx context.Context) (metric.Reader, error) {
 		return newNoopMetricReader(), nil
@@ -148,6 +172,10 @@ func init() {
 
 		return readerWithServer{lis.Addr(), reader, &server}, nil
 	})
+
+	RegisterMetricProducer("prometheus", func(ctx context.Context) (metric.Producer, error) {
+		return prometheusbridge.NewMetricProducer(), nil
+	})
 }
 
 type readerWithServer struct {
@@ -170,3 +198,26 @@ func getenv(key, fallback string) string {
 	}
 	return result
 }
+
+type producerRegistry struct {
+	envKey   string
+	registry *registry[metric.Producer]
+}
+
+func newProducerRegistry(envKey string) producerRegistry {
+	return producerRegistry{
+		envKey: envKey,
+		registry: &registry[metric.Producer]{
+			names: make(map[string]func(context.Context) (metric.Producer, error)),
+		},
+	}
+}
+
+func (pr producerRegistry) create(ctx context.Context) (metric.Producer, error) {
+	expType := os.Getenv(pr.envKey)
+	if expType == "" {
+		return nil, nil
+	}
+
+	return pr.registry.load(ctx, expType)
+}

exporters/autoexport/registry.go

@@ -21,9 +21,9 @@ type registry[T any] struct {
 }
 
 var (
-	// errUnknownExporter is returned when an unknown exporter name is used in
-	// the OTEL_*_EXPORTER environment variables.
-	errUnknownExporter = errors.New("unknown exporter")
+	// errUnknownExporterProducer is returned when an unknown exporter name is used in
+	// the OTEL_*_EXPORTER or OTEL_METRICS_PRODUCERS environment variables.
+	errUnknownExporterProducer = errors.New("unknown exporter or metrics producer")
 
 	// errInvalidOTLPProtocol is returned when an invalid protocol is used in
 	// the OTEL_EXPORTER_OTLP_PROTOCOL environment variable.
@@ -35,15 +35,15 @@ var (
 
 // load returns tries to find the exporter factory with the key and
 // then execute the factory, returning the created SpanExporter.
-// errUnknownExporter is returned if the registration is missing and the error from
+// errUnknownExporterProducer is returned if the registration is missing and the error from
 // executing the factory if not nil.
 func (r *registry[T]) load(ctx context.Context, key string) (T, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	factory, ok := r.names[key]
 	if !ok {
 		var zero T
-		return zero, errUnknownExporter
+		return zero, errUnknownExporterProducer
 	}
 	return factory(ctx)
 }

exporters/autoexport/registry_test.go

@@ -34,7 +34,7 @@ func TestCanStoreExporterFactory(t *testing.T) {
 func TestLoadOfUnknownExporterReturnsError(t *testing.T) {
 	r := newTestRegistry()
 	exp, err := r.load(context.Background(), "non-existent")
-	assert.Equal(t, err, errUnknownExporter, "empty registry should hold nothing")
+	assert.Equal(t, err, errUnknownExporterProducer, "empty registry should hold nothing")
 	assert.Nil(t, exp, "non-nil exporter returned")
 }