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

core/metrics/multiexporter.go

@@ -0,0 +1,46 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
+	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+)
+
+type multiExporter struct {
+	exporters []*otlptrace.Exporter
+}
+
+// NewMultiExporter creates a new multi exporter that forwards spans to multiple OTLP trace exporters.
+// It takes in one or more otlptrace.Exporter 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 ...*otlptrace.Exporter) tracesdk.SpanExporter {
+	return &multiExporter{
+		exporters: exporters,
+	}
+}
+
+// ExportSpans exports a batch of spans.
+func (m *multiExporter) ExportSpans(ctx context.Context, ss []tracesdk.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
+}
+
+var _ tracesdk.SpanExporter = &multiExporter{}

core/metrics/otlp.go

@@ -3,15 +3,15 @@
 import (
 	"context"
 	"fmt"
+	"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 +28,48 @@
 }
 
 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))
+	// TODO: generalize this to allow for more than two exporters.
+	client, err := buildClientFromTransport(
+		transportFromString(
+			core.GetEnv(otlpTransportEnv, otlpTransportGRPC.String()),
+		),
+		core.GetEnv(otlpTransportEnv, "localhost:4317"),
+	)
+	if err != nil {
+		return fmt.Errorf("could not create client: %w", err)
+	}
+
+	secondaryClient, err := buildClientFromTransport(
+		transportFromString(
+			core.GetEnv(otlpTransportEnvSecondary, otlpTransportGRPC.String()),
+		),
+		core.GetEnv(otlpTransportEnvSecondary, "localhost:4317"),
+	)
+	if err != nil {
+		return fmt.Errorf("could not create secondary client: %w", err)
 	}
 
 	exporter, err := otlptrace.New(ctx, client)
 	if err != nil {
 		return fmt.Errorf("failed to create otlp exporter: %w", err)
 	}
 
-	n.baseHandler = newBaseHandler(n.buildInfo, tracesdk.WithBatcher(exporter, tracesdk.WithMaxQueueSize(1000000), tracesdk.WithMaxExportBatchSize(2000)), tracesdk.WithSampler(tracesdk.AlwaysSample()))
+	secondaryExporter, err := otlptrace.New(ctx, secondaryClient)
+	if err != nil {
+		return fmt.Errorf("failed to create secondary otlp exporter: %w", err)
+	}
+
+	multiExporter := NewMultiExporter(exporter, secondaryExporter)
+
+	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 +115,8 @@
 }
 
 const (
-	otlpTransportEnv = "OTEL_EXPORTER_OTLP_TRANSPORT"
+	otlpTransportEnv          = "OTEL_EXPORTER_OTLP_TRANSPORT_PRIMARY"
+	otlpTransportEnvSecondary = "OTEL_EXPORTER_OTLP_TRANSPORT_SECONDARY"
 )
 
 //go:generate go run golang.org/x/tools/cmd/stringer -type=otlpTransportType -linecomment
@@ -101,6 +127,17 @@
 	otlpTransportGRPC                              // grpc
 )
 
+func buildClientFromTransport(transport otlpTransportType, url string) (otlptrace.Client, error) {
+	switch transport {
+	case otlpTransportHTTP:
+		return otlptracehttp.NewClient(otlptracehttp.WithEndpointURL(url)), nil
+	case otlpTransportGRPC:
+		return otlptracegrpc.NewClient(otlptracegrpc.WithEndpointURL(url)), nil
+	default:
+		return nil, fmt.Errorf("unknown transport type: %s", transport.String())
+	}
+}
+
 // transportFromString converts a string to a transport type.
 // Defaults to http if the string is not recognized.
 func transportFromString(transport string) otlpTransportType {
@@ -114,3 +151,8 @@
 	// (see uber's go stye guide for details)
 	return otlpTransportType(0)
 }
+
+const (
+	defaultMaxQueueSize   = 1000000
+	defaultMaxExportBatch = 2000
+)

core/metrics/rookout.go

@@ -1,12 +1,15 @@
+//go:build go1.16 && !go1.23
+
 package metrics
 
 import (
+	"os"
+
 	rookout "github.com/Rookout/GoSDK"
 	"github.com/Rookout/GoSDK/pkg/config"
 	"github.com/synapsecns/sanguine/core"
 	synconfig "github.com/synapsecns/sanguine/core/config"
 	"github.com/synapsecns/sanguine/core/metrics/internal"
-	"os"
 )
 
 // DefaultGitRepo is the default git repo for sanguine.