1package telemetry23import (4 "context"5 "crypto/subtle"6 "errors"7 "fmt"8 "log/slog"9 "net/http"10 "net/http/pprof"11 "net/url"12 "strings"13 "time"1415 "github.com/prometheus/client_golang/prometheus"16 "github.com/prometheus/client_golang/prometheus/promhttp"17 "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"18 sdktrace "go.opentelemetry.io/otel/sdk/trace"19 "go.opentelemetry.io/otel/trace"20)2122// Config controls optional telemetry integrations. OTLPEndpoint is reserved for23// the OTLP exporter integration; the base package has no collector dependency.24type Config struct {25 Version string26 Revision string27 OTLPEndpoint string28 TraceSampleRatio float6429 DebugToken string30}3132// Telemetry owns process-local metrics, tracing, and diagnostics handlers.33type Telemetry struct {34 registry *prometheus.Registry35 tracerProvider *sdktrace.TracerProvider36 tracer trace.Tracer37 token string38 logger *slog.Logger3940 connections prometheus.Gauge41 subscriptions prometheus.Gauge42 messageCount *prometheus.CounterVec43 messageTime *prometheus.HistogramVec44 storageCount *prometheus.CounterVec45 storageTime *prometheus.HistogramVec46 deliveryCount *prometheus.CounterVec47 deliveryTime *prometheus.HistogramVec48 queueDepth *prometheus.GaugeVec49 workCount *prometheus.CounterVec50 workTime *prometheus.HistogramVec51}5253// New creates an isolated telemetry instance. It does not mutate global54// OpenTelemetry or Prometheus registries.55func New(ctx context.Context, cfg Config) (*Telemetry, error) {56 if ctx == nil {57 return nil, errors.New("telemetry: nil context")58 }59 ratio := cfg.TraceSampleRatio60 if ratio <= 0 {61 ratio = 0.0162 }63 if ratio > 1 {64 ratio = 165 }66 provider, err := newTracerProvider(ctx, cfg, ratio)67 if err != nil {68 return nil, err69 }70 t := &Telemetry{71 registry: prometheus.NewRegistry(),72 tracerProvider: provider,73 tracer: provider.Tracer("tinyrelay/telemetry"),74 token: cfg.DebugToken,75 logger: slog.Default().With("service", "tinyrelay", "version", cfg.Version, "revision", cfg.Revision),76 connections: prometheus.NewGauge(prometheus.GaugeOpts{Name: "relay_connections", Help: "Current relay connections."}),77 subscriptions: prometheus.NewGauge(prometheus.GaugeOpts{Name: "relay_subscriptions", Help: "Current relay subscriptions."}),78 messageCount: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "relay_messages_received_total", Help: "Received relay messages."}, []string{"transport", "outcome"}),79 messageTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "relay_message_duration_seconds", Help: "Relay message handling duration.", Buckets: latencyBuckets()}, []string{"operation", "outcome"}),80 storageCount: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "relay_storage_operations_total", Help: "Storage operations."}, []string{"operation", "outcome"}),81 storageTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "relay_storage_operation_duration_seconds", Help: "Storage operation duration.", Buckets: latencyBuckets()}, []string{"operation", "outcome"}),82 deliveryCount: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "relay_delivery_attempts_total", Help: "Outbox delivery attempts."}, []string{"outcome"}),83 deliveryTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "relay_delivery_duration_seconds", Help: "Outbox delivery duration.", Buckets: latencyBuckets()}, []string{"outcome"}),84 queueDepth: prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "relay_queue_depth", Help: "Current queue depth."}, []string{"queue"}),85 workCount: prometheus.NewCounterVec(prometheus.CounterOpts{Name: "relay_work_attempts_total", Help: "Durable work handler attempts."}, []string{"kind", "outcome"}),86 workTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{Name: "relay_work_duration_seconds", Help: "Durable work handler duration.", Buckets: latencyBuckets()}, []string{"kind", "outcome"}),87 }88 if err := t.register(); err != nil {89 return nil, err90 }91 return t, nil92}9394func newTracerProvider(ctx context.Context, cfg Config, ratio float64) (*sdktrace.TracerProvider, error) {95 options := []sdktrace.TracerProviderOption{sdktrace.WithSampler(sdktrace.TraceIDRatioBased(ratio))}96 if cfg.OTLPEndpoint == "" {97 return sdktrace.NewTracerProvider(options...), nil98 }99 exporterOptions, err := exporterOptions(cfg.OTLPEndpoint)100 if err != nil {101 return nil, err102 }103 exporter, err := otlptracehttp.New(ctx, exporterOptions...)104 if err != nil {105 return nil, fmt.Errorf("telemetry: create OTLP exporter: %w", err)106 }107 options = append(options, sdktrace.WithBatcher(exporter))108 return sdktrace.NewTracerProvider(options...), nil109}110111func exporterOptions(endpoint string) ([]otlptracehttp.Option, error) {112 if !strings.Contains(endpoint, "://") {113 if endpoint == "" {114 return nil, fmt.Errorf("telemetry: invalid OTLP endpoint %q", endpoint)115 }116 return []otlptracehttp.Option{otlptracehttp.WithEndpoint(endpoint)}, nil117 }118 parsed, err := url.Parse(endpoint)119 if err != nil {120 return nil, fmt.Errorf("telemetry: invalid OTLP endpoint %q", endpoint)121 }122 host := parsed.Host123 if host == "" {124 return nil, fmt.Errorf("telemetry: invalid OTLP endpoint %q", endpoint)125 }126 options := []otlptracehttp.Option{otlptracehttp.WithEndpoint(host)}127 if parsed.Scheme == "http" {128 options = append(options, otlptracehttp.WithInsecure())129 }130 if parsed.Path != "" && parsed.Path != "/" {131 options = append(options, otlptracehttp.WithURLPath(parsed.Path))132 }133 return options, nil134}135136func (t *Telemetry) register() error {137 collectors := []prometheus.Collector{t.connections, t.subscriptions, t.messageCount, t.messageTime, t.storageCount, t.storageTime, t.deliveryCount, t.deliveryTime, t.queueDepth, t.workCount, t.workTime, prometheus.NewGoCollector()}138 for _, collector := range collectors {139 if err := t.registry.Register(collector); err != nil {140 return fmt.Errorf("telemetry: register collector: %w", err)141 }142 }143 return nil144}145146// Register adds a host-owned collector to this telemetry instance's private registry.147func (t *Telemetry) Register(collector prometheus.Collector) error {148 if collector == nil {149 return errors.New("telemetry: nil collector")150 }151 return t.registry.Register(collector)152}153154// Close flushes and shuts down the tracer provider.155func (t *Telemetry) Close(ctx context.Context) error {156 if ctx == nil {157 return errors.New("telemetry: nil context")158 }159 return t.tracerProvider.Shutdown(ctx)160}161162// Logger returns the structured logger associated with this telemetry instance.163func (t *Telemetry) Logger() *slog.Logger { return t.logger }164165// Handler returns an authenticated diagnostics handler.166func (t *Telemetry) Handler() http.Handler {167 mux := http.NewServeMux()168 mux.Handle("/metrics", promhttp.HandlerFor(t.registry, promhttp.HandlerOpts{}))169 mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })170 mux.HandleFunc("/debug/pprof/", pprof.Index)171 mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)172 mux.HandleFunc("/debug/pprof/profile", pprof.Profile)173 mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)174 mux.HandleFunc("/debug/pprof/trace", pprof.Trace)175 return t.auth(mux)176}177178func (t *Telemetry) auth(next http.Handler) http.Handler {179 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {180 token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")181 if t.token == "" || subtle.ConstantTimeCompare([]byte(token), []byte(t.token)) != 1 {182 http.Error(w, "unauthorized", http.StatusUnauthorized)183 return184 }185 next.ServeHTTP(w, r)186 })187}188189// Start creates a span for a bounded relay operation.190func (t *Telemetry) Start(ctx context.Context, operation string) (context.Context, func(string)) {191 name := bounded(operation, operationNames)192 spanCtx, span := t.tracer.Start(ctx, "relay."+name)193 started := time.Now()194 return spanCtx, func(outcome string) {195 result := bounded(outcome, outcomeNames)196 span.End()197 t.messageTime.WithLabelValues(name, result).Observe(time.Since(started).Seconds())198 t.messageCount.WithLabelValues(transport(name), result).Inc()199 }200}201202// ObserveStorage records a storage operation.203func (t *Telemetry) ObserveStorage(operation, outcome string, duration time.Duration) {204 t.storageCount.WithLabelValues(bounded(operation, operationNames), bounded(outcome, outcomeNames)).Inc()205 t.storageTime.WithLabelValues(bounded(operation, operationNames), bounded(outcome, outcomeNames)).Observe(duration.Seconds())206}207208// Connection changes the current connection gauge.209func (t *Telemetry) Connection(delta int) { t.connections.Add(float64(delta)) }210211// Subscription changes the current subscription gauge.212func (t *Telemetry) Subscription(delta int) { t.subscriptions.Add(float64(delta)) }213214// Delivery records an outbox delivery attempt.215func (t *Telemetry) Delivery(outcome string, duration time.Duration) {216 result := bounded(outcome, outcomeNames)217 t.deliveryCount.WithLabelValues(result).Inc()218 t.deliveryTime.WithLabelValues(result).Observe(duration.Seconds())219}220221// Queue records a bounded queue depth.222func (t *Telemetry) Queue(name string, depth int) {223 t.queueDepth.WithLabelValues(bounded(name, queueNames)).Set(float64(depth))224}225226// ObserveWork records a durable work handler attempt.227func (t *Telemetry) ObserveWork(kind, outcome string, duration time.Duration) {228 result := bounded(outcome, outcomeNames)229 name := bounded(kind, workNames)230 t.workCount.WithLabelValues(name, result).Inc()231 t.workTime.WithLabelValues(name, result).Observe(duration.Seconds())232}233234func latencyBuckets() []float64 {235 return []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10}236}237238var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browse-download": {}, "push": {}, "mcp": {}, "rooms": {}, "blob-expiry": {}, "view": {}}239var outcomeNames = map[string]struct{}{"success": {}, "ok": {}, "invalid": {}, "unauthorized": {}, "timeout": {}, "busy": {}, "closed": {}, "error": {}, "paused": {}}240var queueNames = map[string]struct{}{"inbox": {}, "outbox": {}, "write": {}, "delivery": {}}241var workNames = map[string]struct{}{"delivery": {}, "push": {}, "replication": {}, "webhook": {}, "job": {}, "records-projection": {}, "site-mirror": {}, "git-metadata": {}, "catalog-owner": {}, "callback": {}, "delivery-discovery": {}, "notification-delivery": {}, "notification-broadcast": {}, "notification-push": {}, "notification": {}, "callback-delivery": {}, "view-transform": {}}242243func bounded(value string, allowed map[string]struct{}) string {244 if _, ok := allowed[value]; ok {245 return value246 }247 return "other"248}249250func transport(operation string) string {251 if operation == "http" {252 return "http"253 }254 if operation == "websocket" {255 return "websocket"256 }257 return "internal"258}