internal/observability/otel.go

Rama main · réplica de lectura del repositorio original.

package observability

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"slices"
	"strings"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/collectors"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
	otelprom "go.opentelemetry.io/otel/exporters/prometheus"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/sdk/log"
	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
	"go.opentelemetry.io/otel/sdk/resource"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
	"go.opentelemetry.io/otel/trace"

	"github.com/sarathsp06/sparrow"
)

// metricExportInterval is how often the OTLP metric reader flushes.
const metricExportInterval = 30 * time.Second

// OTLP transport protocols, as spelled by OTEL_EXPORTER_OTLP_PROTOCOL.
const (
	ProtocolHTTPProtobuf = "http/protobuf"
	ProtocolGRPC         = "grpc"
)

// OTLP signals, as spelled in Config.OTLPSignals.
const (
	SignalTraces  = "traces"
	SignalMetrics = "metrics"
	SignalLogs    = "logs"
)

var allSignals = []string{SignalTraces, SignalMetrics, SignalLogs}

// Config holds OpenTelemetry configuration
type Config struct {
	ServiceName    string
	ServiceVersion string
	Environment    string
	// OTLPEndpoint gates export: empty disables it. The exporters read the
	// standard OTEL_EXPORTER_OTLP_* env vars themselves (endpoint URL,
	// headers, TLS, timeouts, per-signal overrides). A bare host:port
	// without a scheme is still accepted and sent over plaintext.
	OTLPEndpoint string
	// OTLPProtocol is ProtocolHTTPProtobuf (default when empty) or ProtocolGRPC.
	OTLPProtocol string
	// OTLPSignals lists which of SignalTraces, SignalMetrics and SignalLogs
	// are sent to OTLPEndpoint. Empty means all three. Backends such as
	// Jaeger or Tempo accept traces only and answer the other exporters
	// with "unknown service"; listing just "traces" never creates them.
	OTLPSignals []string
	// Prometheus also exposes every OTel metric for scraping through
	// MetricsHandler, with or without OTLP export.
	Prometheus bool
}

// DefaultConfig returns a default OpenTelemetry configuration.
// OTLPEndpoint is empty by default -- set it to enable export.
// When OTLPEndpoint is empty, Setup() is a no-op (no exporters created).
func DefaultConfig() *Config {
	return &Config{
		ServiceName:    "sparrow",
		ServiceVersion: sparrow.Version,
		Environment:    "development",
		OTLPEndpoint:   "", // empty = OTel export disabled
	}
}

// Setup initializes OpenTelemetry with the provided configuration.
// When config.OTLPEndpoint is empty, no OTLP exporters are created (this
// avoids noisy connection errors when no collector is available); metrics are
// still collected when config.Prometheus is set, for MetricsHandler. With
// neither, Setup returns a no-op shutdown function.
func Setup(ctx context.Context, config *Config) (func(context.Context) error, error) {
	noop := func(context.Context) error { return nil }

	// Install the propagator unconditionally so trace context propagation
	// works even without an exporter (e.g. inbound headers are parsed).
	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
		propagation.TraceContext{},
		propagation.Baggage{},
	))

	otlp := config.OTLPEndpoint != ""
	if !otlp && !config.Prometheus {
		return noop, nil
	}

	if otlp {
		for _, signal := range config.OTLPSignals {
			if !slices.Contains(allSignals, signal) {
				return nil, fmt.Errorf("unknown OTLP signal %q (use %s)",
					signal, strings.Join(allSignals, ", "))
			}
		}
		switch config.OTLPProtocol {
		case "":
			config.OTLPProtocol = ProtocolHTTPProtobuf
		case ProtocolHTTPProtobuf, ProtocolGRPC:
		default:
			return nil, fmt.Errorf("unsupported OTLP protocol %q (use %q or %q)",
				config.OTLPProtocol, ProtocolGRPC, ProtocolHTTPProtobuf)
		}
	}

	// Create resource with service information
	res, err := resource.New(ctx,
		resource.WithAttributes(
			semconv.ServiceNameKey.String(config.ServiceName),
			semconv.ServiceVersionKey.String(config.ServiceVersion),
			semconv.DeploymentEnvironmentKey.String(config.Environment),
		),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create resource: %w", err)
	}

	var shutdownFuncs []func(context.Context) error
	// On partial failure, shut down already-built providers before returning
	// so we don't leak exporter connections.
	cleanup := func() {
		for _, fn := range shutdownFuncs {
			_ = fn(ctx)
		}
	}

	var tracerProvider *sdktrace.TracerProvider
	var loggerProvider *log.LoggerProvider
	if otlp && config.exports(SignalTraces) {
		tracerProvider, err = setupTracing(ctx, res, config)
		if err != nil {
			return nil, fmt.Errorf("failed to setup tracing: %w", err)
		}
		shutdownFuncs = append(shutdownFuncs, tracerProvider.Shutdown)
	}

	meterProvider, promHandler, err := setupMetrics(ctx, res, config)
	if err != nil {
		cleanup()
		return nil, fmt.Errorf("failed to setup metrics: %w", err)
	}
	shutdownFuncs = append(shutdownFuncs, meterProvider.Shutdown)

	if otlp && config.exports(SignalLogs) {
		loggerProvider, err = newLoggerProvider(ctx, config)
		if err != nil {
			cleanup()
			return nil, fmt.Errorf("failed to setup logger: %w", err)
		}
		shutdownFuncs = append(shutdownFuncs, loggerProvider.Shutdown)
	}

	// Install globals only after every provider was built successfully.
	if tracerProvider != nil {
		otel.SetTracerProvider(tracerProvider)
	}
	otel.SetMeterProvider(meterProvider)
	if loggerProvider != nil {
		otel.SetLoggerProvider(loggerProvider)
	}
	metricsHandler = promHandler

	// Return shutdown function
	return func(ctx context.Context) error {
		var errs []error
		for _, fn := range shutdownFuncs {
			if err := fn(ctx); err != nil {
				errs = append(errs, err)
			}
		}
		if len(errs) > 0 {
			return fmt.Errorf("failed to shutdown OpenTelemetry: %w", errors.Join(errs...))
		}
		return nil
	}, nil
}

// metricsHandler serves the Prometheus scrape endpoint; nil until Setup ran
// with Config.Prometheus.
var metricsHandler http.Handler

// MetricsHandler returns the handler serving every OTel metric in Prometheus
// text format, or nil when Setup did not enable Prometheus.
func MetricsHandler() http.Handler {
	return metricsHandler
}

// exports reports whether signal is among the OTLP signals to export; an
// empty OTLPSignals list exports every signal.
func (c *Config) exports(signal string) bool {
	return len(c.OTLPSignals) == 0 || slices.Contains(c.OTLPSignals, signal)
}

// legacyEndpoint reports whether endpoint is a bare host:port (the form
// Sparrow originally required). Those are passed explicitly as a plaintext
// endpoint; URL-form endpoints are left to the exporters' own env parsing,
// which derives TLS from the scheme.
func legacyEndpoint(endpoint string) bool {
	return !strings.Contains(endpoint, "://")
}

// setupTracing configures OpenTelemetry tracing
func setupTracing(ctx context.Context, res *resource.Resource, config *Config) (*sdktrace.TracerProvider, error) {
	var exporter sdktrace.SpanExporter
	var err error
	if config.OTLPProtocol == ProtocolGRPC {
		var opts []otlptracegrpc.Option
		if legacyEndpoint(config.OTLPEndpoint) {
			opts = append(opts, otlptracegrpc.WithEndpoint(config.OTLPEndpoint), otlptracegrpc.WithInsecure())
		}
		exporter, err = otlptracegrpc.New(ctx, opts...)
	} else {
		var opts []otlptracehttp.Option
		if legacyEndpoint(config.OTLPEndpoint) {
			opts = append(opts, otlptracehttp.WithEndpoint(config.OTLPEndpoint), otlptracehttp.WithInsecure())
		}
		exporter, err = otlptracehttp.New(ctx, opts...)
	}
	if err != nil {
		return nil, fmt.Errorf("failed to create OTLP trace exporter: %w", err)
	}

	// Create tracer provider
	tracerProvider := sdktrace.NewTracerProvider(
		sdktrace.WithBatcher(exporter),
		sdktrace.WithResource(res),
		sdktrace.WithSampler(sdktrace.AlwaysSample()),
	)

	return tracerProvider, nil
}

func newLoggerProvider(ctx context.Context, config *Config) (*log.LoggerProvider, error) {
	var exporter log.Exporter
	var err error
	if config.OTLPProtocol == ProtocolGRPC {
		var opts []otlploggrpc.Option
		if legacyEndpoint(config.OTLPEndpoint) {
			opts = append(opts, otlploggrpc.WithEndpoint(config.OTLPEndpoint), otlploggrpc.WithInsecure())
		}
		exporter, err = otlploggrpc.New(ctx, opts...)
	} else {
		var opts []otlploghttp.Option
		if legacyEndpoint(config.OTLPEndpoint) {
			opts = append(opts, otlploghttp.WithEndpoint(config.OTLPEndpoint), otlploghttp.WithInsecure())
		}
		exporter, err = otlploghttp.New(ctx, opts...)
	}
	if err != nil {
		return nil, fmt.Errorf("failed to create OTLP log exporter: %w", err)
	}

	loggerProvider := log.NewLoggerProvider(
		log.WithProcessor(log.NewBatchProcessor(exporter)),
	)
	return loggerProvider, nil
}

// setupMetrics configures OpenTelemetry metrics: a periodic OTLP reader when
// an endpoint is configured, and a Prometheus pull reader (returned as its
// scrape handler) when config.Prometheus is set. Both read the same
// instruments, so every metric is available through either path.
func setupMetrics(ctx context.Context, res *resource.Resource, config *Config) (*sdkmetric.MeterProvider, http.Handler, error) {
	opts := []sdkmetric.Option{sdkmetric.WithResource(res)}

	if config.OTLPEndpoint != "" && config.exports(SignalMetrics) {
		var exporter sdkmetric.Exporter
		var err error
		if config.OTLPProtocol == ProtocolGRPC {
			var opts []otlpmetricgrpc.Option
			if legacyEndpoint(config.OTLPEndpoint) {
				opts = append(opts, otlpmetricgrpc.WithEndpoint(config.OTLPEndpoint), otlpmetricgrpc.WithInsecure())
			}
			exporter, err = otlpmetricgrpc.New(ctx, opts...)
		} else {
			var opts []otlpmetrichttp.Option
			if legacyEndpoint(config.OTLPEndpoint) {
				opts = append(opts, otlpmetrichttp.WithEndpoint(config.OTLPEndpoint), otlpmetrichttp.WithInsecure())
			}
			exporter, err = otlpmetrichttp.New(ctx, opts...)
		}
		if err != nil {
			return nil, nil, fmt.Errorf("failed to create OTLP metric exporter: %w", err)
		}
		opts = append(opts, sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exporter,
			sdkmetric.WithInterval(metricExportInterval))))
	}

	var handler http.Handler
	if config.Prometheus {
		// A private registry keeps the scrape output to Sparrow's own
		// metrics plus the standard Go runtime and process collectors.
		registry := prometheus.NewRegistry()
		registry.MustRegister(
			collectors.NewGoCollector(),
			collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
		)
		exporter, err := otelprom.New(otelprom.WithRegisterer(registry))
		if err != nil {
			return nil, nil, fmt.Errorf("failed to create Prometheus exporter: %w", err)
		}
		opts = append(opts, sdkmetric.WithReader(exporter))
		handler = promhttp.HandlerFor(registry, promhttp.HandlerOpts{})
	}

	return sdkmetric.NewMeterProvider(opts...), handler, nil
}

// GetTracer returns a tracer for the given name
func GetTracer(name string) trace.Tracer {
	return otel.Tracer(name, trace.WithInstrumentationVersion(sparrow.Version))
}

// GetMeter returns a meter for the given name
func GetMeter(name string) metric.Meter {
	return otel.Meter(name, metric.WithInstrumentationVersion(sparrow.Version))
}

// SparrowMetrics holds application-specific metrics
type SparrowMetrics struct {
	WebhookRegistrations metric.Int64Counter
	EventsPushed         metric.Int64Counter
}

// NewSparrowMetrics creates application-specific metrics
func NewSparrowMetrics() (*SparrowMetrics, error) {
	meter := GetMeter("sparrow")

	webhookRegistrations, err := meter.Int64Counter(
		"sparrow_webhook_registrations_total",
		metric.WithDescription("Total number of webhook registrations"),
	)
	if err != nil {
		return nil, err
	}

	eventsPushed, err := meter.Int64Counter(
		"sparrow_events_pushed_total",
		metric.WithDescription("Total number of events pushed"),
	)
	if err != nil {
		return nil, err
	}

	return &SparrowMetrics{
		WebhookRegistrations: webhookRegistrations,
		EventsPushed:         eventsPushed,
	}, nil
}