Billy D.
d321c9852b
Replace Python handler-base library with Go module providing: - config: environment-based configuration - health: HTTP health/readiness server for k8s probes - natsutil: NATS/JetStream client with msgpack serialization - telemetry: OpenTelemetry tracing and metrics setup - clients: HTTP clients for LLM, embeddings, reranker, STT, TTS - handler: base Handler runner wiring NATS + health + telemetry Implements ADR-0061 Phase 1.
123 lines
3.4 KiB
Go
123 lines
3.4 KiB
Go
// Package telemetry provides OpenTelemetry tracing and metrics setup.
|
|
package telemetry
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"os"
|
|
|
|
"go.opentelemetry.io/otel"
|
|
"go.opentelemetry.io/otel/attribute"
|
|
"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
|
|
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
|
|
"go.opentelemetry.io/otel/metric"
|
|
"go.opentelemetry.io/otel/propagation"
|
|
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"
|
|
)
|
|
|
|
// Config holds the telemetry configuration parameters.
|
|
type Config struct {
|
|
ServiceName string
|
|
ServiceVersion string
|
|
ServiceNamespace string
|
|
DeploymentEnv string
|
|
Enabled bool
|
|
Endpoint string
|
|
}
|
|
|
|
// Provider holds the initialized tracer and meter providers.
|
|
type Provider struct {
|
|
TracerProvider *sdktrace.TracerProvider
|
|
MeterProvider *sdkmetric.MeterProvider
|
|
Tracer trace.Tracer
|
|
Meter metric.Meter
|
|
}
|
|
|
|
// Setup initialises OpenTelemetry tracing and metrics.
|
|
// Returns a Provider and a shutdown function.
|
|
func Setup(ctx context.Context, cfg Config) (*Provider, func(context.Context), error) {
|
|
if !cfg.Enabled {
|
|
slog.Info("OpenTelemetry disabled")
|
|
return &Provider{
|
|
Tracer: otel.Tracer(cfg.ServiceName),
|
|
Meter: otel.Meter(cfg.ServiceName),
|
|
}, func(context.Context) {}, nil
|
|
}
|
|
|
|
hostname, _ := os.Hostname()
|
|
res, err := resource.New(ctx,
|
|
resource.WithAttributes(
|
|
semconv.ServiceNameKey.String(cfg.ServiceName),
|
|
semconv.ServiceVersionKey.String(cfg.ServiceVersion),
|
|
semconv.ServiceNamespaceKey.String(cfg.ServiceNamespace),
|
|
attribute.String("deployment.environment", cfg.DeploymentEnv),
|
|
attribute.String("host.name", hostname),
|
|
),
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
// Trace exporter (gRPC)
|
|
traceExp, err := otlptracegrpc.New(ctx,
|
|
otlptracegrpc.WithEndpoint(stripScheme(cfg.Endpoint)),
|
|
otlptracegrpc.WithInsecure(),
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
tp := sdktrace.NewTracerProvider(
|
|
sdktrace.WithBatcher(traceExp),
|
|
sdktrace.WithResource(res),
|
|
)
|
|
otel.SetTracerProvider(tp)
|
|
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
|
|
propagation.TraceContext{},
|
|
propagation.Baggage{},
|
|
))
|
|
|
|
// Metric exporter (gRPC)
|
|
metricExp, err := otlpmetricgrpc.New(ctx,
|
|
otlpmetricgrpc.WithEndpoint(stripScheme(cfg.Endpoint)),
|
|
otlpmetricgrpc.WithInsecure(),
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
mp := sdkmetric.NewMeterProvider(
|
|
sdkmetric.WithReader(sdkmetric.NewPeriodicReader(metricExp)),
|
|
sdkmetric.WithResource(res),
|
|
)
|
|
otel.SetMeterProvider(mp)
|
|
|
|
slog.Info("OpenTelemetry initialized", "service", cfg.ServiceName, "endpoint", cfg.Endpoint)
|
|
|
|
shutdown := func(ctx context.Context) {
|
|
_ = tp.Shutdown(ctx)
|
|
_ = mp.Shutdown(ctx)
|
|
}
|
|
|
|
return &Provider{
|
|
TracerProvider: tp,
|
|
MeterProvider: mp,
|
|
Tracer: tp.Tracer(cfg.ServiceName, trace.WithInstrumentationVersion(cfg.ServiceVersion)),
|
|
Meter: mp.Meter(cfg.ServiceName),
|
|
}, shutdown, nil
|
|
}
|
|
|
|
// stripScheme removes http:// or https:// from an endpoint for gRPC dialers.
|
|
func stripScheme(endpoint string) string {
|
|
for _, prefix := range []string{"https://", "http://"} {
|
|
if len(endpoint) > len(prefix) && endpoint[:len(prefix)] == prefix {
|
|
return endpoint[len(prefix):]
|
|
}
|
|
}
|
|
return endpoint
|
|
}
|