// Package observability exposes process-local operational metrics. It does not // participate in durable state, authorization or execution decisions. package observability import ( "fmt" "io" "math" "net/http" "sort" "strconv" "strings" "sync" "sync/atomic" "time" ) type Labels map[string]string // Registry accepts only the closed catalog below: neither names nor label // values can grow from caller-controlled request or business data. type Registry struct { mu sync.Mutex families map[string]*family rejected atomic.Uint64 scrape chan struct{} started time.Time } type metricKind string const ( counter metricKind = "counter" gauge metricKind = "gauge" histogram metricKind = "histogram" ) type descriptor struct { kind metricKind help string labels map[string][]string } type family struct { desc descriptor series map[string]*series } type series struct { labels string // Immutable, already escaped and validated against the catalog. value atomic.Uint64 hist distribution } type distribution struct { mu sync.Mutex buckets [11]uint64 count uint64 sum float64 } var bounds = [...]float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10} func NewRegistry() *Registry { r := &Registry{families: make(map[string]*family), scrape: make(chan struct{}, 2), started: time.Now()} for name, d := range catalog() { f := &family{desc: d, series: make(map[string]*series)} if len(d.labels) == 0 { f.series[""] = &series{} } r.families[name] = f } r.initializeProcess() return r } func finite(v float64) bool { return !math.IsNaN(v) && !math.IsInf(v, 0) } func (r *Registry) lookup(name string, kind metricKind, labels Labels) *series { if r == nil { return nil } r.mu.Lock() defer r.mu.Unlock() f, ok := r.families[name] if !ok || f.desc.kind != kind || len(labels) != len(f.desc.labels) { r.rejected.Add(1) return nil } keys := make([]string, 0, len(labels)) for k, v := range labels { valid := false for _, allowed := range f.desc.labels[k] { if v == allowed { valid = true break } } if !valid { r.rejected.Add(1) return nil } keys = append(keys, k) } sort.Strings(keys) var parts []string for _, k := range keys { parts = append(parts, k+`="`+escape(labels[k])+`"`) } encoded := "" if len(parts) > 0 { encoded = "{" + strings.Join(parts, ",") + "}" } s := f.series[encoded] if s == nil { s = &series{labels: encoded} f.series[encoded] = s } return s } func (r *Registry) reject() { if r != nil { r.rejected.Add(1) } } func (r *Registry) AddCounter(name string, delta float64, labels Labels) { if !finite(delta) || delta < 0 { r.reject() return } if s := r.lookup(name, counter, labels); s != nil { add(&s.value, delta) } } // SetCounter only accepts a monotone source, such as database/sql.DBStats. func (r *Registry) SetCounter(name string, value float64, labels Labels) { if !finite(value) || value < 0 { r.reject() return } if s := r.lookup(name, counter, labels); s != nil { for { old := s.value.Load() if value < math.Float64frombits(old) { r.reject() return } if s.value.CompareAndSwap(old, math.Float64bits(value)) { return } } } } func (r *Registry) AddGauge(name string, delta float64, labels Labels) { if !finite(delta) { r.reject() return } if s := r.lookup(name, gauge, labels); s != nil { add(&s.value, delta) } } func (r *Registry) SetGauge(name string, value float64, labels Labels) { if !finite(value) { r.reject() return } if s := r.lookup(name, gauge, labels); s != nil { s.value.Store(math.Float64bits(value)) } } func (r *Registry) ObserveHistogram(name string, value float64, labels Labels) { if !finite(value) || value < 0 { r.reject() return } s := r.lookup(name, histogram, labels) if s == nil { return } h := &s.hist h.mu.Lock() defer h.mu.Unlock() for i, b := range bounds { if value <= b { h.buckets[i]++ } } h.count++ h.sum += value } func add(dst *atomic.Uint64, delta float64) { for { old := dst.Load() next := math.Float64frombits(old) + delta if !finite(next) { return } if dst.CompareAndSwap(old, math.Float64bits(next)) { return } } } func (r *Registry) ServeHTTP(w http.ResponseWriter, req *http.Request) { w.Header().Set("Cache-Control", "no-store") w.Header().Set("X-Content-Type-Options", "nosniff") if req.URL.RawQuery != "" { http.Error(w, "metrics does not accept query parameters", 400) return } select { case r.scrape <- struct{}{}: defer func() { <-r.scrape }() default: w.Header().Set("Retry-After", "1") http.Error(w, "metrics busy", 503) return } w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") _, _ = io.WriteString(w, r.Render()) } // Render snapshots pointers under the registry lock; formatting and network I/O // never hold that lock. Each histogram's buckets/count/sum use one local snapshot. func (r *Registry) Render() string { type view struct { name string desc descriptor series []*series } var views []view r.mu.Lock() for n, f := range r.families { v := view{name: n, desc: f.desc} for _, s := range f.series { v.series = append(v.series, s) } views = append(views, v) } r.mu.Unlock() sort.Slice(views, func(i, j int) bool { return views[i].name < views[j].name }) var b strings.Builder for _, v := range views { fmt.Fprintf(&b, "# HELP %s %s\n# TYPE %s %s\n", v.name, v.desc.help, v.name, v.desc.kind) sort.Slice(v.series, func(i, j int) bool { return v.series[i].labels < v.series[j].labels }) for _, s := range v.series { if v.desc.kind != histogram { fmt.Fprintf(&b, "%s%s %s\n", v.name, s.labels, number(math.Float64frombits(s.value.Load()))) continue } h := &s.hist h.mu.Lock() buckets, count, sum := h.buckets, h.count, h.sum h.mu.Unlock() for i, bound := range bounds { fmt.Fprintf(&b, "%s_bucket%s %d\n", v.name, withLabel(s.labels, "le", number(bound)), buckets[i]) } fmt.Fprintf(&b, "%s_bucket%s %d\n%s_sum%s %s\n%s_count%s %d\n", v.name, withLabel(s.labels, "le", "+Inf"), count, v.name, s.labels, number(sum), v.name, s.labels, count) } } b.WriteString("# HELP orbit_observability_rejected_samples_total Invalid internal metric updates rejected\n# TYPE orbit_observability_rejected_samples_total counter\n") fmt.Fprintf(&b, "orbit_observability_rejected_samples_total %d\n", r.rejected.Load()) return b.String() } // Replacer is safe for concurrent use. Construct it once rather than allocating // its lookup tables for every label of every observed request. var labelEscaper = strings.NewReplacer(`\`, `\\`, `"`, `\"`, "\n", `\n`) func escape(s string) string { return labelEscaper.Replace(s) } func number(v float64) string { return strconv.FormatFloat(v, 'g', -1, 64) } func withLabel(labels, k, v string) string { if labels == "" { return "{" + k + `="` + escape(v) + `"}` } return labels[:len(labels)-1] + "," + k + `="` + escape(v) + `"}` }