initial commit

internal/metrics/metrics.go

@@ -0,0 +1,478 @@
+// Package metrics holds the Prometheus collectors and the in-process
+// time-series buffers that back the admin UI's charts.
+//
+// The UI needs finer-grained series than Prometheus's default histogram gives,
+// so a small fixed-bucket ring per metric is kept in memory: cheap, bounded,
+// and enough to draw a dashboard without shipping raw samples to a browser.
+package metrics
+
+import (
+	"math"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"net/http"
+)
+
+// Namespace prefixes every metric name.
+const Namespace = "fancydns"
+
+// Counters for the request pipeline.
+var (
+	queriesTotal   prometheus.Counter
+	blockedTotal   prometheus.Counter
+	cacheHits      prometheus.Counter
+	cacheMisses    prometheus.Counter
+	cacheStaleHits prometheus.Counter
+	upstreamErrors prometheus.Counter
+	dnsErrors      prometheus.Counter
+	dropsTotal     prometheus.Counter
+	refusedTotal   prometheus.Counter
+)
+
+// Gauges for live state.
+var (
+	upstreamHealth *prometheus.GaugeVec
+	concurrentNow  prometheus.Gauge
+	listCount      prometheus.Gauge
+	listBytes      prometheus.Gauge
+	aclGroups      prometheus.Gauge
+	localRecords   prometheus.Gauge
+	up             prometheus.Gauge
+)
+
+// Histograms for latency, with buckets that matter for DNS: sub-millisecond
+// cache hits, a few milliseconds for a local answer, tens for an upstream.
+var (
+	queryDuration *prometheus.HistogramVec
+	upstreamDur   *prometheus.HistogramVec
+	parseDuration prometheus.Histogram
+)
+
+var registerOnce sync.Once
+
+// Registry is the registry served on /metrics.
+var Registry = prometheus.NewRegistry()
+
+// initCollectors registers everything exactly once. It is safe to call from
+// several components.
+func initCollectors() {
+	registerOnce.Do(func() {
+		queriesTotal = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "queries_total",
+			Help: "Total DNS queries received across all transports.",
+		})
+		blockedTotal = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "blocked_total",
+			Help: "Queries answered with a blocklist response.",
+		})
+		cacheHits = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "cache_hits_total",
+			Help: "Cache hits (fresh).",
+		})
+		cacheMisses = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "cache_misses_total",
+			Help: "Cache misses that required an upstream lookup.",
+		})
+		cacheStaleHits = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "cache_stale_hits_total",
+			Help: "Answers served from the serve-stale window.",
+		})
+		upstreamErrors = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "upstream_errors_total",
+			Help: "Upstream exchange failures.",
+		})
+		dnsErrors = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "errors_total",
+			Help: "Responses with a non-success RCODE.",
+		})
+		dropsTotal = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "dropped_total",
+			Help: "Queries dropped without a response (blocklist drop or overload).",
+		})
+		refusedTotal = prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace, Name: "refused_total",
+			Help: "Queries answered REFUSED by ACL policy.",
+		})
+		upstreamHealth = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: Namespace, Name: "upstream_health",
+			Help: "1 when an upstream is up, 0 when down.",
+		}, []string{"addr", "proto"})
+		concurrentNow = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace, Name: "concurrent_queries",
+			Help: "Queries currently in flight.",
+		})
+		listCount = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace, Name: "blocklist_rules",
+			Help: "Total compiled blocklist rules.",
+		})
+		listBytes = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace, Name: "blocklist_source_bytes",
+			Help: "Bytes of source data compiled into the blocklist.",
+		})
+		aclGroups = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace, Name: "acl_groups",
+			Help: "Number of configured ACL groups.",
+		})
+		localRecords = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace, Name: "local_records",
+			Help: "Records in the local zone store.",
+		})
+		up = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace, Name: "up",
+			Help: "1 if the process is serving, 0 otherwise.",
+		})
+		queryDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: Namespace, Name: "query_duration_seconds",
+			Help:    "End-to-end query handling time.",
+			Buckets: []float64{0.0002, 0.0005, 0.001, 0.002, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5},
+		}, []string{"transport", "outcome"})
+		upstreamDur = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: Namespace, Name: "upstream_duration_seconds",
+			Help:    "Time spent talking to a specific upstream.",
+			Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2},
+		}, []string{"addr"})
+		parseDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: Namespace, Name: "blocklist_compile_seconds",
+			Help:    "Time to fetch and compile all blocklist sources.",
+			Buckets: []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30},
+		})
+
+		Registry.MustRegister(
+			queriesTotal, blockedTotal, cacheHits, cacheMisses, cacheStaleHits,
+			upstreamErrors, dnsErrors, dropsTotal, refusedTotal, upstreamHealth,
+			concurrentNow, listCount, listBytes, aclGroups, localRecords, up,
+			queryDuration, upstreamDur, parseDuration,
+			collectors.NewGoCollector(),
+			collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
+		)
+	})
+}
+
+// Outcome labels for queryDuration.
+const (
+	OutcomeCache    = "cache"
+	OutcomeBlocked  = "blocked"
+	OutcomeLocal    = "local"
+	OutcomeUpstream = "upstream"
+	OutcomeError    = "error"
+	OutcomeRefused  = "refused"
+	OutcomeDropped  = "dropped"
+)
+
+// Record is the single entry point for a pipeline outcome. Both the Prometheus
+// counters and the in-process atomics the dashboard reads are bumped here, so
+// the two views can never disagree.
+//
+// Callers must not also call CountBlocked/CountError/etc: that would double
+// count. Those helpers exist for the few places that report an outcome without
+// going through Record.
+func Record(transport, outcome string, d time.Duration) {
+	initCollectors()
+	queriesTotal.Inc()
+	queries.Add(1)
+	if d > 0 {
+		queryDuration.WithLabelValues(transport, outcome).Observe(d.Seconds())
+	}
+	switch outcome {
+	case OutcomeCache:
+		cacheHits.Inc()
+		hits.Add(1)
+	case OutcomeBlocked:
+		blockedTotal.Inc()
+		blocked.Add(1)
+	case OutcomeError:
+		dnsErrors.Inc()
+		errs.Add(1)
+	case OutcomeDropped:
+		dropsTotal.Inc()
+		drops.Add(1)
+		// A drop is both a drop and an error for the operator's purposes.
+		errs.Add(1)
+	case OutcomeRefused:
+		refusedTotal.Inc()
+		refused.Add(1)
+	}
+}
+
+// RecordMiss counts a cache miss. A miss is not an error: it just means the
+// cache could not answer, so only the Prometheus counter moves.
+func RecordMiss() { initCollectors(); cacheMisses.Inc() }
+
+// RecordStaleHit counts a serve-stale hit.
+func RecordStaleHit() { initCollectors(); cacheStaleHits.Inc() }
+
+// RecordUpstreamError counts an upstream failure.
+func RecordUpstreamError() { initCollectors(); upstreamErrors.Inc() }
+
+// RecordUpstreamLatency observes a single upstream exchange.
+func RecordUpstreamLatency(addr string, d time.Duration) {
+	initCollectors()
+	upstreamDur.WithLabelValues(addr).Observe(d.Seconds())
+}
+
+// SetConcurrent publishes the in-flight gauge.
+func SetConcurrent(n int) { initCollectors(); concurrentNow.Set(float64(n)) }
+
+// SetUp publishes the liveness gauge.
+func SetUp(v bool) {
+	initCollectors()
+	if v {
+		up.Set(1)
+		return
+	}
+	up.Set(0)
+}
+
+// SetUpstreamHealth publishes per-upstream health.
+func SetUpstreamHealth(addr, proto string, healthy bool) {
+	initCollectors()
+	v := 0.0
+	if healthy {
+		v = 1
+	}
+	upstreamHealth.WithLabelValues(addr, proto).Set(v)
+}
+
+// SetBlocklistStats publishes the compiled rule count and source bytes.
+func SetBlocklistStats(rules int, sourceBytes int64) {
+	initCollectors()
+	listCount.Set(float64(rules))
+	listBytes.Set(float64(sourceBytes))
+}
+
+// SetACLGroups publishes the group count.
+func SetACLGroups(n int) { initCollectors(); aclGroups.Set(float64(n)) }
+
+// SetLocalRecords publishes the local zone record count.
+func SetLocalRecords(n int) { initCollectors(); localRecords.Set(float64(n)) }
+
+// RecordBlocklistCompile observes a full compile pass.
+func RecordBlocklistCompile(d time.Duration) {
+	initCollectors()
+	parseDuration.Observe(d.Seconds())
+}
+
+// Handler returns the HTTP handler for /metrics.
+func Handler() http.Handler {
+	initCollectors()
+	return promhttp.HandlerFor(Registry, promhttp.HandlerOpts{
+		Registry:            Registry,
+		EnableOpenMetrics:   true,
+		MaxRequestsInFlight: 4,
+	})
+}
+
+// Totals is a point-in-time snapshot of the counters, used by the dashboard.
+type Totals struct {
+	Queries    uint64
+	Blocked    uint64
+	Hits       uint64
+	Misses     uint64
+	StaleHits  uint64
+	Errors     uint64
+	Drops      uint64
+	Refused    uint64
+	Concurrent int64
+}
+
+var (
+	queries   atomic.Uint64
+	blocked   atomic.Uint64
+	hits      atomic.Uint64
+	misses    atomic.Uint64
+	staleHits atomic.Uint64
+	errs      atomic.Uint64
+	drops     atomic.Uint64
+	refused   atomic.Uint64
+	inflight  atomic.Int64
+)
+
+// TotalsNow returns the current totals. These mirror the Prometheus counters
+// (both are incremented together) so the dashboard does not have to scrape.
+func TotalsNow() Totals {
+	return Totals{
+		Queries:    queries.Load(),
+		Blocked:    blocked.Load(),
+		Hits:       hits.Load(),
+		Misses:     misses.Load(),
+		StaleHits:  staleHits.Load(),
+		Errors:     errs.Load(),
+		Drops:      drops.Load(),
+		Refused:    refused.Load(),
+		Concurrent: inflight.Load(),
+	}
+}
+
+func bumpCounter(c *atomic.Uint64) func() {
+	c.Add(1)
+	return func() { c.Add(^uint64(0)) }
+}
+
+// BeginQuery marks a query as in flight and returns its completion func.
+func BeginQuery() func() {
+	inflight.Add(1)
+	queries.Add(1)
+	return func() { inflight.Add(-1) }
+}
+
+// CountBlocked increments the blocked counter.
+func CountBlocked() { blocked.Add(1) }
+
+// CountError increments the error counter.
+func CountError() { errs.Add(1) }
+
+// CountDrop increments the drop counter.
+func CountDrop() { drops.Add(1) }
+
+// CountRefused increments the refused counter.
+func CountRefused() { refused.Add(1) }
+
+// Series is a fixed-window time series for the UI charts.
+type Series struct {
+	// Buckets are the sample timestamps (epoch seconds), oldest first.
+	Buckets []int64 `json:"t"`
+	Values  []int64 `json:"v"`
+	// ResolutionMS is the bucket width.
+	ResolutionMS int64 `json:"resolution_ms"`
+}
+
+// Ring is a fixed-bucket counter ring: a circular array of counts indexed by
+// bucket number. Lock-free on the write path, which matters because it is hit
+// once per DNS query.
+type Ring struct {
+	buckets []atomic.Int64
+	res     time.Duration
+	start   atomic.Int64 // first bucket timestamp, epoch seconds
+	pos     atomic.Uint32
+	sealed  atomic.Bool
+	count   int
+}
+
+// NewRing builds a ring of n buckets at the given resolution.
+func NewRing(n int, res time.Duration) *Ring {
+	if n <= 0 {
+		n = 120
+	}
+	if res <= 0 {
+		res = 5 * time.Second
+	}
+	r := &Ring{buckets: make([]atomic.Int64, n), res: res, count: n}
+	r.start.Store(time.Now().Unix())
+	return r
+}
+
+// Inc bumps the current bucket by 1, rotating when the window advances.
+func (r *Ring) Inc() { r.Add(1) }
+
+// Add bumps the current bucket.
+func (r *Ring) Add(delta int64) {
+	if r == nil {
+		return
+	}
+	now := time.Now().Unix()
+	bucket := now / int64(r.res.Seconds())
+	start := r.start.Load()
+	// If more than count buckets elapsed, we skipped a full window: reset
+	// everything so a long idle period does not report phantom zeros followed
+	// by a sudden spike.
+	if bucket-start >= int64(r.count) {
+		if r.start.CompareAndSwap(start, bucket-int64(r.count)+1) {
+			for i := range r.buckets {
+				r.buckets[i].Store(0)
+			}
+		}
+		start = r.start.Load()
+	}
+	if bucket < start {
+		return
+	}
+	idx := int(bucket - start)
+	if idx >= r.count {
+		idx = r.count - 1
+	}
+	// Store at the position implied by the bucket so readers can map index to
+	// time without extra bookkeeping.
+	r.buckets[idx%r.count].Add(delta)
+}
+
+// AddAt bumps an explicit bucket (used by tests).
+func (r *Ring) AddAt(bucket int64, delta int64) {
+	start := r.start.Load()
+	idx := int(bucket - start)
+	if idx < 0 || idx >= r.count {
+		return
+	}
+	r.buckets[idx%r.count].Add(delta)
+}
+
+// Series snapshots the ring into aligned arrays.
+func (r *Ring) Snapshot() ([]int64, []int64) {
+	if r == nil {
+		return nil, nil
+	}
+	start := r.start.Load()
+	now := time.Now().Unix()
+	step := int64(r.res.Seconds())
+	ts := make([]int64, r.count)
+	vs := make([]int64, r.count)
+	for i := 0; i < r.count; i++ {
+		b := start + int64(i)
+		ts[i] = b * step
+		vs[i] = r.buckets[i%r.count].Load()
+	}
+	// Trim leading buckets that are in the future (ring not yet filled) so the
+	// chart starts at "now" rather than at zero time.
+	if now > ts[len(ts)-1] {
+		shift := int((now-ts[len(ts)-1])/step) + 1
+		if shift >= len(ts) {
+			return []int64{now}, []int64{0}
+		}
+		ts = append(ts[:0], ts[shift:]...)
+		vs = append(vs[:0], vs[shift:]...)
+	}
+	return ts, vs
+}
+
+// Series is the snapshot form.
+func (r *Ring) Series() Series {
+	ts, vs := r.Snapshot()
+	return Series{Buckets: ts, Values: vs, ResolutionMS: r.res.Milliseconds()}
+}
+
+// LCG is a tiny deterministic PRNG for jitter, so the test suite does not
+// depend on the global rand.
+type LCG struct{ s uint64 }
+
+func NewLCG(seed uint64) *LCG { return &LCG{s: seed | 1} }
+
+func (l *LCG) Next() uint64 {
+	l.s = l.s*6364136223846793005 + 1442695040888963407
+	return l.s >> 11
+}
+
+func (l *LCG) Float() float64 { return float64(l.Next()) / float64(1<<53) }
+
+// SafeFloat converts an int count to a float for charting, guarding NaN.
+func SafeFloat(n int64) float64 {
+	if n <= 0 {
+		return 0
+	}
+	return float64(n)
+}
+
+// Percent computes a safe percentage.
+func Percent(part, total int64) float64 {
+	if total <= 0 {
+		return 0
+	}
+	p := float64(part) / float64(total) * 100
+	if math.IsNaN(p) || math.IsInf(p, 0) {
+		return 0
+	}
+	return p
+}