obs.go

// Package obs is the atomdrift shared OpenTelemetry wiring.
//
// Services call [Init] once during startup with a [Config] (zero values
// fall back to OTEL_* env vars), defer the returned shutdown into their
// signal-handling path, and then use [Meter], [Tracer], [Job], and the
// HTTP helpers as needed.
//
// With nothing configured, OTLP push defaults to the self-hosted Grafana
// fleet at host `otel` ([DefaultEndpoint]), following the per-signal split:
//
//   - Metrics use the base endpoint ([Config.Endpoint] or
//     OTEL_EXPORTER_OTLP_ENDPOINT, else OTEL_EXPORTER_OTLP_METRICS_ENDPOINT,
//     else [DefaultEndpoint]), which points at the Prometheus OTLP receiver.
//   - Logs go to Loki. A logs-specific OTEL_EXPORTER_OTLP_LOGS_ENDPOINT is
//     honored verbatim; otherwise, whenever a base endpoint is set, logs fall
//     back to [DefaultLogsEndpoint] instead of the base — the Prometheus OTLP
//     receiver 404s logs.
//   - Traces push only when a traces-specific OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
//     is set. The fleet has no trace backend, so a bare base endpoint does not
//     enable trace export (sending spans to Prometheus would 404).
//
// The Prometheus scrape handler exposed by [MetricsHandler] is always available
// so callers can mount /_/metrik regardless of OTLP wiring.
package obs

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	promclient "github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
	otelprom "go.opentelemetry.io/otel/exporters/prometheus"
	"go.opentelemetry.io/otel/metric"
	sdklog "go.opentelemetry.io/otel/sdk/log"
	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/trace"

	"go.opentelemetry.io/otel/log/global"
)

// Config controls obs initialization. The zero value is invalid
// (ServiceName is required); all other fields fall back to env vars
// or sensible defaults.
type Config struct {
	// ServiceName is the OTel service.name resource attribute. Required.
	ServiceName string

	// ServiceVersion is the OTel service.version resource attribute.
	// If empty, falls back to debug.ReadBuildInfo() then "devel".
	ServiceVersion string

	// Environment is the OTel deployment.environment resource attribute.
	// If empty, falls back to OTEL_DEPLOYMENT_ENVIRONMENT then "dev".
	Environment string

	// Endpoint is the OTLP/HTTP base endpoint applied to any signal
	// that does not have a signal-specific OTEL_EXPORTER_OTLP_*_ENDPOINT
	// override. If empty and no per-signal env vars are set, obs falls
	// back to [DefaultEndpoint] (the self-hosted Prometheus OTLP receiver
	// at host `otel`).
	Endpoint string

	// Insecure forces http:// when true. If false, the URL scheme of the
	// resolved endpoint decides.
	Insecure bool

	// SampleRatio is the trace head-sampling ratio (0.0–1.0). If zero,
	// falls back to OTEL_TRACES_SAMPLER_ARG, then 0.1.
	SampleRatio float64

	// DisableSlog skips Init's slog.SetDefault. Use this when the
	// caller has its own fan-out chain (e.g. stderr + file + OTLP) and
	// wants to compose [OTLPSlogHandler] in itself.
	DisableSlog bool
}

var (
	initOnce       atomic.Bool
	promRegistry   *promclient.Registry
	meterProvider  *sdkmetric.MeterProvider
	tracerProvider *sdktrace.TracerProvider
	loggerProvider *sdklog.LoggerProvider
)

// promScrapeHandler returns the http.Handler that serves the
// Prometheus-format snapshot of the OTel meter provider, or nil if
// Init has not yet been called.
func promScrapeHandler() http.Handler {
	if promRegistry == nil {
		return nil
	}
	return promhttp.HandlerFor(promRegistry, promhttp.HandlerOpts{})
}

// Init wires up OpenTelemetry for the calling service. It is safe to
// call exactly once per process; a second call returns an error.
//
// Init installs global trace, meter, and logger providers; replaces
// slog.Default with a handler that tees to stderr and the OTel log
// pipeline; and enables push exporters only for signals whose endpoint
// is configured. The returned shutdown should be invoked with a
// deadline-bearing context during graceful termination.
func Init(ctx context.Context, cfg Config) (func(context.Context) error, error) {
	if cfg.ServiceName == "" {
		return nil, errors.New("obs: ServiceName is required")
	}
	if !initOnce.CompareAndSwap(false, true) {
		return nil, errors.New("obs: Init called twice")
	}

	if cfg.ServiceVersion == "" {
		cfg.ServiceVersion = detectVersion()
	}
	if cfg.Environment == "" {
		if v := os.Getenv("OTEL_DEPLOYMENT_ENVIRONMENT"); v != "" {
			cfg.Environment = v
		} else {
			cfg.Environment = "dev"
		}
	}
	if cfg.SampleRatio == 0 {
		if v := os.Getenv("OTEL_TRACES_SAMPLER_ARG"); v != "" {
			if f, err := strconv.ParseFloat(v, 64); err == nil {
				cfg.SampleRatio = f
			}
		}
		if cfg.SampleRatio == 0 {
			cfg.SampleRatio = 0.1
		}
	}

	res, err := buildResource(ctx, cfg)
	if err != nil {
		return nil, err
	}

	shutdowns := []func(context.Context) error{}

	// --- Metrics ---
	// Prometheus exporter is always wired so /_/metrik works even when
	// OTLP push is disabled. Use a private registry so we don't conflict
	// with anything else the service might register globally.
	reg := promclient.NewRegistry()
	promExp, err := otelprom.New(otelprom.WithRegisterer(reg))
	if err != nil {
		return nil, fmt.Errorf("obs: prometheus exporter: %w", err)
	}
	promRegistry = reg

	// Expose Go-runtime and process metrics (memory, CPU, goroutines, FDs, GC)
	// on the same scrape registry, so every service gets them for free.
	if err := enableRuntimeMetrics(reg); err != nil {
		return nil, fmt.Errorf("obs: runtime metrics: %w", err)
	}

	metricOpts := []sdkmetric.Option{
		sdkmetric.WithResource(res),
		sdkmetric.WithReader(promExp),
	}
	if sc := signalEndpoint(cfg, "METRICS"); sc.endpoint != "" {
		mexp, err := newMetricExporter(ctx, sc, cfg.Insecure)
		if err != nil {
			return nil, fmt.Errorf("obs: metric exporter: %w", err)
		}
		metricOpts = append(metricOpts,
			sdkmetric.WithReader(sdkmetric.NewPeriodicReader(mexp,
				sdkmetric.WithInterval(15*time.Second))))
		shutdowns = append(shutdowns, mexp.Shutdown)
	}
	mp := sdkmetric.NewMeterProvider(metricOpts...)
	otel.SetMeterProvider(mp)
	meterProvider = mp
	shutdowns = append(shutdowns, mp.Shutdown)

	// --- Traces ---
	traceOpts := []sdktrace.TracerProviderOption{
		sdktrace.WithResource(res),
		sdktrace.WithSampler(sdktrace.ParentBased(
			sdktrace.TraceIDRatioBased(cfg.SampleRatio))),
	}
	if sc := resolveTraces(cfg); sc.endpoint != "" {
		texp, err := newTraceExporter(ctx, sc, cfg.Insecure)
		if err != nil {
			return nil, fmt.Errorf("obs: trace exporter: %w", err)
		}
		traceOpts = append(traceOpts, sdktrace.WithBatcher(texp))
	}
	tp := sdktrace.NewTracerProvider(traceOpts...)
	otel.SetTracerProvider(tp)
	tracerProvider = tp
	shutdowns = append(shutdowns, tp.Shutdown)

	// --- Logs ---
	logOpts := []sdklog.LoggerProviderOption{sdklog.WithResource(res)}
	if sc := resolveLogs(cfg); sc.endpoint != "" {
		lexp, err := newLogExporter(ctx, sc, cfg.Insecure)
		if err != nil {
			return nil, fmt.Errorf("obs: log exporter: %w", err)
		}
		logOpts = append(logOpts, sdklog.WithProcessor(sdklog.NewBatchProcessor(lexp)))
	}
	lp := sdklog.NewLoggerProvider(logOpts...)
	global.SetLoggerProvider(lp)
	loggerProvider = lp
	shutdowns = append(shutdowns, lp.Shutdown)

	// Install slog default that tees stderr + OTLP-via-LoggerProvider,
	// unless the caller has its own logging chain to compose with.
	if !cfg.DisableSlog {
		slog.SetDefault(slog.New(newSlogHandler(cfg.ServiceName)))
	}

	// Initialize per-package instruments now that providers exist.
	initJobInstruments()

	return makeShutdown(shutdowns), nil
}

// NoOp returns a shutdown that does nothing. Useful in tests so test
// binaries don't try to dial an OTLP endpoint.
func NoOp() func(context.Context) error {
	return func(context.Context) error { return nil }
}

// Meter returns a Meter scoped to the obs package. Callers that want a
// service-scoped meter should use otel.Meter("their-package") instead.
func Meter() metric.Meter {
	return otel.Meter("github.com/atomdrift-project/obs")
}

// Tracer returns a Tracer scoped to the obs package.
func Tracer() trace.Tracer {
	return otel.Tracer("github.com/atomdrift-project/obs")
}

// signalSource carries the resolved endpoint for one signal plus how
// to interpret it. The OTel spec gives different semantics to the base
// env var (a URL prefix; SDK appends /v1/<signal>) vs the per-signal
// env var (a full URL; SDK uses it verbatim). [Config.Endpoint] is
// treated as the base.
type signalSource struct {
	endpoint     string
	appendSignal bool // true → caller must append /v1/<signal> to the path
}

// DefaultEndpoint is the atomdrift Prometheus OTLP metrics receiver on the
// self-hosted Grafana host (`otel`). Used when no OTEL_* endpoint override
// and no [Config.Endpoint] is set.
const DefaultEndpoint = "http://otel:9090/api/v1/otlp"

// DefaultLogsEndpoint is the atomdrift Loki OTLP logs receiver on the same
// host. obs routes logs here when a base endpoint is set but no logs-specific
// endpoint was given, because the base endpoint is the Prometheus OTLP
// receiver, which 404s logs. Override per service with
// OTEL_EXPORTER_OTLP_LOGS_ENDPOINT.
const DefaultLogsEndpoint = "http://otel:3100/otlp/v1/logs"

// resolveLogs applies the fleet log-routing rule on top of the generic
// per-signal resolution: a logs-specific OTEL_EXPORTER_OTLP_LOGS_ENDPOINT is
// honored verbatim, but a bare base endpoint points at the Prometheus OTLP
// receiver — which 404s logs — so logs fall back to [DefaultLogsEndpoint].
func resolveLogs(cfg Config) signalSource {
	sc := signalEndpoint(cfg, "LOGS")
	if sc.appendSignal { // resolved from the base, not a logs-specific endpoint
		return signalSource{endpoint: DefaultLogsEndpoint, appendSignal: false}
	}
	return sc
}

// resolveTraces gates trace push: the fleet has no trace backend and the base
// endpoint is the metrics receiver (which 404s spans), so traces export only
// when a traces-specific OTEL_EXPORTER_OTLP_TRACES_ENDPOINT is set. A
// base-derived endpoint is ignored, leaving the tracer provider span-only
// (no-op export).
func resolveTraces(cfg Config) signalSource {
	sc := signalEndpoint(cfg, "TRACES")
	if sc.appendSignal { // came from the base, not a real trace backend
		return signalSource{}
	}
	return sc
}

// signalEndpoint resolves the endpoint for a specific signal, honoring
// the per-signal OTEL_EXPORTER_OTLP_<SIGNAL>_ENDPOINT before falling
// back to OTEL_EXPORTER_OTLP_ENDPOINT, then [Config.Endpoint], and
// finally [DefaultEndpoint].
func signalEndpoint(cfg Config, signal string) signalSource {
	if v := os.Getenv("OTEL_EXPORTER_OTLP_" + signal + "_ENDPOINT"); v != "" {
		return signalSource{endpoint: v, appendSignal: false}
	}
	if v := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); v != "" {
		return signalSource{endpoint: v, appendSignal: true}
	}
	if cfg.Endpoint != "" {
		return signalSource{endpoint: cfg.Endpoint, appendSignal: true}
	}
	return signalSource{endpoint: DefaultEndpoint, appendSignal: true}
}

// parseEndpoint splits an OTLP endpoint into the host[:port] and path
// pieces the otlphttp exporters expect. Both http:// and https:// are
// accepted; the insecure flag is returned alongside.
func parseEndpoint(raw string, forceInsecure bool) (host, path string, insecure bool, err error) {
	// otlphttp exporters accept either a full URL or a bare host:port.
	// Normalize: if there's no scheme, assume http://.
	if !strings.Contains(raw, "://") {
		raw = "http://" + raw
	}
	u, err := url.Parse(raw)
	if err != nil {
		return "", "", false, fmt.Errorf("parse %q: %w", raw, err)
	}
	host = u.Host
	path = u.Path
	insecure = forceInsecure || u.Scheme == "http"
	return host, path, insecure, nil
}

// signalPath returns the full URL path the exporter should POST to,
// appending /v1/<signal> when the source is a base endpoint (cfg.Endpoint
// or OTEL_EXPORTER_OTLP_ENDPOINT) and leaving the path alone when it
// came from a signal-specific env var.
func signalPath(rawPath string, sc signalSource, signal string) string {
	if !sc.appendSignal {
		return rawPath
	}
	trimmed := strings.TrimSuffix(rawPath, "/")
	return trimmed + "/v1/" + signal
}

func newMetricExporter(ctx context.Context, sc signalSource, forceInsecure bool) (*otlpmetrichttp.Exporter, error) {
	host, path, insecure, err := parseEndpoint(sc.endpoint, forceInsecure)
	if err != nil {
		return nil, err
	}
	path = signalPath(path, sc, "metrics")
	opts := []otlpmetrichttp.Option{otlpmetrichttp.WithEndpoint(host)}
	if path != "" {
		opts = append(opts, otlpmetrichttp.WithURLPath(path))
	}
	if insecure {
		opts = append(opts, otlpmetrichttp.WithInsecure())
	}
	return otlpmetrichttp.New(ctx, opts...)
}

func newTraceExporter(ctx context.Context, sc signalSource, forceInsecure bool) (*otlptrace.Exporter, error) {
	host, path, insecure, err := parseEndpoint(sc.endpoint, forceInsecure)
	if err != nil {
		return nil, err
	}
	path = signalPath(path, sc, "traces")
	opts := []otlptracehttp.Option{otlptracehttp.WithEndpoint(host)}
	if path != "" {
		opts = append(opts, otlptracehttp.WithURLPath(path))
	}
	if insecure {
		opts = append(opts, otlptracehttp.WithInsecure())
	}
	return otlptracehttp.New(ctx, opts...)
}

func newLogExporter(ctx context.Context, sc signalSource, forceInsecure bool) (*otlploghttp.Exporter, error) {
	host, path, insecure, err := parseEndpoint(sc.endpoint, forceInsecure)
	if err != nil {
		return nil, err
	}
	path = signalPath(path, sc, "logs")
	opts := []otlploghttp.Option{otlploghttp.WithEndpoint(host)}
	if path != "" {
		opts = append(opts, otlploghttp.WithURLPath(path))
	}
	if insecure {
		opts = append(opts, otlploghttp.WithInsecure())
	}
	return otlploghttp.New(ctx, opts...)
}

func makeShutdown(fns []func(context.Context) error) func(context.Context) error {
	var once sync.Once
	var stored error
	return func(ctx context.Context) error {
		once.Do(func() {
			// Run shutdowns in reverse order so exporters drain before
			// providers, and so independent failures all get a chance.
			var firstErr error
			for i := len(fns) - 1; i >= 0; i-- {
				if err := fns[i](ctx); err != nil && firstErr == nil {
					firstErr = err
				}
			}
			stored = firstErr
		})
		return stored
	}
}