[SLT-141] feat(metrics): multiple exports

core/go.mod

@@ -63,6 +63,7 @@ require (
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/sync v0.8.0
+	google.golang.org/grpc v1.64.0
 	gorm.io/driver/sqlite v1.5.6
 	gorm.io/gorm v1.25.10
 	k8s.io/apimachinery v0.29.3
@@ -189,7 +190,6 @@ require (
 	golang.org/x/tools v0.24.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240701130421-f6361c86f094 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240701130421-f6361c86f094 // indirect
-	google.golang.org/grpc v1.64.0 // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	gopkg.in/warnings.v0 v0.1.2 // indirect

core/metrics/README.md

@@ -12,7 +12,23 @@ There's also a `NAME_PREFIX` environment variable that will prefix all the metri
 
 ## OTLP
 
-We do our best to support enviornment variables specified in the [Otel Spec](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/) and have added a few of our own. Key ones to note are:
+We do our best to support enviornment variables specified in the [Otel Spec](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/) and the [OTLP Spec](https://opentelemetry.io/docs/languages/sdk-configuration/otlp-exporter/) and have added a few of our own. This was to allow for multiple exporter backends for traces, as otel clients only allow for one URL. The relevant multi exporter code is in `multiexporter.go`, and simply wraps multiple otel clients.
+
+The additional environment variables to note are:
+| Enviornment Variable                     | Description                               | Default |
+|------------------------------------------|-------------------------------------------|---------|
+| `OTEL_EXPORTER_OTLP_ENDPOINT`            | The endpoint for the primary OTLP exporter | None    |
+| `OTEL_EXPORTER_OTLP_ENDPOINT_1`          | The endpoint for the first additional OTLP exporter | None    |
+| `OTEL_EXPORTER_OTLP_ENDPOINT_2`          | The endpoint for the second additional OTLP exporter | None    |
+| `OTEL_EXPORTER_OTLP_ENDPOINT_3`          | The endpoint for the third additional OTLP exporter | None    |
+| ...                                      | Additional endpoints can be added by incrementing the number | None    |
+| `OTEL_EXPORTER_OTLP_TRANSPORT`           | The transport protocol for the primary OTLP exporter | `http` |
+| `OTEL_EXPORTER_OTLP_TRANSPORT_1`         | The transport protocol for the first additional OTLP exporter | `http` |
+| `OTEL_EXPORTER_OTLP_TRANSPORT_2`         | The transport protocol for the second additional OTLP exporter | `http` |
+| `OTEL_EXPORTER_OTLP_TRANSPORT_3`         | The transport protocol for the third additional OTLP exporter | `http` |
+| ...                                      | Additional transports can be specified by incrementing the number | `http` |
+
+Note: The OTLP exporter endpoints and transports can be specified for multiple exporters by using incrementing numbers (1, 2, 3, etc.) in the environment variable names. This allows for configuration of multiple OTLP exporters. The primary exporter uses the base names without numbers.
 
 
 ## Jaeger

core/metrics/export_test.go

@@ -0,0 +1,6 @@
+package metrics
+
+// HeadersToMap converts a string of headers to a map.
+func HeadersToMap(val string) map[string]string {
+	return headersToMap(val)
+}

core/metrics/multiexporter.go

@@ -0,0 +1,57 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/otel/sdk/trace"
+	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+)
+
+// MultiExporter is an interface that allows exporting spans to multiple OTLP trace exporters.
+type MultiExporter interface {
+	trace.SpanExporter
+	AddExporter(exporter trace.SpanExporter)
+}
+
+type multiExporter struct {
+	exporters []trace.SpanExporter
+}
+
+// NewMultiExporter creates a new multi exporter that forwards spans to multiple OTLP trace exporters.
+// It takes in one or more trace.SpanExporter instances and ensures that spans are sent to all of them.
+// This is useful when you need to send trace data to multiple backends or endpoints.
+func NewMultiExporter(exporters ...trace.SpanExporter) MultiExporter {
+	return &multiExporter{
+		exporters: exporters,
+	}
+}
+
+// ExportSpans exports a batch of spans.
+func (m *multiExporter) ExportSpans(ctx context.Context, ss []trace.ReadOnlySpan) error {
+	for _, exporter := range m.exporters {
+		err := exporter.ExportSpans(ctx, ss)
+		if err != nil {
+			return fmt.Errorf("could not export spans: %w", err)
+		}
+	}
+	return nil
+}
+
+// Shutdown notifies the exporter of a pending halt to operations.
+func (m *multiExporter) Shutdown(ctx context.Context) error {
+	for _, exporter := range m.exporters {
+		err := exporter.Shutdown(ctx)
+		if err != nil {
+			return fmt.Errorf("could not stop exporter: %w", err)
+		}
+	}
+	return nil
+}
+
+// AddExporter adds an exporter to the multi exporter.
+func (m *multiExporter) AddExporter(exporter trace.SpanExporter) {
+	m.exporters = append(m.exporters, exporter)
+}
+
+var _ tracesdk.SpanExporter = &multiExporter{}

core/metrics/multiexporter_test.go

@@ -0,0 +1,45 @@
+package metrics_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/synapsecns/sanguine/core/metrics"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+)
+
+func TestMultiExporter(t *testing.T) {
+	// Create in-memory exporters
+	exporter1 := tracetest.NewInMemoryExporter()
+	exporter2 := tracetest.NewInMemoryExporter()
+
+	// Create multi-exporter
+	multiExporter := metrics.NewMultiExporter(exporter1, exporter2)
+
+	// Create test spans
+	spans := []sdktrace.ReadOnlySpan{
+		tracetest.SpanStub{}.Snapshot(),
+		tracetest.SpanStub{}.Snapshot(),
+	}
+
+	// Test ExportSpans
+	err := multiExporter.ExportSpans(context.Background(), spans)
+	require.NoError(t, err)
+
+	// Verify that spans were exported to both exporters
+	assert.Equal(t, 2, len(exporter1.GetSpans()))
+	assert.Equal(t, 2, len(exporter2.GetSpans()))
+
+	// Test Shutdown
+	err = multiExporter.Shutdown(context.Background())
+	require.NoError(t, err)
+
+	// Verify that both exporters were shut down
+	// Note: InMemoryExporter doesn't have a Stopped() method, so we can't check this directly
+	// Instead, we can try to export spans again and check for an error
+	err = multiExporter.ExportSpans(context.Background(), spans)
+	assert.NoError(t, err, "Expected no error after shutdown")
+}

core/metrics/otlp.go

@@ -3,15 +3,16 @@
 import (
 	"context"
 	"fmt"
+	"google.golang.org/grpc/credentials"
+	"strings"
+	"time"
+
 	"github.com/synapsecns/sanguine/core"
 	"github.com/synapsecns/sanguine/core/config"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
-	"os"
-	"strings"
-	"time"
 )
 
 type otlpHandler struct {
@@ -28,23 +29,45 @@
 }
 
 func (n *otlpHandler) Start(ctx context.Context) (err error) {
-	var client otlptrace.Client
-	transport := transportFromString(core.GetEnv(otlpTransportEnv, otlpTransportGRPC.String()))
-	switch transport {
-	case otlpTransportHTTP:
-		client = otlptracehttp.NewClient()
-	case otlpTransportGRPC:
-		client = otlptracegrpc.NewClient()
-	default:
-		return fmt.Errorf("unknown transport type: %s", os.Getenv(otlpTransportEnv))
-	}
+	var exporters []tracesdk.SpanExporter
 
-	exporter, err := otlptrace.New(ctx, client)
+	primaryExporter, err := makeOTLPExporter(ctx, "")
 	if err != nil {
-		return fmt.Errorf("failed to create otlp exporter: %w", err)
+		return fmt.Errorf("could not create default client: %w", err)
+	}
+	exporters = append(exporters, primaryExporter)
+
+	// Loop to create additional exporters
+	for i := 1; ; i++ {
+		envSuffix := fmt.Sprintf("_%d", i)
+		// if this is empty we can assume no config exists at all.
+		endpointEnv := otelEndpointEnv + envSuffix
+
+		// no more transports to add.
+		if !core.HasEnv(endpointEnv) {
+			break
+		}
+
+		exporter, err := makeOTLPExporter(ctx, envSuffix)
+		if err != nil {
+			return fmt.Errorf("could not create exporter %d: %v", i, err)
+		}
+
+		exporters = append(exporters, exporter)
 	}
 
-	n.baseHandler = newBaseHandler(n.buildInfo, tracesdk.WithBatcher(exporter, tracesdk.WithMaxQueueSize(1000000), tracesdk.WithMaxExportBatchSize(2000)), tracesdk.WithSampler(tracesdk.AlwaysSample()))
+	// create the multi-exporter with all the exporters
+	multiExporter := NewMultiExporter(exporters...)
+
+	n.baseHandler = newBaseHandler(
+		n.buildInfo,
+		tracesdk.WithBatcher(
+			multiExporter,
+			tracesdk.WithMaxQueueSize(defaultMaxQueueSize),
+			tracesdk.WithMaxExportBatchSize(defaultMaxExportBatch),
+		),
+		tracesdk.WithSampler(tracesdk.AlwaysSample()),
+	)
 
 	// start the new parent
 	err = n.baseHandler.Start(ctx)
@@ -90,7 +113,10 @@
 }
 
 const (
-	otlpTransportEnv = "OTEL_EXPORTER_OTLP_TRANSPORT"
+	otelEndpointEnv  = "OTEL_EXPORTER_OTLP_ENDPOINT"
+	otelTransportEnv = "OTEL_EXPORTER_OTLP_TRANSPORT"
+	otelInsecureEvn  = "INSECURE_MODE"
+	otelHeadersEnv   = "OTEL_EXPORTER_OTLP_HEADERS"
 )
 
 //go:generate go run golang.org/x/tools/cmd/stringer -type=otlpTransportType -linecomment
@@ -101,6 +127,127 @@
 	otlpTransportGRPC                              // grpc
 )
 
+// getEnvSuffix returns the value of an environment variable with a suffix.
+func getEnvSuffix(env, suffix, defaultRet string) string {
+	newEnv := env + suffix
+	return core.GetEnv(newEnv, defaultRet)
+}
+
+// makeOTLPTrace creates a new OTLP client based on the transport type and url.
+func makeOTLPExporter(ctx context.Context, envSuffix string) (*otlptrace.Exporter, error) {
+	transport := transportFromString(getEnvSuffix(otelTransportEnv, envSuffix, otlpTransportHTTP.String()))
+	url := getEnvSuffix(otelEndpointEnv, envSuffix, "")
+	insecure := getEnvSuffix(otelInsecureEvn, envSuffix, "false")
+	headers := getEnvSuffix(otelHeadersEnv, envSuffix, "")
+
+	if url == "" {
+		return nil, fmt.Errorf("could not create exporter: url is empty")
+	}
+
+	oteltraceClient, err := buildClientFromTransport(
+		transport,
+		WithURL(url),
+		WithInsecure(insecure == "true"),
+		WithHeaders(headers),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("could not create client from transport: %w", err)
+	}
+
+	exporter, err := otlptrace.New(ctx, oteltraceClient)
+	if err != nil {
+		return nil, fmt.Errorf("ocould not create client: %w", err)
+	}
+	return exporter, nil
+}
+
+// buildClientFromTransport creates a new OTLP client based on the transport type and url.
+func buildClientFromTransport(transport otlpTransportType, options ...Option) (otlptrace.Client, error) {
+	to := transportOptions{}
+	for _, option := range options {
+		if err := option(&to); err != nil {
+			return nil, fmt.Errorf("could not apply option: %w", err)
+		}
+	}
+
+	// TODO: make sure url is validated
+
+	switch transport {
+	case otlpTransportHTTP:
+		return otlptracehttp.NewClient(to.httpOptions...), nil
+	case otlpTransportGRPC:
+		return otlptracegrpc.NewClient(to.grpcOptions...), nil
+	default:
+		return nil, fmt.Errorf("unknown transport type: %s", transport.String())
+	}
+}
+
+type transportOptions struct {
+	// httpOptions are the options for the http transport.
+	httpOptions []otlptracehttp.Option
+	// grpcOptions are the options for the grpc transport.
+	grpcOptions []otlptracegrpc.Option
+}
+
+// Option Each option appends the correspond option for both http and grpc options.
+// only one will be used in creating the actual client.
+type Option func(*transportOptions) error
+
+func WithURL(url string) Option {
+	return func(o *transportOptions) error {
+		o.httpOptions = append(o.httpOptions, otlptracehttp.WithEndpointURL(url))
+		o.grpcOptions = append(o.grpcOptions, otlptracegrpc.WithEndpointURL(url))
+
+		return nil
+	}
+}
+
+func WithInsecure(isInsecure bool) Option {
+	return func(o *transportOptions) error {
+		if isInsecure {
+			o.httpOptions = append(o.httpOptions, otlptracehttp.WithInsecure())
+			o.grpcOptions = append(o.grpcOptions, otlptracegrpc.WithInsecure())
+		} else {
+			tlsCreds := credentials.NewClientTLSFromCert(nil, "")
+			// note: you do not need to specify the tls creds for http, this happens automatically when https:// is used as the protocol scheme.
+			o.grpcOptions = append(o.grpcOptions, otlptracegrpc.WithTLSCredentials(tlsCreds))
+		}
+
+		return nil
+	}
+}
+
+func WithHeaders(headers string) Option {
+	return func(o *transportOptions) error {
+		realHeaders := headersToMap(headers)
+		o.httpOptions = append(o.httpOptions, otlptracehttp.WithHeaders(realHeaders))
+		o.grpcOptions = append(o.grpcOptions, otlptracegrpc.WithHeaders(realHeaders))
+		return nil
+	}
+}
+
+func headersToMap(input string) map[string]string {
+	// Initialize the map
+	result := make(map[string]string)
+
+	// Split the input string by comma to get key=value pairs
+	pairs := strings.Split(input, ",")
+
+	// Iterate over each pair
+	for _, pair := range pairs {
+		// Split each pair by '=' to get the key and value
+		kv := strings.SplitN(pair, "=", 2)
+		if len(kv) == 2 {
+			key := kv[0]
+			value := kv[1]
+			// Add the key and value to the map
+			result[key] = value
+		}
+	}
+
+	return result
+}
+
 // transportFromString converts a string to a transport type.
 // Defaults to http if the string is not recognized.
 func transportFromString(transport string) otlpTransportType {
@@ -114,3 +261,8 @@
 	// (see uber's go stye guide for details)
 	return otlpTransportType(0)
 }
+
+const (
+	defaultMaxQueueSize   = 1000000
+	defaultMaxExportBatch = 2000
+)