feat(flagd): custom error handling for OTel errors (open-feature#769)

Co-authored-by: Michael Beemer <[email protected]>

core/pkg/runtime/from_config.go

@@ -85,6 +85,9 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime,
 		CollectorTarget: config.OtelCollectorURI,
 	}
 
+	// register error handling for OpenTelemetry
+	telemetry.RegisterErrorHandling(logger)
+
 	// register trace provider for the runtime
 	err := telemetry.BuildTraceProvider(context.Background(), logger, svcName, version, telCfg)
 	if err != nil {

core/pkg/telemetry/builder.go

@@ -18,6 +18,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/resource"
 	"go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -33,6 +34,12 @@ type Config struct {
 	CollectorTarget string
 }
 
+func RegisterErrorHandling(log *logger.Logger) {
+	otel.SetErrorHandler(otelErrorsHandler{
+		logger: log,
+	})
+}
+
 // BuildMetricsRecorder is a helper to build telemetry.MetricsRecorder based on configurations
 func BuildMetricsRecorder(
 	ctx context.Context, svcName string, svcVersion string, config Config,
@@ -58,7 +65,7 @@ func BuildMetricsRecorder(
 // This results in tracers having NoopTracerProvider and propagator having No-Op TextMapPropagator performing no action
 func BuildTraceProvider(ctx context.Context, logger *logger.Logger, svc string, svcVersion string, cfg Config) error {
 	if cfg.CollectorTarget == "" {
-		logger.Warn("skipping trace provider setup as collector target is not set." +
+		logger.Debug("skipping trace provider setup as collector target is not set." +
 			" Traces will use NoopTracerProvider provider and propagator will use no-Op TextMapPropagator")
 		return nil
 	}
@@ -172,3 +179,13 @@ func buildResourceFor(ctx context.Context, serviceName string, serviceVersion st
 	}
 	return r, nil
 }
+
+// OTelErrorsHandler is a custom error interceptor for OpenTelemetry
+type otelErrorsHandler struct {
+	logger *logger.Logger
+}
+
+func (h otelErrorsHandler) Handle(err error) {
+	msg := fmt.Sprintf("OpenTelemetry Error: %s", err.Error())
+	h.logger.WithFields(zap.String("component", "otel")).Debug(msg)
+}

core/pkg/telemetry/builder_test.go

@@ -2,12 +2,20 @@ package telemetry
 
 import (
 	"context"
+	"fmt"
 	"testing"
+	"time"
 
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/aggregation"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/resource"
 	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 )
 
 func TestBuildMetricsRecorder(t *testing.T) {
@@ -151,3 +159,45 @@ func TestBuildResourceFor(t *testing.T) {
 		Value: attribute.StringValue(svcVersion),
 	}, "expected resource to contain service version")
 }
+
+func TestErrorIntercepted(t *testing.T) {
+	// register the OTel error handling
+	observedZapCore, observedLogs := observer.New(zap.DebugLevel)
+	observedLogger := zap.New(observedZapCore)
+	log := logger.NewLogger(observedLogger, true)
+	RegisterErrorHandling(log)
+
+	// configure a metric reader with an exporter that only returns error
+	reader := metric.NewPeriodicReader(&errorExp{}, metric.WithInterval(1*time.Millisecond))
+	rs := resource.NewWithAttributes("testSchema")
+	NewOTelRecorder(reader, rs, "testSvc")
+	var data metricdata.ResourceMetrics
+	err := reader.Collect(context.TODO(), &data)
+	require.Nil(t, err)
+
+	// we should have some logs that were intercepted
+	require.True(t, observedLogs.FilterField(zap.String("component", "otel")).Len() > 0)
+}
+
+// errorExp is an exporter that always fails
+type errorExp struct{}
+
+func (e *errorExp) Temporality(k metric.InstrumentKind) metricdata.Temporality {
+	return metric.DefaultTemporalitySelector(k)
+}
+
+func (e *errorExp) Aggregation(_ metric.InstrumentKind) aggregation.Aggregation {
+	return nil
+}
+
+func (e *errorExp) Export(_ context.Context, _ *metricdata.ResourceMetrics) error {
+	return fmt.Errorf("I am an error")
+}
+
+func (e *errorExp) ForceFlush(_ context.Context) error {
+	return fmt.Errorf("I am an error")
+}
+
+func (e *errorExp) Shutdown(_ context.Context) error {
+	return fmt.Errorf("I am an error")
+}