Merge pull request #243 from go-faster/fix/telemetry-leveling

fix(autologs): filter otlp logs by level enabler

Author: ernado

autologs/autologs.go

@@ -14,6 +14,7 @@ import (
 	"go.opentelemetry.io/otel/log/noop"
 	sdklog "go.opentelemetry.io/otel/sdk/log"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 
 	"github.com/go-faster/sdk/zctx"
 )
@@ -58,8 +59,8 @@ type ShutdownFunc func(ctx context.Context) error
 
 // NewLoggerProvider initializes new [log.LoggerProvider] with the given options from environment variables.
 func NewLoggerProvider(ctx context.Context, options ...Option) (
-	meterProvider log.LoggerProvider,
-	meterShutdown ShutdownFunc,
+	logProvider log.LoggerProvider,
+	logShutdown ShutdownFunc,
 	err error,
 ) {
 	cfg := newConfig(options)
@@ -68,11 +69,16 @@ func NewLoggerProvider(ctx context.Context, options ...Option) (
 	if cfg.res != nil {
 		logOptions = append(logOptions, sdklog.WithResource(cfg.res))
 	}
+
 	ret := func(e sdklog.Exporter) (log.LoggerProvider, func(ctx context.Context) error, error) {
-		logOptions = append(logOptions, sdklog.WithProcessor(
-			sdklog.NewBatchProcessor(e),
-		))
-		return sdklog.NewLoggerProvider(logOptions...), e.Shutdown, nil
+		logOptions = append(logOptions,
+			sdklog.WithProcessor(&levelFilterProcessor{
+				next:     sdklog.NewBatchProcessor(e),
+				severity: zapLevelToOTelSeverity(lg.Level()),
+			}),
+		)
+		provider := sdklog.NewLoggerProvider(logOptions...)
+		return provider, provider.Shutdown, nil
 	}
 	exporter := strings.TrimSpace(getEnvOr("OTEL_LOGS_EXPORTER", expOTLP))
 	switch exporter {
@@ -134,3 +140,57 @@ func NewLoggerProvider(ctx context.Context, options ...Option) (
 	}
 	return nil, nil, errors.Errorf("unsupported OTEL_LOGS_EXPORTER %q", exporter)
 }
+
+// levelFilterProcessor implements level filtering, since otlplog does not.
+//
+// Fuck you too, OpenTelemetry.
+type levelFilterProcessor struct {
+	next     sdklog.Processor
+	severity log.Severity
+}
+
+var (
+	_ sdklog.FilterProcessor = (*levelFilterProcessor)(nil)
+	_ sdklog.Processor       = (*levelFilterProcessor)(nil)
+)
+
+// Enabled implements [sdklog.FilterProcessor].
+func (l *levelFilterProcessor) Enabled(ctx context.Context, param sdklog.EnabledParameters) bool {
+	return param.Severity >= l.severity
+}
+
+// OnEmit implements [sdklog.Processor].
+func (l *levelFilterProcessor) OnEmit(ctx context.Context, record *sdklog.Record) error {
+	return l.next.OnEmit(ctx, record)
+}
+
+// ForceFlush implements [sdklog.Processor].
+func (l *levelFilterProcessor) ForceFlush(ctx context.Context) error {
+	return l.next.ForceFlush(ctx)
+}
+
+// Shutdown implements [sdklog.Processor].
+func (l *levelFilterProcessor) Shutdown(ctx context.Context) error {
+	return l.next.Shutdown(ctx)
+}
+
+func zapLevelToOTelSeverity(level zapcore.Level) log.Severity {
+	switch level {
+	case zapcore.DebugLevel:
+		return log.SeverityDebug
+	case zapcore.InfoLevel:
+		return log.SeverityInfo
+	case zapcore.WarnLevel:
+		return log.SeverityWarn
+	case zapcore.ErrorLevel:
+		return log.SeverityError
+	case zapcore.DPanicLevel:
+		return log.SeverityFatal1
+	case zapcore.PanicLevel:
+		return log.SeverityFatal2
+	case zapcore.FatalLevel:
+		return log.SeverityFatal3
+	default:
+		return log.SeverityUndefined
+	}
+}

autologs/autologs_test.go

@@ -0,0 +1,94 @@
+package autologs_test
+
+import (
+	"context"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/go-faster/errors"
+	"github.com/go-faster/sdk/autologs"
+	"github.com/go-faster/sdk/zctx"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/contrib/bridges/otelzap"
+	sdklog "go.opentelemetry.io/otel/sdk/log"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest"
+)
+
+func TestNewLoggerProviderLevel(t *testing.T) {
+	ctx := context.Background()
+	const testExporterName = "amongus"
+	t.Setenv("OTEL_LOGS_EXPORTER", testExporterName)
+
+	baseLogger := zaptest.NewLogger(t, zaptest.Level(zap.InfoLevel))
+	ctx = zctx.Base(ctx, baseLogger)
+
+	exporter := &testLogExporter{}
+	provider, shutdown, err := autologs.NewLoggerProvider(ctx, autologs.WithLookupExporter(func(ctx context.Context, name string) (sdklog.Exporter, bool, error) {
+		if name != testExporterName {
+			return nil, false, errors.Errorf("wrong exporter %q", name)
+		}
+		return exporter, true, nil
+	}))
+	require.NoError(t, err)
+
+	otelCore := otelzap.NewCore("github.com/go-faster/sdk/app",
+		otelzap.WithLoggerProvider(provider),
+	)
+	otelLg := zap.New(otelCore)
+	otelLg.Debug("hot and lonely GPUs around you")
+	otelLg.Info("information")
+	otelLg.Warn("warning")
+
+	require.NoError(t, otelLg.Sync())
+	require.NoError(t, shutdown(ctx))
+	require.True(t, exporter.shutdown.Load())
+
+	var msgs []string
+	for _, r := range exporter.Records() {
+		msgs = append(msgs, r.Body().AsString())
+	}
+	require.Equal(t,
+		[]string{
+			"information",
+			"warning",
+		},
+		msgs,
+	)
+}
+
+type testLogExporter struct {
+	records    []sdklog.Record
+	recordsMux sync.Mutex
+	shutdown   atomic.Bool
+}
+
+var _ sdklog.Exporter = (*testLogExporter)(nil)
+
+func (t *testLogExporter) Records() []sdklog.Record {
+	t.recordsMux.Lock()
+	r := slices.Clone(t.records)
+	t.recordsMux.Unlock()
+	return r
+}
+
+// Export implements [sdklog.Exporter].
+func (t *testLogExporter) Export(ctx context.Context, records []sdklog.Record) error {
+	t.recordsMux.Lock()
+	t.records = append(t.records, records...)
+	t.recordsMux.Unlock()
+	return nil
+}
+
+// ForceFlush implements [sdklog.Exporter].
+func (t *testLogExporter) ForceFlush(ctx context.Context) error {
+	return nil
+}
+
+// Shutdown implements [sdklog.Exporter].
+func (t *testLogExporter) Shutdown(ctx context.Context) error {
+	t.shutdown.Store(true)
+	return nil
+}

autometer/autometer.go

@@ -78,7 +78,8 @@ func NewMeterProvider(ctx context.Context, options ...Option) (
 
 	ret := func(r sdkmetric.Reader) (metric.MeterProvider, func(ctx context.Context) error, error) {
 		metricOptions = append(metricOptions, sdkmetric.WithReader(r))
-		return sdkmetric.NewMeterProvider(metricOptions...), r.Shutdown, nil
+		provider := sdkmetric.NewMeterProvider(metricOptions...)
+		return provider, provider.Shutdown, nil
 	}
 
 	// Metrics exporter.

autotracer/autotracer.go

@@ -70,7 +70,8 @@ func NewTracerProvider(ctx context.Context, options ...Option) (
 	}
 	ret := func(e sdktrace.SpanExporter) (trace.TracerProvider, func(ctx context.Context) error, error) {
 		traceOptions = append(traceOptions, sdktrace.WithBatcher(e))
-		return sdktrace.NewTracerProvider(traceOptions...), e.Shutdown, nil
+		provider := sdktrace.NewTracerProvider(traceOptions...)
+		return provider, provider.Shutdown, nil
 	}
 
 	exporter := strings.TrimSpace(getEnvOr("OTEL_TRACES_EXPORTER", expOTLP))