app/common/telemetry.go

package common

import (
	"context"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"strings"
	"time"

	health "github.com/AppsFlyer/go-sundheit"
	healthhttp "github.com/AppsFlyer/go-sundheit/http"
	"github.com/go-chi/chi/v5"
	"github.com/go-chi/chi/v5/middleware"
	"github.com/go-slog/otelslog"
	"github.com/google/uuid"
	"github.com/google/wire"
	"github.com/oklog/ulid/v2"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/collectors"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	slogmulti "github.com/samber/slog-multi"
	realotelslog "go.opentelemetry.io/contrib/bridges/otelslog"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.opentelemetry.io/contrib/instrumentation/runtime"
	"go.opentelemetry.io/otel/log"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/propagation"
	sdklog "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.27.0"
	"go.opentelemetry.io/otel/trace"

	"github.com/openmeterio/openmeter/app/config"
	"github.com/openmeterio/openmeter/openmeter/server"
	"github.com/openmeterio/openmeter/pkg/contextx"
	"github.com/openmeterio/openmeter/pkg/framework/transport/httptransport"
	"github.com/openmeterio/openmeter/pkg/gosundheit"
)

var TelemetryWithoutServer = wire.NewSet(
	NewTelemetryResource,

	NewLoggerProvider,
	wire.Bind(new(log.LoggerProvider), new(*sdklog.LoggerProvider)),
	NewLogger,

	NewMeterProvider,
	wire.Bind(new(metric.MeterProvider), new(*sdkmetric.MeterProvider)),
	NewMeter,
	NewTracerProvider,
	wire.Bind(new(trace.TracerProvider), new(*sdktrace.TracerProvider)),
	NewTracer,

	NewRuntimeMetricsCollector,
)

var Telemetry = wire.NewSet(
	NewTelemetryResource,

	NewLoggerProvider,
	wire.Bind(new(log.LoggerProvider), new(*sdklog.LoggerProvider)),
	NewLogger,

	NewMeterProvider,
	wire.Bind(new(metric.MeterProvider), new(*sdkmetric.MeterProvider)),
	NewMeter,
	NewTracerProvider,
	wire.Bind(new(trace.TracerProvider), new(*sdktrace.TracerProvider)),
	NewTracer,

	NewHealthChecker,

	NewTelemetryHandler,
	NewTelemetryServer,

	NewRuntimeMetricsCollector,
)

// Set the default logger to JSON for messages emitted before the "real" logger is initialized.
//
// We use JSON as a best-effort to make the logs machine-readable.
func init() {
	slog.SetDefault(slog.New(config.NewJSONHandler(os.Stderr, slog.LevelInfo)))
}

const (
	DefaultShutdownTimeout = 5 * time.Second
)

func NewTelemetryResource(metadata Metadata) *resource.Resource {
	extraResources, _ := resource.New(
		// TODO: use the globally available context here?
		context.Background(),
		resource.WithContainer(),
		resource.WithAttributes(
			semconv.ServiceName(metadata.ServiceName),
			semconv.ServiceVersion(metadata.Version),
			semconv.DeploymentEnvironmentName(metadata.Environment),
		),
	)

	res, _ := resource.Merge(
		resource.Default(),
		extraResources,
	)

	return res
}

func NewLoggerProvider(ctx context.Context, conf config.LogTelemetryConfig, res *resource.Resource) (*sdklog.LoggerProvider, func(), error) {
	loggerProvider, err := conf.NewLoggerProvider(ctx, res)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to initialize OpenTelemetry Trace provider: %w", err)
	}

	return loggerProvider, func() {
		// Use dedicated context with timeout for shutdown as parent context might be canceled
		// by the time the execution reaches this stage.
		ctx, cancel := context.WithTimeout(context.Background(), DefaultShutdownTimeout)
		defer cancel()

		if err := loggerProvider.ForceFlush(ctx); err != nil {
			// no logger initialized at this point yet, so we are using the global logger
			slog.Error("flushing logger provider", slog.Any("error", err))
		}

		if err := loggerProvider.Shutdown(ctx); err != nil {
			// no logger initialized at this point yet, so we are using the global logger
			slog.Error("shutting down logger provider", slog.Any("error", err))
		}
	}, nil
}

func NewLogger(conf config.LogTelemetryConfig, res *resource.Resource, loggerProvider log.LoggerProvider, metadata Metadata, additionalMiddlewares []slogmulti.Middleware) *slog.Logger {
	baseMiddlewares := []slogmulti.Middleware{
		otelslog.ResourceMiddleware(res),
		otelslog.NewHandler,
	}

	baseMiddlewares = append(baseMiddlewares, additionalMiddlewares...)

	// Stdout logger
	stdoutLogger := slogmulti.
		Pipe(baseMiddlewares...).
		Handler(conf.NewHandler(os.Stdout))

	// OTel logger
	// It already has the resource middleware applied by the loggerProvider
	otelLogger := NewLevelHandler(
		realotelslog.NewHandler(metadata.OpenTelemetryName, realotelslog.WithLoggerProvider(loggerProvider)),
		conf.Level,
	)

	// Fanout logger to stdout and OTel logger
	out := slogmulti.Fanout(
		stdoutLogger,
		otelLogger,
	)

	// Enrich log records
	middlewares := slogmulti.Pipe(
		contextx.NewLogHandler,
	)

	return slog.New(middlewares.Handler(out))
}

func TelemetryLoggerNoAdditionalMiddlewares() []slogmulti.Middleware {
	return nil
}

func NewMeterProvider(ctx context.Context, conf config.MetricsTelemetryConfig, res *resource.Resource, logger *slog.Logger) (*sdkmetric.MeterProvider, func(), error) {
	meterProvider, err := conf.NewMeterProvider(ctx, res)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to initialize OpenTelemetry Metrics provider: %w", err)
	}

	return meterProvider, func() {
		// Use dedicated context with timeout for shutdown as parent context might be canceled
		// by the time the execution reaches this stage.
		ctx, cancel := context.WithTimeout(context.Background(), DefaultShutdownTimeout)
		defer cancel()

		if err := meterProvider.Shutdown(ctx); err != nil {
			logger.Error("shutting down meter provider", slog.Any("error", err))
		}
	}, nil
}

func NewMeter(meterProvider metric.MeterProvider, metadata Metadata) metric.Meter {
	return meterProvider.Meter(metadata.OpenTelemetryName)
}

func NewTracerProvider(ctx context.Context, conf config.TraceTelemetryConfig, res *resource.Resource, logger *slog.Logger) (*sdktrace.TracerProvider, func(), error) {
	tracerProvider, err := conf.NewTracerProvider(ctx, res)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to initialize OpenTelemetry Trace provider: %w", err)
	}

	// Apply the global per-operation child-span toggle once, at trace init.
	httptransport.EnableOperationSpans(conf.OperationSpans)

	return tracerProvider, func() {
		// Use dedicated context with timeout for shutdown as parent context might be canceled
		// by the time the execution reaches this stage.
		ctx, cancel := context.WithTimeout(context.Background(), DefaultShutdownTimeout)
		defer cancel()

		if err := tracerProvider.Shutdown(ctx); err != nil {
			logger.Error("shutting down tracer provider", slog.Any("error", err))
		}
	}, nil
}

func NewTracer(tracerProvider trace.TracerProvider, metadata Metadata) trace.Tracer {
	return tracerProvider.Tracer(metadata.OpenTelemetryName)
}

func NewDefaultTextMapPropagator() propagation.TextMapPropagator {
	return propagation.TraceContext{}
}

func NewHealthChecker(logger *slog.Logger) health.Health {
	return health.New(health.WithCheckListeners(gosundheit.NewLogger(logger.With(slog.String("component", "healthcheck")))))
}

type TelemetryHandler http.Handler

func NewTelemetryHandler(
	metricsConf config.MetricsTelemetryConfig,
	healthChecker health.Health,
	logger *slog.Logger,
) TelemetryHandler {
	if metricsConf.Exporters.Prometheus.DisableDefaultCollectors {
		prometheus.Unregister(collectors.NewGoCollector())
		prometheus.Unregister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
	}

	router := chi.NewRouter()
	router.Mount("/debug", middleware.Profiler())

	if metricsConf.Exporters.Prometheus.Enabled {
		router.Handle("/metrics", promhttp.Handler())
	}

	// Health
	{
		handler := healthhttp.HandleHealthJSON(healthChecker)
		router.Handle("/healthz", handler)

		// Kubernetes style health checks
		router.HandleFunc("/healthz/live", func(w http.ResponseWriter, _ *http.Request) {
			_, _ = w.Write([]byte("ok"))
		})
		router.Handle("/healthz/ready", handler)
	}

	return router
}

type TelemetryServer = *http.Server

func NewTelemetryServer(conf config.TelemetryConfig, handler TelemetryHandler) (TelemetryServer, func()) {
	server := &http.Server{
		Addr:              conf.Address,
		Handler:           handler,
		ReadHeaderTimeout: conf.ReadHeaderTimeout,
	}

	return server, func() { server.Close() }
}

type TelemetryMiddlewareHook server.MiddlewareHook

func NewTelemetryRouterHook(meterProvider metric.MeterProvider, tracerProvider trace.TracerProvider) TelemetryMiddlewareHook {
	return func(m server.MiddlewareManager) {
		m.Use(func(h http.Handler) http.Handler {
			return otelhttp.NewHandler(
				http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
					h.ServeHTTP(w, r)

					routePattern := chi.RouteContext(r.Context()).RoutePattern()

					// Record the route template as http.route on the span and request
					// metrics — the canonical low-cardinality routing dimension. The span
					// *name* is set separately by WithSpanNameFormatter below: otelhttp
					// fixes the name at tracer.Start and a post-start SetName here is a
					// no-op in this setup, so we cannot rename the span from this point.
					span := trace.SpanFromContext(r.Context())
					span.SetAttributes(semconv.URLPath(r.URL.Path), semconv.HTTPRoute(routePattern))

					if labeler, ok := otelhttp.LabelerFromContext(r.Context()); ok {
						labeler.Add(semconv.HTTPRoute(routePattern))
					}
				}),
				"",
				otelhttp.WithMeterProvider(meterProvider),
				otelhttp.WithTracerProvider(tracerProvider),
				otelhttp.WithSpanNameFormatter(func(_ string, r *http.Request) string {
					// Name the server span "METHOD /route/{template}" using the lowest-
					// cardinality route identifier available.
					//
					// Two-pass note: otelhttp calls this formatter once when the span
					// starts (before chi has finished routing) and, in v0.68, again after
					// the handler only when r.Pattern is set. r.Pattern is the net/http
					// ServeMux pattern; chi does not populate it, so under chi the name is
					// set on the first (pre-routing) call only. We therefore resolve the
					// route from the available signals, most specific first:
					//   1. r.Pattern           — net/http ServeMux pattern; chi sets it to the
					//                            mount prefix ("/api/v3/*") for sub-routers
					//   2. chi RoutePattern    — matched chi template ("{param}" placeholders)
					//   3. ULID-collapsed path — fallback when no resolved template is available
					// The exact template is also recorded on the http.route attribute.
					//
					// A value containing "*" is an unresolved mount prefix (e.g. "/api/v3/*"
					// when otelhttp runs inside a sub-router, before the leaf is matched —
					// this surfaces via r.Pattern on otelhttp's post-handler pass). It is not
					// a usable route name, so skip it (from any source) and fall back to the
					// ULID-collapsed path.
					route := r.Pattern
					if route == "" {
						if rctx := chi.RouteContext(r.Context()); rctx != nil {
							route = rctx.RoutePattern()
						}
					}
					if route == "" || strings.Contains(route, "*") {
						route = lowCardinalityPath(r.URL.Path)
					}
					return r.Method + " " + route
				}),
			)
		})
	}
}

const (
	// maxRouteSegmentLen: segments longer than this are treated as opaque identifiers
	// (tokens, keys, random slugs — typical of unmatched/scanner traffic). Legitimate
	// route segments are well under this (longest is ~"entitlement-access").
	maxRouteSegmentLen = 32
	// maxRouteSegments: paths deeper than this are truncated. Real routes are shallow;
	// deeper paths are almost always scanner traversal.
	maxRouteSegments = 12
)

// lowCardinalityPath turns a raw request path into a bounded span-name fragment.
// It masks high-cardinality segments (ULIDs, UUIDs, numeric ids, over-long opaque
// tokens) to ":id" and truncates pathologically deep paths, so unmatched/scanner
// traffic cannot blow up span-name cardinality. It is the name source both for v3
// routes (whose chi pattern is the skipped "/api/v3/*" mount wildcard) and for
// unmatched requests. The exact route template, when one matched, is carried
// separately on the http.route attribute.
func lowCardinalityPath(path string) string {
	segments := strings.Split(path, "/")

	truncated := false
	if len(segments) > maxRouteSegments {
		segments = segments[:maxRouteSegments]
		truncated = true
	}

	for i, seg := range segments {
		if isHighCardinalitySegment(seg) {
			segments[i] = ":id"
		}
	}

	out := strings.Join(segments, "/")
	if truncated {
		out += "/..."
	}
	return out
}

// isHighCardinalitySegment reports whether a path segment looks like an identifier or
// opaque token rather than a fixed route word.
func isHighCardinalitySegment(seg string) bool {
	if len(seg) > maxRouteSegmentLen {
		return true
	}
	if isAllDigits(seg) {
		return true
	}
	if _, err := ulid.ParseStrict(seg); err == nil {
		return true
	}
	if _, err := uuid.Parse(seg); err == nil {
		return true
	}
	return false
}

func isAllDigits(s string) bool {
	if s == "" {
		return false
	}
	for _, r := range s {
		if r < '0' || r > '9' {
			return false
		}
	}
	return true
}

type RuntimeMetricsCollector struct{}

func NewRuntimeMetricsCollector(
	meterProvider metric.MeterProvider,
	logger *slog.Logger,
) (RuntimeMetricsCollector, error) {
	err := runtime.Start(
		runtime.WithMinimumReadMemStatsInterval(time.Second),
		runtime.WithMeterProvider(meterProvider),
	)
	if err != nil {
		return RuntimeMetricsCollector{}, fmt.Errorf("starting runtime metrics: %w", err)
	}

	logger.Debug("started collecting runtime metrics")
	return RuntimeMetricsCollector{}, nil
}

// Compile-time check LevelHandler implements slog.Handler.
var _ slog.Handler = (*LevelHandler)(nil)

// NewLevelHandler returns a new LevelHandler.
func NewLevelHandler(handler slog.Handler, level slog.Leveler) *LevelHandler {
	return &LevelHandler{
		handler: handler,
		level:   level,
	}
}

// LevelHandler is a slog.Handler that filters log records based on the log level.
type LevelHandler struct {
	handler slog.Handler
	level   slog.Leveler
}

func (h *LevelHandler) Enabled(ctx context.Context, level slog.Level) bool {
	// The higher the level, the more important or severe the event.
	return level >= h.level.Level() && h.handler.Enabled(ctx, level)
}

func (h *LevelHandler) WithGroup(name string) slog.Handler {
	return NewLevelHandler(h.handler.WithGroup(name), h.level)
}

func (h *LevelHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
	return NewLevelHandler(h.handler.WithAttrs(attrs), h.level)
}

func (h *LevelHandler) Handle(ctx context.Context, record slog.Record) error {
	return h.handler.Handle(ctx, record)
}