| 1 | package telemetry |
| 2 | |
| 3 | import ( |
| 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" |
| 14 | |
| 15 | "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 | ) |
| 21 | |
| 22 | // Config controls optional telemetry integrations. OTLPEndpoint is reserved for |
| 23 | // the OTLP exporter integration; the base package has no collector dependency. |
| 24 | type Config struct { |
| 25 | Version string |
| 26 | Revision string |
| 27 | OTLPEndpoint string |
| 28 | TraceSampleRatio float64 |
| 29 | DebugToken string |
| 30 | } |
| 31 | |
| 32 | // Telemetry owns process-local metrics, tracing, and diagnostics handlers. |
| 33 | type Telemetry struct { |
| 34 | registry *prometheus.Registry |
| 35 | tracerProvider *sdktrace.TracerProvider |
| 36 | tracer trace.Tracer |
| 37 | token string |
| 38 | logger *slog.Logger |
| 39 | |
| 40 | connections prometheus.Gauge |
| 41 | subscriptions prometheus.Gauge |
| 42 | messageCount *prometheus.CounterVec |
| 43 | messageTime *prometheus.HistogramVec |
| 44 | storageCount *prometheus.CounterVec |
| 45 | storageTime *prometheus.HistogramVec |
| 46 | deliveryCount *prometheus.CounterVec |
| 47 | deliveryTime *prometheus.HistogramVec |
| 48 | queueDepth *prometheus.GaugeVec |
| 49 | workCount *prometheus.CounterVec |
| 50 | workTime *prometheus.HistogramVec |
| 51 | } |
| 52 | |
| 53 | // New creates an isolated telemetry instance. It does not mutate global |
| 54 | // OpenTelemetry or Prometheus registries. |
| 55 | func New(ctx context.Context, cfg Config) (*Telemetry, error) { |
| 56 | if ctx == nil { |
| 57 | return nil, errors.New("telemetry: nil context" ) |
| 58 | } |
| 59 | ratio := cfg.TraceSampleRatio |
| 60 | if ratio <= 0 { |
| 61 | ratio = 0.01 |
| 62 | } |
| 63 | if ratio > 1 { |
| 64 | ratio = 1 |
| 65 | } |
| 66 | provider, err := newTracerProvider(ctx, cfg, ratio) |
| 67 | if err != nil { |
| 68 | return nil, err |
| 69 | } |
| 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, err |
| 90 | } |
| 91 | return t, nil |
| 92 | } |
| 93 | |
| 94 | func 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...), nil |
| 98 | } |
| 99 | exporterOptions, err := exporterOptions(cfg.OTLPEndpoint) |
| 100 | if err != nil { |
| 101 | return nil, err |
| 102 | } |
| 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...), nil |
| 109 | } |
| 110 | |
| 111 | func 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)}, nil |
| 117 | } |
| 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.Host |
| 123 | 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, nil |
| 134 | } |
| 135 | |
| 136 | func (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 nil |
| 144 | } |
| 145 | |
| 146 | // Register adds a host-owned collector to this telemetry instance's private registry. |
| 147 | func (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 | } |
| 153 | |
| 154 | // Close flushes and shuts down the tracer provider. |
| 155 | func (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 | } |
| 161 | |
| 162 | // Logger returns the structured logger associated with this telemetry instance. |
| 163 | func (t *Telemetry) Logger() *slog.Logger { return t.logger } |
| 164 | |
| 165 | // Handler returns an authenticated diagnostics handler. |
| 166 | func (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 | } |
| 177 | |
| 178 | func (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 | return |
| 184 | } |
| 185 | next.ServeHTTP(w, r) |
| 186 | }) |
| 187 | } |
| 188 | |
| 189 | // Start creates a span for a bounded relay operation. |
| 190 | func (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 | } |
| 201 | |
| 202 | // ObserveStorage records a storage operation. |
| 203 | func (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 | } |
| 207 | |
| 208 | // Connection changes the current connection gauge. |
| 209 | func (t *Telemetry) Connection(delta int) { t.connections.Add(float64(delta)) } |
| 210 | |
| 211 | // Subscription changes the current subscription gauge. |
| 212 | func (t *Telemetry) Subscription(delta int) { t.subscriptions.Add(float64(delta)) } |
| 213 | |
| 214 | // Delivery records an outbox delivery attempt. |
| 215 | func (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 | } |
| 220 | |
| 221 | // Queue records a bounded queue depth. |
| 222 | func (t *Telemetry) Queue(name string, depth int) { |
| 223 | t.queueDepth.WithLabelValues(bounded(name, queueNames)).Set(float64(depth)) |
| 224 | } |
| 225 | |
| 226 | // ObserveWork records a durable work handler attempt. |
| 227 | func (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 | } |
| 233 | |
| 234 | func latencyBuckets() []float64 { |
| 235 | return []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10} |
| 236 | } |
| 237 | |
| 238 | var 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" : {}} |
| 239 | var outcomeNames = map[string]struct{}{"success" : {}, "ok" : {}, "invalid" : {}, "unauthorized" : {}, "timeout" : {}, "busy" : {}, "closed" : {}, "error" : {}, "paused" : {}} |
| 240 | var queueNames = map[string]struct{}{"inbox" : {}, "outbox" : {}, "write" : {}, "delivery" : {}} |
| 241 | var 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" : {}} |
| 242 | |
| 243 | func bounded(value string, allowed map[string]struct{}) string { |
| 244 | if _, ok := allowed[value]; ok { |
| 245 | return value |
| 246 | } |
| 247 | return "other" |
| 248 | } |
| 249 | |
| 250 | func transport(operation string) string { |
| 251 | if operation == "http" { |
| 252 | return "http" |
| 253 | } |
| 254 | if operation == "websocket" { |
| 255 | return "websocket" |
| 256 | } |
| 257 | return "internal" |
| 258 | } |