admin dashboard, metrics and telemetry

- embedded React admin UI at /admin/ui (go:embed), login with ADMIN_SECRET
  via signed HttpOnly session cookie, X-ADMIN-KEY still works for scripts
- admin API: overview stats, indexing queue, addon management (priority,
  reindex, delete/restore/purge, bulk), whitelist CRUD, duplicate hashes,
  traffic, metrics explorer, services, events/audit log, system info
- built-in metrics: request latency per route, visitors, top addons and
  searches, indexer/scraper counters; minute buckets in the DB with hourly
  rollups and retention; optional Prometheus /metrics (METRICS_TOKEN)
- POST /back/telemetry for indexers/scrapers (metrics, events, heartbeat)
  and a stdlib-only Go client in telemetry/
- auth hardening: empty secrets no longer allow access, constant-time
  compare, CORS not applied to /admin and /back, TRUSTED_PROXIES setting
- Dockerfile builds the UI and no longer bakes config.yaml into the image;
  CI builds the UI, runs tests and builds arm64 with CGO

metrics/aggregator.go

@@ -0,0 +1,286 @@
+package metrics
+
+import (
+	"encoding/json"
+	"math"
+	"sort"
+	"sync"
+	"sync/atomic"
+)
+
+// disabled turns all recording into no-ops (METRICS_ENABLED=false).
+var disabled atomic.Bool
+
+// SetEnabled switches metric recording on or off.
+func SetEnabled(on bool) { disabled.Store(!on) }
+
+// Agg is the aggregate of all samples of one series inside one bucket.
+type Agg struct {
+	Count  int64
+	Sum    float64
+	Min    float64
+	Max    float64
+	Last   float64
+	LastTs int64
+	Hist   []uint64 // len(bounds)+1, last element is +Inf
+}
+
+func (a *Agg) add(v float64, ts int64, bounds []float64) {
+	if a.Count == 0 {
+		a.Min, a.Max = v, v
+	} else {
+		a.Min = math.Min(a.Min, v)
+		a.Max = math.Max(a.Max, v)
+	}
+	a.Count++
+	a.Sum += v
+	if ts >= a.LastTs {
+		a.Last, a.LastTs = v, ts
+	}
+	if len(bounds) > 0 {
+		if a.Hist == nil {
+			a.Hist = make([]uint64, len(bounds)+1)
+		}
+		a.Hist[bucketIndex(bounds, v)]++
+	}
+}
+
+func (a *Agg) addSummary(count int64, sum, min, max float64, ts int64) {
+	if count <= 0 {
+		return
+	}
+	if a.Count == 0 {
+		a.Min, a.Max = min, max
+	} else {
+		a.Min = math.Min(a.Min, min)
+		a.Max = math.Max(a.Max, max)
+	}
+	a.Count += count
+	a.Sum += sum
+	if ts >= a.LastTs {
+		a.Last, a.LastTs = sum/float64(count), ts
+	}
+}
+
+// merge folds b into a.
+func (a *Agg) merge(b *Agg) {
+	if b == nil || b.Count == 0 {
+		return
+	}
+	if a.Count == 0 {
+		a.Min, a.Max = b.Min, b.Max
+	} else {
+		a.Min = math.Min(a.Min, b.Min)
+		a.Max = math.Max(a.Max, b.Max)
+	}
+	a.Count += b.Count
+	a.Sum += b.Sum
+	if b.LastTs >= a.LastTs {
+		a.Last, a.LastTs = b.Last, b.LastTs
+	}
+	if len(b.Hist) > 0 {
+		if len(a.Hist) == 0 {
+			a.Hist = make([]uint64, len(b.Hist))
+		}
+		if len(a.Hist) == len(b.Hist) {
+			for i, v := range b.Hist {
+				a.Hist[i] += v
+			}
+		}
+	}
+}
+
+func (a *Agg) clone() *Agg {
+	c := *a
+	if a.Hist != nil {
+		c.Hist = append([]uint64(nil), a.Hist...)
+	}
+	return &c
+}
+
+func (a *Agg) histJSON() string {
+	if len(a.Hist) == 0 {
+		return ""
+	}
+	b, _ := json.Marshal(a.Hist)
+	return string(b)
+}
+
+func bucketIndex(bounds []float64, v float64) int {
+	return sort.SearchFloat64s(bounds, v) // first bound >= v; len(bounds) = +Inf
+}
+
+// quantile estimates the q-quantile from histogram buckets with linear
+// interpolation inside the bucket (like Prometheus), clamped to [min, max].
+func quantile(a *Agg, bounds []float64, q float64) (float64, bool) {
+	if a == nil || a.Count == 0 || len(a.Hist) != len(bounds)+1 {
+		return 0, false
+	}
+	var total uint64
+	for _, c := range a.Hist {
+		total += c
+	}
+	if total == 0 {
+		return 0, false
+	}
+	rank := q * float64(total)
+	var cum uint64
+	for i, c := range a.Hist {
+		if c == 0 {
+			continue
+		}
+		if float64(cum+c) >= rank {
+			lower := a.Min
+			if i > 0 {
+				lower = math.Max(bounds[i-1], a.Min)
+			}
+			upper := a.Max
+			if i < len(bounds) {
+				upper = math.Min(bounds[i], a.Max)
+			}
+			if upper < lower {
+				upper = lower
+			}
+			frac := (rank - float64(cum)) / float64(c)
+			return lower + (upper-lower)*frac, true
+		}
+		cum += c
+	}
+	return a.Max, true
+}
+
+type pointKey struct {
+	s      *Series
+	bucket int64
+}
+
+// cumulative is the since-start state of a series, used for /metrics.
+type cumulative struct {
+	count int64
+	sum   float64
+	last  float64
+	hist  []uint64
+}
+
+// aggregator holds the not yet persisted minute buckets and cumulative state.
+type aggregator struct {
+	mu   sync.Mutex
+	open map[pointKey]*Agg
+	cum  map[*Series]*cumulative
+}
+
+var defaultAgg = &aggregator{
+	open: make(map[pointKey]*Agg),
+	cum:  make(map[*Series]*cumulative),
+}
+
+func minuteOf(ts int64) int64 { return ts - ts%60 }
+
+func (g *aggregator) entry(s *Series, ts int64) (*Agg, *cumulative) {
+	k := pointKey{s, minuteOf(ts)}
+	a := g.open[k]
+	if a == nil {
+		a = &Agg{}
+		g.open[k] = a
+	}
+	c := g.cum[s]
+	if c == nil {
+		c = &cumulative{}
+		g.cum[s] = c
+	}
+	return a, c
+}
+
+func (g *aggregator) observe(s *Series, ts int64, v float64) {
+	if disabled.Load() {
+		return
+	}
+	g.mu.Lock()
+	a, c := g.entry(s, ts)
+	a.add(v, ts, s.Bounds)
+	c.count++
+	c.sum += v
+	c.last = v
+	if len(s.Bounds) > 0 {
+		if c.hist == nil {
+			c.hist = make([]uint64, len(s.Bounds)+1)
+		}
+		c.hist[bucketIndex(s.Bounds, v)]++
+	}
+	g.mu.Unlock()
+	reg.touch(s, ts)
+}
+
+func (g *aggregator) observeSummary(s *Series, ts int64, count int64, sum, min, max float64) {
+	if disabled.Load() {
+		return
+	}
+	g.mu.Lock()
+	a, c := g.entry(s, ts)
+	a.addSummary(count, sum, min, max, ts)
+	c.count += count
+	c.sum += sum
+	c.last = sum / float64(count)
+	g.mu.Unlock()
+	reg.touch(s, ts)
+}
+
+// takeClosed removes and returns all buckets that started before `before`.
+func (g *aggregator) takeClosed(before int64) map[pointKey]*Agg {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	out := make(map[pointKey]*Agg)
+	for k, a := range g.open {
+		if k.bucket < before {
+			out[k] = a
+			delete(g.open, k)
+		}
+	}
+	return out
+}
+
+// putBack re-queues buckets after a failed flush.
+func (g *aggregator) putBack(points map[pointKey]*Agg) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	for k, a := range points {
+		if cur := g.open[k]; cur != nil {
+			cur.merge(a)
+		} else {
+			g.open[k] = a
+		}
+	}
+}
+
+// snapshotOpen copies the open buckets of the given series.
+func (g *aggregator) snapshotOpen(want map[*Series]bool) map[pointKey]*Agg {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	out := make(map[pointKey]*Agg)
+	for k, a := range g.open {
+		if want[k.s] {
+			out[k] = a.clone()
+		}
+	}
+	return out
+}
+
+func (g *aggregator) openCount() int {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return len(g.open)
+}
+
+func (g *aggregator) snapshotCum() map[*Series]cumulative {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	out := make(map[*Series]cumulative, len(g.cum))
+	for s, c := range g.cum {
+		cc := *c
+		if c.hist != nil {
+			cc.hist = append([]uint64(nil), c.hist...)
+		}
+		out[s] = cc
+	}
+	return out
+}

metrics/internal.go

@@ -0,0 +1,38 @@
+package metrics
+
+// Metrics emitted by the API itself (service "main").
+
+// HTTPBoundsMs are the latency histogram buckets in milliseconds.
+var HTTPBoundsMs = []float64{1, 2, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000}
+
+var (
+	HTTPDuration = NewHistogram("http_request_duration_ms", "HTTP request latency in milliseconds", HTTPBoundsMs,
+		"method", "route", "status", "group", "client")
+	HTTPResponseBytes = NewCounter("http_response_bytes_total", "HTTP response body bytes", "group")
+
+	IndexerClaims        = NewCounter("indexer_claims_total", "Indexing jobs requested by indexers, by outcome", "queue")
+	IndexerTimeoutsReset = NewCounter("indexer_timeouts_reset_total", "Indexing claims released after the timeout")
+	IndexerSubmissions   = NewCounter("indexer_submissions_total", "Indexing results submitted", "result")
+	IndexerJobDuration   = NewHistogram("indexer_job_duration_seconds", "Time from claim to submitted result",
+		[]float64{1, 5, 15, 30, 60, 120, 300, 600, 1200, 1800, 3600})
+	IndexerFilesPerSubmission = NewHistogram("indexer_files_per_submission", "Files per indexing result",
+		[]float64{1, 10, 50, 100, 500, 1000, 5000, 10000, 50000})
+	IndexerBytesIndexed = NewCounter("indexer_bytes_indexed_total", "Addon bytes indexed")
+
+	ScraperAddons     = NewCounter("scraper_addons_total", "Addon reports from scrapers", "action")
+	ScraperDeletes    = NewCounter("scraper_addons_deleted_total", "Addon deletions requested by scrapers", "result")
+	DependencyReports = NewCounter("dependency_reports_total", "Dependency reports received")
+	DependencyEdges   = NewCounter("dependency_edges_reported_total", "Dependency edges stored")
+	DuplicatesFound   = NewHistogram("duplicates_found_addons", "Addons sharing files with a looked-up addon",
+		[]float64{0, 1, 2, 5, 10, 25, 50, 100})
+	DiscordWebhooks = NewCounter("discord_webhook_total", "Discord webhook sends", "kind", "result")
+	AdminActions    = NewCounter("admin_actions_total", "Admin dashboard actions", "action")
+
+	AddonsTotal    = NewGauge("addons_total", "Addons in the database (not deleted)")
+	QueueDepth     = NewGauge("queue_depth", "Addons waiting to be indexed", "kind")
+	BeingIndexed   = NewGauge("being_indexed", "Addons currently claimed by an indexer")
+	VisitorsUnique = NewGauge("visitors_unique", "Unique website visitors so far today (approximate)", "kind")
+
+	TelemetrySamples = NewCounter("telemetry_samples_total", "Telemetry samples received", "result")
+	PointsDropped    = NewCounter("metrics_points_dropped_total", "Metric points dropped after failed writes")
+)

metrics/metrics_test.go

@@ -0,0 +1,216 @@
+package metrics
+
+import (
+	"math"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/models"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+)
+
+func setupDB(t *testing.T) {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "m.db")), &gorm.Config{Logger: logger.Discard})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&models.Addon{}, &models.MetricSeries{}, &models.MetricPoint{},
+		&models.ServiceInstance{}, &models.ServiceEvent{}, &models.DailyTopItem{}); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if sqlDB, err := db.DB(); err == nil {
+			sqlDB.Close()
+		}
+	})
+	initializers.DB = db
+	initializers.MetricsMinuteRetentionDays = 7
+	initializers.MetricsHourRetentionDays = 180
+
+	// fresh in-memory state per test
+	reg = &registry{series: map[seriesKey]*Series{}, perName: map[string]int{}, lastSeen: map[*Series]int64{}}
+	defaultAgg = &aggregator{open: map[pointKey]*Agg{}, cum: map[*Series]*cumulative{}}
+	seriesID = map[*Series]uint{}
+	rolledUntil.Store(0)
+	for _, f := range families {
+		f.cache = map[string]*Series{}
+	}
+}
+
+func TestQuantileInterpolation(t *testing.T) {
+	bounds := []float64{10, 20, 30}
+	a := &Agg{}
+	for i := 1; i <= 100; i++ {
+		a.add(float64(i%30)+0.5, 0, bounds)
+	}
+	p50, ok := quantile(a, bounds, 0.5)
+	if !ok || p50 < 10 || p50 > 20 {
+		t.Fatalf("p50 = %v, want within (10, 20]", p50)
+	}
+	p99, _ := quantile(a, bounds, 0.99)
+	if p99 > a.Max || p99 < 20 {
+		t.Fatalf("p99 = %v, want in [20, max=%v]", p99, a.Max)
+	}
+	if _, ok := quantile(&Agg{}, bounds, 0.5); ok {
+		t.Fatal("empty histogram must not produce a quantile")
+	}
+}
+
+func TestAggMerge(t *testing.T) {
+	a, b := &Agg{}, &Agg{}
+	a.add(5, 100, nil)
+	a.add(1, 101, nil)
+	b.add(9, 99, nil)
+	a.merge(b)
+	if a.Count != 3 || a.Sum != 15 || a.Min != 1 || a.Max != 9 || a.Last != 1 {
+		t.Fatalf("unexpected merge result %+v", a)
+	}
+}
+
+func TestConcurrentObserve(t *testing.T) {
+	setupDB(t)
+	f := NewCounter("test_concurrent_total", "test", "worker")
+	var wg sync.WaitGroup
+	for w := range 8 {
+		wg.Add(1)
+		go func(w int) {
+			defer wg.Done()
+			for range 1000 {
+				f.Inc(string(rune('a' + w%2)))
+			}
+		}(w)
+	}
+	wg.Wait()
+	var total float64
+	for s, c := range defaultAgg.snapshotCum() {
+		if s.Name == "test_concurrent_total" {
+			total += c.sum
+		}
+	}
+	if total != 8000 {
+		t.Fatalf("total = %v, want 8000", total)
+	}
+}
+
+func TestFlushQueryRollup(t *testing.T) {
+	setupDB(t)
+	h := NewHistogram("test_latency_ms", "test", []float64{1, 10, 100}, "route")
+	s := h.series([]string{"/a"})
+
+	// two hours ago, spread over three minutes, plus one point in the current minute
+	base := hourOf(time.Now().Unix()) - 2*3600
+	for i := range 30 {
+		defaultAgg.observe(s, base+int64(i%3)*60, float64(i))
+	}
+	now := time.Now().Unix()
+	defaultAgg.observe(s, now, 50)
+
+	if err := flush(minuteOf(now)); err != nil {
+		t.Fatal(err)
+	}
+	var rows int64
+	initializers.DB.Model(&models.MetricPoint{}).Where("resolution = ?", resMinute).Count(&rows)
+	if rows != 3 {
+		t.Fatalf("minute rows = %d, want 3 (current minute stays in memory)", rows)
+	}
+
+	// flushing the same bucket again must merge, not overwrite
+	defaultAgg.observe(s, base, 1000)
+	if err := flush(minuteOf(now)); err != nil {
+		t.Fatal(err)
+	}
+	var p models.MetricPoint
+	initializers.DB.Where("resolution = ? AND bucket = ?", resMinute, base).First(&p)
+	if p.Count != 11 || p.Max != 1000 {
+		t.Fatalf("merged point = %+v, want count 11 and max 1000", p)
+	}
+
+	// query sees DB rows plus the unflushed current minute
+	sel := Selector{Service: MainService, Name: "test_latency_ms"}
+	res, err := Query(sel, "count", now-3*3600, now, 60)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var count float64
+	for _, pt := range res.Series[0].Points {
+		if pt.V != nil {
+			count += *pt.V
+		}
+	}
+	if count != 32 {
+		t.Fatalf("queried count = %v, want 32", count)
+	}
+
+	if err := rollup(now); err != nil {
+		t.Fatal(err)
+	}
+	var hourly models.MetricPoint
+	if err := initializers.DB.Where("resolution = ? AND bucket = ?", resHour, base).First(&hourly).Error; err != nil {
+		t.Fatalf("missing hourly rollup: %v", err)
+	}
+	if hourly.Count != 31 || hourly.Max != 1000 {
+		t.Fatalf("hourly = %+v, want count 31 max 1000", hourly)
+	}
+	// rolling up again is idempotent
+	rolledUntil.Store(0)
+	if err := rollup(now); err != nil {
+		t.Fatal(err)
+	}
+	initializers.DB.Where("resolution = ? AND bucket = ?", resHour, base).First(&hourly)
+	if hourly.Count != 31 {
+		t.Fatalf("re-rollup changed count to %d", hourly.Count)
+	}
+
+	totals, err := Totals(sel, now-3*3600, now)
+	if err != nil || len(totals) != 1 || totals[0].P95 == nil || math.IsNaN(*totals[0].P95) {
+		t.Fatalf("totals = %+v, err %v", totals, err)
+	}
+}
+
+func TestIngestValidation(t *testing.T) {
+	setupDB(t)
+	now := time.Now()
+	res := Ingest("indexer", "i-1", []models.TelemetryMetric{
+		{Name: "ok_total", Type: "counter", Value: 2},
+		{Name: "Bad-Name", Type: "gauge", Value: 1},
+		{Name: "neg_total", Type: "counter", Value: -1},
+		{Name: "lat", Type: "summary", Count: 2, Sum: 3, Min: 1, Max: 2},
+		{Name: "ok_total", Type: "gauge", Value: 1},
+		{Name: "weird", Type: "nope"},
+	}, now)
+	if res.Accepted != 2 || res.Dropped != 4 {
+		t.Fatalf("accepted %d dropped %d (%v)", res.Accepted, res.Dropped, res.Reasons)
+	}
+	if res.Reasons["type_mismatch"] != 1 || res.Reasons["invalid_name"] != 1 {
+		t.Fatalf("unexpected reasons %v", res.Reasons)
+	}
+}
+
+func TestPrometheusOutput(t *testing.T) {
+	setupDB(t)
+	c := NewCounter("test_prom_total", "a \"quoted\" help", "kind")
+	c.Add(3, `va"l`)
+	h := NewHistogram("test_prom_ms", "hist", []float64{1, 10})
+	h.Observe(5)
+	var b strings.Builder
+	WritePrometheus(&b)
+	out := b.String()
+	for _, want := range []string{
+		`test_prom_total{kind="va\"l"} 3`,
+		`test_prom_ms_bucket{le="1"} 0`,
+		`test_prom_ms_bucket{le="10"} 1`,
+		`test_prom_ms_bucket{le="+Inf"} 1`,
+		`# TYPE test_prom_ms histogram`,
+	} {
+		if !strings.Contains(out, want) {
+			t.Errorf("missing %q in output", want)
+		}
+	}
+}

metrics/pipeline.go

@@ -0,0 +1,156 @@
+package metrics
+
+import (
+	"context"
+	"log"
+	"math"
+	"sync"
+	"time"
+
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
+)
+
+var startOnce sync.Once
+
+// Start launches the background jobs: minute flush and samplers, top-N
+// persistence, hourly rollup and retention, and service heartbeat checks.
+func Start(ctx context.Context) {
+	startOnce.Do(func() {
+		Top.seed()
+		go every(ctx, "flush", time.Minute, 5*time.Second, func(now time.Time) {
+			sample()
+			if err := flush(minuteOf(now.Unix())); err != nil {
+				log.Printf("metrics: flush failed: %v", err)
+			}
+		})
+		go every(ctx, "top", 5*time.Minute, 0, func(time.Time) {
+			if err := Top.flush(); err != nil {
+				log.Printf("metrics: top items flush failed: %v", err)
+			}
+		})
+		go every(ctx, "rollup", time.Hour, 2*time.Minute, func(now time.Time) {
+			RunMaintenance(now)
+		})
+		go every(ctx, "services", 30*time.Second, 0, func(now time.Time) {
+			selfHeartbeat(now)
+			checkServices(now)
+		})
+		// initialize the rollup watermark and service state right away
+		go func() {
+			selfHeartbeat(time.Now())
+			checkServices(time.Now())
+			if err := rollup(time.Now().Unix()); err != nil {
+				log.Printf("metrics: rollup failed: %v", err)
+			}
+		}()
+	})
+}
+
+// RunMaintenance runs the rollup and retention jobs.
+func RunMaintenance(now time.Time) {
+	if err := rollup(now.Unix()); err != nil {
+		log.Printf("metrics: rollup failed: %v", err)
+	}
+	if err := retention(now); err != nil {
+		log.Printf("metrics: retention failed: %v", err)
+	}
+}
+
+// every runs fn on a fixed period aligned to wall clock (+offset).
+func every(ctx context.Context, name string, period, offset time.Duration, fn func(time.Time)) {
+	for {
+		now := time.Now()
+		next := now.Truncate(period).Add(offset)
+		if !next.After(now) {
+			next = next.Add(period)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case t := <-time.After(time.Until(next)):
+			func() {
+				defer func() {
+					if r := recover(); r != nil {
+						log.Printf("metrics: %s job panicked: %v", name, r)
+					}
+				}()
+				fn(t)
+			}()
+		}
+	}
+}
+
+// FlushAll writes everything still in memory; used on shutdown.
+func FlushAll() {
+	sample()
+	if err := flush(math.MaxInt64); err != nil {
+		log.Printf("metrics: final flush failed: %v", err)
+	}
+	if err := Top.flush(); err != nil {
+		log.Printf("metrics: final top items flush failed: %v", err)
+	}
+}
+
+// QueueCounts is a cheap snapshot of the indexing queue.
+type QueueCounts struct {
+	Total    int64 `json:"total"`
+	Queued   int64 `json:"queued"`
+	Priority int64 `json:"priority"`
+	Indexing int64 `json:"indexing"`
+}
+
+// LoadQueueCounts counts addons by queue state with one query.
+func LoadQueueCounts() (QueueCounts, error) {
+	var qc QueueCounts
+	err := initializers.DB.Raw(`SELECT COUNT(*) AS total,
+		COALESCE(SUM(CASE WHEN to_be_indexed = ? AND is_being_indexed = ? THEN 1 ELSE 0 END), 0) AS queued,
+		COALESCE(SUM(CASE WHEN to_be_indexed = ? AND is_being_indexed = ? AND priority_indexing = ? THEN 1 ELSE 0 END), 0) AS priority,
+		COALESCE(SUM(CASE WHEN is_being_indexed = ? THEN 1 ELSE 0 END), 0) AS indexing
+		FROM addons WHERE deleted_at IS NULL`,
+		true, false, true, false, true, true).Scan(&qc).Error
+	return qc, err
+}
+
+// sample records gauges that are read from the database or memory.
+func sample() {
+	if qc, err := LoadQueueCounts(); err == nil {
+		AddonsTotal.Set(float64(qc.Total))
+		QueueDepth.Set(float64(qc.Queued), "total")
+		QueueDepth.Set(float64(qc.Priority), "priority")
+		BeingIndexed.Set(float64(qc.Indexing))
+	}
+	humans, bots := liveVisitors()
+	VisitorsUnique.Set(float64(humans), "human")
+	VisitorsUnique.Set(float64(bots), "bot")
+}
+
+// PipelineStatus describes the health of the metrics pipeline.
+type PipelineStatus struct {
+	Enabled       bool   `json:"enabled"`
+	LastFlush     int64  `json:"lastFlush"`
+	LastFlushMs   int64  `json:"lastFlushMs"`
+	LastFlushErr  string `json:"lastFlushError"`
+	LastRollup    int64  `json:"lastRollup"`
+	LastRetention int64  `json:"lastRetention"`
+	RolledUntil   int64  `json:"rolledUntil"`
+	OpenPoints    int    `json:"openPoints"`
+	LiveSeries    int    `json:"liveSeries"`
+	DroppedPoints int64  `json:"droppedPoints"`
+}
+
+func Status() PipelineStatus {
+	status.mu.Lock()
+	defer status.mu.Unlock()
+	return PipelineStatus{
+		Enabled:       initializers.MetricsEnabled,
+		LastFlush:     status.LastFlush,
+		LastFlushMs:   status.LastFlushMs,
+		LastFlushErr:  status.LastFlushErr,
+		LastRollup:    status.LastRollup,
+		LastRetention: status.LastRetention,
+		RolledUntil:   rolledUntil.Load(),
+		OpenPoints:    defaultAgg.openCount(),
+		LiveSeries:    len(reg.all()),
+		DroppedPoints: status.Dropped,
+	}
+}

metrics/prometheus.go

@@ -0,0 +1,181 @@
+package metrics
+
+import (
+	"bufio"
+	"io"
+	"runtime"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
+)
+
+// WritePrometheus writes all cumulative series in the Prometheus text format
+// (version 0.0.4). External metrics are exposed as <service>_<name> with an
+// "instance" label; counters are totals since this process started.
+func WritePrometheus(w io.Writer) {
+	bw := bufio.NewWriter(w)
+	defer bw.Flush()
+
+	cum := defaultAgg.snapshotCum()
+	type entry struct {
+		s *Series
+		c cumulative
+	}
+	byName := map[string][]entry{}
+	meta := map[string]*Series{}
+	for s, c := range cum {
+		name := s.Name
+		if s.Service != MainService {
+			name = s.Service + "_" + s.Name
+		}
+		byName[name] = append(byName[name], entry{s, c})
+		if meta[name] == nil {
+			meta[name] = s
+		}
+	}
+	names := make([]string, 0, len(byName))
+	for n := range byName {
+		names = append(names, n)
+	}
+	sort.Strings(names)
+
+	for _, name := range names {
+		s := meta[name]
+		writeHeader(bw, name, s.Help, promType(s.Kind))
+		entries := byName[name]
+		sort.Slice(entries, func(i, j int) bool {
+			if entries[i].s.Instance != entries[j].s.Instance {
+				return entries[i].s.Instance < entries[j].s.Instance
+			}
+			return entries[i].s.LabelsJSON < entries[j].s.LabelsJSON
+		})
+		for _, e := range entries {
+			labels := e.s.Labels
+			if e.s.Service != MainService {
+				labels = withLabel(labels, "instance", e.s.Instance)
+			}
+			switch e.s.Kind {
+			case Counter:
+				writeSample(bw, name, labels, e.c.sum)
+			case Gauge:
+				writeSample(bw, name, labels, e.c.last)
+			case Summary:
+				writeSample(bw, name+"_sum", labels, e.c.sum)
+				writeSample(bw, name+"_count", labels, float64(e.c.count))
+			case Histogram:
+				var acc uint64
+				for i, b := range e.s.Bounds {
+					if i < len(e.c.hist) {
+						acc += e.c.hist[i]
+					}
+					writeSample(bw, name+"_bucket", withLabel(labels, "le", formatFloat(b)), float64(acc))
+				}
+				writeSample(bw, name+"_bucket", withLabel(labels, "le", "+Inf"), float64(e.c.count))
+				writeSample(bw, name+"_sum", labels, e.c.sum)
+				writeSample(bw, name+"_count", labels, float64(e.c.count))
+			}
+		}
+	}
+
+	up := ServiceUp()
+	if len(up) > 0 {
+		writeHeader(bw, "reforger_service_up", "Whether a reporting service instance is sending heartbeats", "gauge")
+		keys := make([][2]string, 0, len(up))
+		for k := range up {
+			keys = append(keys, k)
+		}
+		sort.Slice(keys, func(i, j int) bool { return keys[i][0]+keys[i][1] < keys[j][0]+keys[j][1] })
+		for _, k := range keys {
+			v := 0.0
+			if up[k] {
+				v = 1
+			}
+			writeSample(bw, "reforger_service_up", map[string]string{"service": k[0], "instance": k[1]}, v)
+		}
+	}
+
+	var ms runtime.MemStats
+	runtime.ReadMemStats(&ms)
+	writeHeader(bw, "go_goroutines", "Number of goroutines", "gauge")
+	writeSample(bw, "go_goroutines", nil, float64(runtime.NumGoroutine()))
+	writeHeader(bw, "go_memstats_heap_alloc_bytes", "Heap bytes allocated and in use", "gauge")
+	writeSample(bw, "go_memstats_heap_alloc_bytes", nil, float64(ms.HeapAlloc))
+	writeHeader(bw, "go_memstats_sys_bytes", "Bytes obtained from the OS", "gauge")
+	writeSample(bw, "go_memstats_sys_bytes", nil, float64(ms.Sys))
+	writeHeader(bw, "go_gc_cycles_total", "Completed GC cycles", "counter")
+	writeSample(bw, "go_gc_cycles_total", nil, float64(ms.NumGC))
+	writeHeader(bw, "process_start_time_seconds", "Process start time", "gauge")
+	writeSample(bw, "process_start_time_seconds", nil, float64(startedAt.Unix()))
+	writeHeader(bw, "process_uptime_seconds", "Process uptime", "gauge")
+	writeSample(bw, "process_uptime_seconds", nil, time.Since(startedAt).Seconds())
+
+	if sqlDB, err := initializers.DB.DB(); err == nil {
+		st := sqlDB.Stats()
+		writeHeader(bw, "db_open_connections", "Open database connections", "gauge")
+		writeSample(bw, "db_open_connections", nil, float64(st.OpenConnections))
+		writeHeader(bw, "db_in_use_connections", "Database connections in use", "gauge")
+		writeSample(bw, "db_in_use_connections", nil, float64(st.InUse))
+		writeHeader(bw, "db_wait_count_total", "Waits for a database connection", "counter")
+		writeSample(bw, "db_wait_count_total", nil, float64(st.WaitCount))
+		writeHeader(bw, "db_wait_seconds_total", "Time spent waiting for a database connection", "counter")
+		writeSample(bw, "db_wait_seconds_total", nil, st.WaitDuration.Seconds())
+	}
+}
+
+func promType(k Kind) string {
+	switch k {
+	case Counter:
+		return "counter"
+	case Histogram:
+		return "histogram"
+	case Summary:
+		return "summary"
+	}
+	return "gauge"
+}
+
+func withLabel(labels map[string]string, k, v string) map[string]string {
+	out := make(map[string]string, len(labels)+1)
+	for lk, lv := range labels {
+		out[lk] = lv
+	}
+	out[k] = v
+	return out
+}
+
+func formatFloat(v float64) string { return strconv.FormatFloat(v, 'g', -1, 64) }
+
+var helpEscaper = strings.NewReplacer(`\`, `\\`, "\n", `\n`)
+var valueEscaper = strings.NewReplacer(`\`, `\\`, "\n", `\n`, `"`, `\"`)
+
+func writeHeader(w *bufio.Writer, name, help, typ string) {
+	if help != "" {
+		w.WriteString("# HELP " + name + " " + helpEscaper.Replace(help) + "\n")
+	}
+	w.WriteString("# TYPE " + name + " " + typ + "\n")
+}
+
+func writeSample(w *bufio.Writer, name string, labels map[string]string, v float64) {
+	w.WriteString(name)
+	if len(labels) > 0 {
+		keys := make([]string, 0, len(labels))
+		for k := range labels {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		w.WriteByte('{')
+		for i, k := range keys {
+			if i > 0 {
+				w.WriteByte(',')
+			}
+			w.WriteString(k + `="` + valueEscaper.Replace(labels[k]) + `"`)
+		}
+		w.WriteByte('}')
+	}
+	w.WriteByte(' ')
+	w.WriteString(formatFloat(v))
+	w.WriteByte('\n')
+}

metrics/query.go

@@ -0,0 +1,526 @@
+package metrics
+
+import (
+	"encoding/json"
+	"fmt"
+	"math"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/models"
+)
+
+const (
+	maxQuerySeries = 50
+	maxQueryPoints = 1500
+)
+
+// Selector picks series by service, name and exact label matches, and groups
+// them by the GroupBy labels ("service" and "instance" work as labels too).
+type Selector struct {
+	Service string
+	Name    string
+	Match   map[string]string
+	GroupBy []string
+}
+
+// Aggregations supported by Query and Totals.
+var Aggregations = []string{"sum", "count", "rate", "avg", "min", "max", "last", "p50", "p90", "p95", "p99"}
+
+func validAgg(agg string) bool {
+	for _, a := range Aggregations {
+		if a == agg {
+			return true
+		}
+	}
+	return false
+}
+
+// DefaultAgg picks a sensible aggregation for a metric kind.
+func DefaultAgg(kind Kind) string {
+	switch kind {
+	case Counter:
+		return "sum"
+	case Gauge:
+		return "last"
+	case Histogram:
+		return "p95"
+	default:
+		return "avg"
+	}
+}
+
+// Point is [unix seconds, value or null] in JSON.
+type Point struct {
+	T int64
+	V *float64
+}
+
+func (p Point) MarshalJSON() ([]byte, error) {
+	if p.V == nil {
+		return []byte("[" + strconv.FormatInt(p.T, 10) + ",null]"), nil
+	}
+	return []byte("[" + strconv.FormatInt(p.T, 10) + "," + strconv.FormatFloat(*p.V, 'g', 10, 64) + "]"), nil
+}
+
+type SeriesResult struct {
+	Labels map[string]string `json:"labels"`
+	Points []Point           `json:"points"`
+}
+
+type QueryResult struct {
+	Name       string         `json:"name"`
+	Kind       Kind           `json:"kind"`
+	Agg        string         `json:"agg"`
+	Resolution int64          `json:"resolution"`
+	Step       int64          `json:"step"`
+	From       int64          `json:"from"`
+	To         int64          `json:"to"`
+	Series     []SeriesResult `json:"series"`
+	Truncated  bool           `json:"truncated"`
+}
+
+// matched is a series selected for a query, possibly only known from the DB.
+type matched struct {
+	id     uint
+	live   *Series // nil when this process never saw the series
+	labels map[string]string
+	svc    string
+	inst   string
+	kind   Kind
+	bounds []float64
+}
+
+func (m *matched) label(name string) string {
+	switch name {
+	case "service":
+		if v, ok := m.labels["service"]; ok {
+			return v
+		}
+		return m.svc
+	case "instance":
+		if v, ok := m.labels["instance"]; ok {
+			return v
+		}
+		return m.inst
+	}
+	return m.labels[name]
+}
+
+func (sel Selector) matches(m *matched) bool {
+	if sel.Service != "" && m.svc != sel.Service {
+		return false
+	}
+	for k, v := range sel.Match {
+		if m.label(k) != v {
+			return false
+		}
+	}
+	return true
+}
+
+func findSeries(sel Selector) ([]*matched, error) {
+	byKey := map[seriesKey]*matched{}
+
+	q := initializers.DB.Model(&models.MetricSeries{}).Where("name = ?", sel.Name)
+	if sel.Service != "" {
+		q = q.Where("service = ?", sel.Service)
+	}
+	var rows []models.MetricSeries
+	if err := q.Find(&rows).Error; err != nil {
+		return nil, err
+	}
+	for _, r := range rows {
+		m := &matched{id: r.ID, svc: r.Service, inst: r.Instance, kind: Kind(r.Kind), labels: map[string]string{}}
+		_ = json.Unmarshal([]byte(r.Labels), &m.labels)
+		if r.Bounds != "" {
+			_ = json.Unmarshal([]byte(r.Bounds), &m.bounds)
+		}
+		byKey[seriesKey{r.Service, r.Instance, r.Name, r.Labels}] = m
+	}
+
+	for _, s := range reg.all() {
+		if s.Name != sel.Name || (sel.Service != "" && s.Service != sel.Service) {
+			continue
+		}
+		k := s.key()
+		if m := byKey[k]; m != nil {
+			m.live = s
+			continue
+		}
+		byKey[k] = &matched{live: s, svc: s.Service, inst: s.Instance, kind: s.Kind, bounds: s.Bounds, labels: s.Labels}
+	}
+
+	out := make([]*matched, 0, len(byKey))
+	for _, m := range byKey {
+		if sel.matches(m) {
+			out = append(out, m)
+		}
+	}
+	return out, nil
+}
+
+type groupAcc struct {
+	labels  map[string]string
+	steps   map[int64]*Agg
+	lastSum map[int64]float64
+}
+
+// collect loads all matching data in [from, to] and folds it into groups and
+// step buckets chosen by bucketFn. Gauge "last" values are summed across the
+// series of a group (e.g. queue depth over all instances).
+func collect(sel Selector, from, to, res int64, bucketFn func(int64) int64) (map[string]*groupAcc, Kind, []float64, error) {
+	series, err := findSeries(sel)
+	if err != nil {
+		return nil, "", nil, err
+	}
+	if len(series) == 0 {
+		return map[string]*groupAcc{}, "", nil, nil
+	}
+	kind, bounds := series[0].kind, series[0].bounds
+
+	perSeries := map[*matched]map[int64]*Agg{}
+	add := func(m *matched, ts int64, a *Agg) {
+		b := bucketFn(ts)
+		steps := perSeries[m]
+		if steps == nil {
+			steps = map[int64]*Agg{}
+			perSeries[m] = steps
+		}
+		if cur := steps[b]; cur != nil {
+			cur.merge(a)
+		} else {
+			steps[b] = a.clone()
+		}
+	}
+
+	byID := map[uint]*matched{}
+	var ids []uint
+	for _, m := range series {
+		if m.id != 0 {
+			byID[m.id] = m
+			ids = append(ids, m.id)
+		}
+	}
+
+	load := func(resolution int, lo, hi int64) error {
+		if lo > hi || len(ids) == 0 {
+			return nil
+		}
+		for start := 0; start < len(ids); start += 300 {
+			end := min(start+300, len(ids))
+			var rows []models.MetricPoint
+			err := initializers.DB.
+				Where("resolution = ? AND series_id IN ? AND bucket >= ? AND bucket <= ?", resolution, ids[start:end], lo, hi).
+				Find(&rows).Error
+			if err != nil {
+				return err
+			}
+			for i := range rows {
+				add(byID[rows[i].SeriesID], rows[i].Bucket, pointToAgg(&rows[i]))
+			}
+		}
+		return nil
+	}
+
+	if res == resHour {
+		cut := rolledUntil.Load()
+		if err := load(resHour, hourOf(from), cut-1); err != nil {
+			return nil, "", nil, err
+		}
+		if err := load(resMinute, max(cut, from), to); err != nil {
+			return nil, "", nil, err
+		}
+	} else if err := load(resMinute, minuteOf(from), to); err != nil {
+		return nil, "", nil, err
+	}
+
+	// not yet flushed buckets
+	live := map[*Series]*matched{}
+	want := map[*Series]bool{}
+	for _, m := range series {
+		if m.live != nil {
+			live[m.live] = m
+			want[m.live] = true
+		}
+	}
+	for k, a := range defaultAgg.snapshotOpen(want) {
+		if k.bucket >= minuteOf(from) && k.bucket <= to {
+			add(live[k.s], k.bucket, a)
+		}
+	}
+
+	groups := map[string]*groupAcc{}
+	for m, steps := range perSeries {
+		labels := make(map[string]string, len(sel.GroupBy))
+		parts := make([]string, len(sel.GroupBy))
+		for i, g := range sel.GroupBy {
+			labels[g] = m.label(g)
+			parts[i] = labels[g]
+		}
+		gk := strings.Join(parts, "\x00")
+		g := groups[gk]
+		if g == nil {
+			g = &groupAcc{labels: labels, steps: map[int64]*Agg{}, lastSum: map[int64]float64{}}
+			groups[gk] = g
+		}
+		for b, a := range steps {
+			if cur := g.steps[b]; cur != nil {
+				cur.merge(a)
+			} else {
+				g.steps[b] = a.clone()
+			}
+			g.lastSum[b] += a.Last
+		}
+	}
+	return groups, kind, bounds, nil
+}
+
+func evaluate(a *Agg, lastSum float64, agg string, kind Kind, bounds []float64, span int64) (float64, bool) {
+	if a == nil || a.Count == 0 {
+		return 0, false
+	}
+	switch agg {
+	case "count":
+		return float64(a.Count), true
+	case "sum":
+		return a.Sum, true
+	case "rate":
+		if span <= 0 {
+			return 0, false
+		}
+		if kind == Counter {
+			return a.Sum / float64(span), true
+		}
+		return float64(a.Count) / float64(span), true
+	case "avg":
+		return a.Sum / float64(a.Count), true
+	case "min":
+		return a.Min, true
+	case "max":
+		return a.Max, true
+	case "last":
+		return lastSum, true
+	case "p50", "p90", "p95", "p99":
+		q, _ := strconv.ParseFloat(agg[1:], 64)
+		return quantile(a, bounds, q/100)
+	}
+	return 0, false
+}
+
+func pickResolution(from, to, step int64) int64 {
+	minuteHorizon := time.Now().Unix() - int64(initializers.MetricsMinuteRetentionDays)*86400
+	if to-from > 48*3600 || step >= resHour || from < minuteHorizon {
+		return resHour
+	}
+	return resMinute
+}
+
+// Query returns one time series per group with values at every step.
+func Query(sel Selector, agg string, from, to, step int64) (*QueryResult, error) {
+	if sel.Name == "" {
+		return nil, fmt.Errorf("name is required")
+	}
+	if to <= from {
+		return nil, fmt.Errorf("to must be after from")
+	}
+	res := pickResolution(from, to, step)
+	if step < res {
+		step = res
+	}
+	if (to-from)/step > maxQueryPoints {
+		step = (to - from) / maxQueryPoints
+	}
+	step = int64(math.Ceil(float64(step)/float64(res))) * res
+	align := func(ts int64) int64 { return ts - ((ts%step)+step)%step }
+
+	groups, kind, bounds, err := collect(sel, from, to, res, align)
+	if err != nil {
+		return nil, err
+	}
+	if agg == "" {
+		agg = DefaultAgg(kind)
+	}
+	if !validAgg(agg) {
+		return nil, fmt.Errorf("unknown aggregation %q", agg)
+	}
+
+	result := &QueryResult{Name: sel.Name, Kind: kind, Agg: agg, Resolution: res, Step: step, From: from, To: to, Series: []SeriesResult{}}
+	keys := make([]string, 0, len(groups))
+	for k := range groups {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	if len(keys) > maxQuerySeries {
+		keys = keys[:maxQuerySeries]
+		result.Truncated = true
+	}
+	for _, k := range keys {
+		g := groups[k]
+		sr := SeriesResult{Labels: g.labels}
+		for b := align(from); b <= to; b += step {
+			p := Point{T: b}
+			if v, ok := evaluate(g.steps[b], g.lastSum[b], agg, kind, bounds, step); ok && !math.IsNaN(v) && !math.IsInf(v, 0) {
+				p.V = &v
+			}
+			sr.Points = append(sr.Points, p)
+		}
+		result.Series = append(result.Series, sr)
+	}
+	return result, nil
+}
+
+// Total is the whole-range aggregate of one group.
+type Total struct {
+	Labels map[string]string `json:"labels"`
+	Count  int64             `json:"count"`
+	Sum    float64           `json:"sum"`
+	Min    *float64          `json:"min"`
+	Max    *float64          `json:"max"`
+	Avg    *float64          `json:"avg"`
+	Last   *float64          `json:"last"`
+	Rate   *float64          `json:"rate"`
+	P50    *float64          `json:"p50"`
+	P95    *float64          `json:"p95"`
+	P99    *float64          `json:"p99"`
+}
+
+// Totals aggregates [from, to] into one value set per group.
+func Totals(sel Selector, from, to int64) ([]Total, error) {
+	res := pickResolution(from, to, 0)
+	groups, kind, bounds, err := collect(sel, from, to, res, func(int64) int64 { return 0 })
+	if err != nil {
+		return nil, err
+	}
+	span := to - from
+	out := make([]Total, 0, len(groups))
+	for _, g := range groups {
+		a := g.steps[0]
+		if a == nil {
+			continue
+		}
+		t := Total{Labels: g.labels, Count: a.Count, Sum: a.Sum}
+		val := func(agg string) *float64 {
+			if v, ok := evaluate(a, g.lastSum[0], agg, kind, bounds, span); ok && !math.IsNaN(v) && !math.IsInf(v, 0) {
+				return &v
+			}
+			return nil
+		}
+		t.Min, t.Max, t.Avg, t.Last, t.Rate = val("min"), val("max"), val("avg"), val("last"), val("rate")
+		t.P50, t.P95, t.P99 = val("p50"), val("p95"), val("p99")
+		out = append(out, t)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Count > out[j].Count })
+	return out, nil
+}
+
+// Sum returns the summed value of a counter (or count of observations for
+// other kinds) over the range, across all matching series.
+func Sum(sel Selector, from, to int64) float64 {
+	sel.GroupBy = nil
+	totals, err := Totals(sel, from, to)
+	if err != nil || len(totals) == 0 {
+		return 0
+	}
+	if f := familyByName(sel.Name); f != nil && f.Kind != Counter {
+		return float64(totals[0].Count)
+	}
+	return totals[0].Sum
+}
+
+// CatalogEntry describes one metric name of one service.
+type CatalogEntry struct {
+	Service   string              `json:"service"`
+	Name      string              `json:"name"`
+	Kind      Kind                `json:"kind"`
+	Help      string              `json:"help"`
+	Labels    map[string][]string `json:"labels"`
+	Instances []string            `json:"instances"`
+	Series    int                 `json:"series"`
+	LastSeen  int64               `json:"lastSeen"`
+}
+
+// Catalog lists every metric known in the DB or in memory.
+func Catalog(service string) ([]CatalogEntry, error) {
+	type acc struct {
+		e      CatalogEntry
+		labels map[string]map[string]bool
+		inst   map[string]bool
+		keys   map[seriesKey]bool
+	}
+	entries := map[string]*acc{}
+	addSeries := func(svc, inst, name string, kind Kind, help, labelsJSON string, labels map[string]string, lastSeen int64) {
+		k := svc + "/" + name
+		a := entries[k]
+		if a == nil {
+			a = &acc{e: CatalogEntry{Service: svc, Name: name, Kind: kind, Help: help}, labels: map[string]map[string]bool{}, inst: map[string]bool{}, keys: map[seriesKey]bool{}}
+			entries[k] = a
+		}
+		if a.e.Help == "" {
+			a.e.Help = help
+		}
+		sk := seriesKey{svc, inst, name, labelsJSON}
+		if !a.keys[sk] {
+			a.keys[sk] = true
+			a.e.Series++
+		}
+		a.inst[inst] = true
+		for lk, lv := range labels {
+			if a.labels[lk] == nil {
+				a.labels[lk] = map[string]bool{}
+			}
+			a.labels[lk][lv] = true
+		}
+		if lastSeen > a.e.LastSeen {
+			a.e.LastSeen = lastSeen
+		}
+	}
+
+	q := initializers.DB.Model(&models.MetricSeries{})
+	if service != "" {
+		q = q.Where("service = ?", service)
+	}
+	var rows []models.MetricSeries
+	if err := q.Find(&rows).Error; err != nil {
+		return nil, err
+	}
+	for _, r := range rows {
+		labels := map[string]string{}
+		_ = json.Unmarshal([]byte(r.Labels), &labels)
+		addSeries(r.Service, r.Instance, r.Name, Kind(r.Kind), r.Help, r.Labels, labels, r.LastSeen)
+	}
+	for _, s := range reg.all() {
+		if service != "" && s.Service != service {
+			continue
+		}
+		addSeries(s.Service, s.Instance, s.Name, s.Kind, s.Help, s.LabelsJSON, s.Labels, reg.seen(s))
+	}
+
+	out := make([]CatalogEntry, 0, len(entries))
+	for _, a := range entries {
+		a.e.Labels = map[string][]string{}
+		for lk, vals := range a.labels {
+			list := make([]string, 0, len(vals))
+			for v := range vals {
+				list = append(list, v)
+			}
+			sort.Strings(list)
+			a.e.Labels[lk] = list
+		}
+		for inst := range a.inst {
+			a.e.Instances = append(a.e.Instances, inst)
+		}
+		sort.Strings(a.e.Instances)
+		out = append(out, a.e)
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Service != out[j].Service {
+			return out[i].Service < out[j].Service
+		}
+		return out[i].Name < out[j].Name
+	})
+	return out, nil
+}

metrics/registry.go

@@ -0,0 +1,284 @@
+package metrics
+
+import (
+	"encoding/json"
+	"errors"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+)
+
+type Kind string
+
+const (
+	Counter   Kind = "counter"   // value is an increment; Sum holds the total
+	Gauge     Kind = "gauge"     // value is a sample; Last holds the latest
+	Summary   Kind = "summary"   // pre-aggregated count/sum/min/max from a client
+	Histogram Kind = "histogram" // observations bucketed by Bounds
+)
+
+// MainService is the service name the API uses for its own metrics.
+const MainService = "main"
+
+var (
+	errSeriesLimit  = errors.New("series limit reached")
+	errKindMismatch = errors.New("metric reported with a different type")
+)
+
+// Limits for series created from external telemetry.
+const (
+	maxSeriesPerName = 200
+	maxExternal      = 5000
+)
+
+var hostname = func() string {
+	h, err := os.Hostname()
+	if err != nil || h == "" {
+		return "api"
+	}
+	return h
+}()
+
+// Instance returns the instance name used for this process' own metrics.
+func Instance() string { return hostname }
+
+// Series is one metric name + label set from one service instance. Series are
+// interned, so pointer equality means same series.
+type Series struct {
+	Service    string
+	Instance   string
+	Name       string
+	Labels     map[string]string
+	LabelsJSON string
+	Kind       Kind
+	Bounds     []float64
+	Help       string
+}
+
+type seriesKey struct{ service, instance, name, labels string }
+
+func (s *Series) key() seriesKey {
+	return seriesKey{s.Service, s.Instance, s.Name, s.LabelsJSON}
+}
+
+// Label returns a label value, including the service/instance pseudo labels.
+func (s *Series) Label(name string) string {
+	switch name {
+	case "service":
+		if v, ok := s.Labels["service"]; ok {
+			return v
+		}
+		return s.Service
+	case "instance":
+		if v, ok := s.Labels["instance"]; ok {
+			return v
+		}
+		return s.Instance
+	}
+	return s.Labels[name]
+}
+
+// canonicalLabels serializes labels deterministically (encoding/json sorts map keys).
+func canonicalLabels(labels map[string]string) string {
+	if len(labels) == 0 {
+		return "{}"
+	}
+	b, _ := json.Marshal(labels)
+	return string(b)
+}
+
+type registry struct {
+	mu       sync.RWMutex
+	series   map[seriesKey]*Series
+	perName  map[string]int // external series per service/name
+	external int
+	lastSeen map[*Series]int64
+}
+
+var reg = &registry{
+	series:   make(map[seriesKey]*Series),
+	perName:  make(map[string]int),
+	lastSeen: make(map[*Series]int64),
+}
+
+func (r *registry) get(service, instance, name string, labels map[string]string, kind Kind, bounds []float64, help string, external bool) (*Series, error) {
+	lj := canonicalLabels(labels)
+	k := seriesKey{service, instance, name, lj}
+
+	r.mu.RLock()
+	s := r.series[k]
+	r.mu.RUnlock()
+	if s != nil {
+		if s.Kind != kind {
+			return nil, errKindMismatch
+		}
+		return s, nil
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if s = r.series[k]; s != nil {
+		return s, nil
+	}
+	nameKey := service + "/" + name
+	if external {
+		if r.external >= maxExternal || r.perName[nameKey] >= maxSeriesPerName {
+			return nil, errSeriesLimit
+		}
+		r.external++
+		r.perName[nameKey]++
+	}
+	copied := make(map[string]string, len(labels))
+	for k, v := range labels {
+		copied[k] = v
+	}
+	s = &Series{
+		Service:    service,
+		Instance:   instance,
+		Name:       name,
+		Labels:     copied,
+		LabelsJSON: lj,
+		Kind:       kind,
+		Bounds:     bounds,
+		Help:       help,
+	}
+	r.series[k] = s
+	return s, nil
+}
+
+func (r *registry) lookup(k seriesKey) *Series {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.series[k]
+}
+
+// all returns every known series, sorted by name then labels.
+func (r *registry) all() []*Series {
+	r.mu.RLock()
+	out := make([]*Series, 0, len(r.series))
+	for _, s := range r.series {
+		out = append(out, s)
+	}
+	r.mu.RUnlock()
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Name != out[j].Name {
+			return out[i].Name < out[j].Name
+		}
+		if out[i].Service != out[j].Service {
+			return out[i].Service < out[j].Service
+		}
+		if out[i].Instance != out[j].Instance {
+			return out[i].Instance < out[j].Instance
+		}
+		return out[i].LabelsJSON < out[j].LabelsJSON
+	})
+	return out
+}
+
+func (r *registry) touch(s *Series, ts int64) {
+	r.mu.Lock()
+	if ts > r.lastSeen[s] {
+		r.lastSeen[s] = ts
+	}
+	r.mu.Unlock()
+}
+
+func (r *registry) seen(s *Series) int64 {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.lastSeen[s]
+}
+
+// Family is a metric emitted by this process, with a fixed set of label names.
+type Family struct {
+	Name       string
+	Help       string
+	Kind       Kind
+	Bounds     []float64
+	LabelNames []string
+
+	mu    sync.Mutex
+	cache map[string]*Series
+}
+
+var (
+	familiesMu sync.Mutex
+	families   = map[string]*Family{}
+)
+
+func newFamily(name, help string, kind Kind, bounds []float64, labels []string) *Family {
+	f := &Family{Name: name, Help: help, Kind: kind, Bounds: bounds, LabelNames: labels, cache: map[string]*Series{}}
+	familiesMu.Lock()
+	families[name] = f
+	familiesMu.Unlock()
+	return f
+}
+
+func familyByName(name string) *Family {
+	familiesMu.Lock()
+	defer familiesMu.Unlock()
+	return families[name]
+}
+
+func NewCounter(name, help string, labels ...string) *Family {
+	return newFamily(name, help, Counter, nil, labels)
+}
+
+func NewGauge(name, help string, labels ...string) *Family {
+	return newFamily(name, help, Gauge, nil, labels)
+}
+
+func NewHistogram(name, help string, bounds []float64, labels ...string) *Family {
+	return newFamily(name, help, Histogram, bounds, labels)
+}
+
+func (f *Family) series(values []string) *Series {
+	key := strings.Join(values, "\x00")
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if s := f.cache[key]; s != nil {
+		return s
+	}
+	labels := make(map[string]string, len(f.LabelNames))
+	for i, n := range f.LabelNames {
+		if i < len(values) {
+			labels[n] = values[i]
+		} else {
+			labels[n] = ""
+		}
+	}
+	s, err := reg.get(MainService, Instance(), f.Name, labels, f.Kind, f.Bounds, f.Help, false)
+	if err != nil {
+		return nil
+	}
+	f.cache[key] = s
+	return s
+}
+
+// Add increments a counter. Zero increments are ignored.
+func (f *Family) Add(v float64, labelValues ...string) {
+	if v == 0 {
+		return
+	}
+	if s := f.series(labelValues); s != nil {
+		defaultAgg.observe(s, time.Now().Unix(), v)
+	}
+}
+
+func (f *Family) Inc(labelValues ...string) { f.Add(1, labelValues...) }
+
+// Set records a gauge sample.
+func (f *Family) Set(v float64, labelValues ...string) {
+	if s := f.series(labelValues); s != nil {
+		defaultAgg.observe(s, time.Now().Unix(), v)
+	}
+}
+
+// Observe records a histogram observation.
+func (f *Family) Observe(v float64, labelValues ...string) {
+	if s := f.series(labelValues); s != nil {
+		defaultAgg.observe(s, time.Now().Unix(), v)
+	}
+}

metrics/services.go

@@ -0,0 +1,325 @@
+package metrics
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math"
+	"regexp"
+	"sync"
+	"time"
+
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/models"
+	"gorm.io/gorm/clause"
+)
+
+var nameRe = regexp.MustCompile(`^[a-z_][a-z0-9_]{0,63}$`)
+
+// ValidName reports whether a service, metric or label name is acceptable.
+func ValidName(s string) bool { return nameRe.MatchString(s) }
+
+const (
+	defaultHeartbeat = 30
+	maxLabels        = 8
+	maxLabelValue    = 128
+)
+
+// normalizeTs converts unix seconds or milliseconds to seconds and clamps it
+// to [now-1h, now+1m]; anything outside becomes now.
+func normalizeTs(ts int64, now int64) int64 {
+	if ts <= 0 {
+		return now
+	}
+	if ts > 1e12 {
+		ts /= 1000
+	}
+	if ts < now-3600 || ts > now+60 {
+		return now
+	}
+	return ts
+}
+
+// IngestResult reports what happened to a telemetry batch.
+type IngestResult struct {
+	Accepted int            `json:"accepted"`
+	Dropped  int            `json:"dropped"`
+	Reasons  map[string]int `json:"reasons,omitempty"`
+}
+
+// Ingest records external metric samples for a service instance.
+func Ingest(service, instance string, samples []models.TelemetryMetric, now time.Time) IngestResult {
+	res := IngestResult{Reasons: map[string]int{}}
+	drop := func(reason string) {
+		res.Dropped++
+		res.Reasons[reason]++
+	}
+	nowTs := now.Unix()
+
+	for _, m := range samples {
+		if !ValidName(m.Name) {
+			drop("invalid_name")
+			continue
+		}
+		if len(m.Labels) > maxLabels {
+			drop("too_many_labels")
+			continue
+		}
+		badLabel := false
+		for k, v := range m.Labels {
+			if !ValidName(k) || len(v) > maxLabelValue {
+				badLabel = true
+				break
+			}
+		}
+		if badLabel {
+			drop("invalid_label")
+			continue
+		}
+		var kind Kind
+		switch m.Type {
+		case "counter":
+			kind = Counter
+		case "gauge", "":
+			kind = Gauge
+		case "summary":
+			kind = Summary
+		default:
+			drop("invalid_type")
+			continue
+		}
+		if math.IsNaN(m.Value) || math.IsInf(m.Value, 0) || math.IsNaN(m.Sum) || math.IsInf(m.Sum, 0) {
+			drop("invalid_value")
+			continue
+		}
+		if kind == Counter && m.Value < 0 {
+			drop("negative_counter")
+			continue
+		}
+		if kind == Summary && m.Count <= 0 {
+			drop("empty_summary")
+			continue
+		}
+		help := m.Help
+		if len(help) > 256 {
+			help = help[:256]
+		}
+		s, err := reg.get(service, instance, m.Name, m.Labels, kind, nil, help, true)
+		if err != nil {
+			if errors.Is(err, errKindMismatch) {
+				drop("type_mismatch")
+			} else {
+				drop("series_limit")
+			}
+			continue
+		}
+		ts := normalizeTs(m.Ts, nowTs)
+		if kind == Summary {
+			lo, hi := m.Min, m.Max
+			if lo == 0 && hi == 0 {
+				avg := m.Sum / float64(m.Count)
+				lo, hi = avg, avg
+			}
+			defaultAgg.observeSummary(s, ts, m.Count, m.Sum, lo, hi)
+		} else {
+			defaultAgg.observe(s, ts, m.Value)
+		}
+		res.Accepted++
+	}
+	TelemetrySamples.Add(float64(res.Accepted), "accepted")
+	TelemetrySamples.Add(float64(res.Dropped), "dropped")
+	return res
+}
+
+func jsonString(v map[string]any) string {
+	if len(v) == 0 {
+		return ""
+	}
+	b, err := json.Marshal(v)
+	if err != nil {
+		return ""
+	}
+	return string(b)
+}
+
+func truncate(s string, n int) string {
+	if len(s) > n {
+		return s[:n]
+	}
+	return s
+}
+
+// Heartbeat upserts the service instance row.
+func Heartbeat(b *models.TelemetryBatch, remoteAddr string, now time.Time) error {
+	interval := b.HeartbeatIntervalSec
+	if interval <= 0 {
+		interval = defaultHeartbeat
+	}
+	status := b.Status
+	if status == "" {
+		status = "ok"
+	}
+	row := models.ServiceInstance{
+		Service:              b.Service,
+		Instance:             b.Instance,
+		Version:              truncate(b.Version, 64),
+		Status:               truncate(status, 16),
+		StatusMessage:        truncate(b.StatusMessage, 512),
+		StartedAt:            normalizeStarted(b.StartedAt),
+		LastSeen:             now.Unix(),
+		HeartbeatIntervalSec: interval,
+		RemoteAddr:           truncate(remoteAddr, 64),
+		Meta:                 jsonString(b.Meta),
+	}
+	return initializers.DB.Clauses(clause.OnConflict{
+		Columns: []clause.Column{{Name: "service"}, {Name: "instance"}},
+		DoUpdates: clause.AssignmentColumns([]string{
+			"version", "status", "status_message", "started_at", "last_seen",
+			"heartbeat_interval_sec", "remote_addr", "meta", "updated_at",
+		}),
+	}).Create(&row).Error
+}
+
+func normalizeStarted(ts int64) int64 {
+	if ts > 1e12 {
+		return ts / 1000
+	}
+	return ts
+}
+
+// RecordEvents stores service events.
+func RecordEvents(service, instance string, events []models.TelemetryEvent, now time.Time) error {
+	if len(events) == 0 {
+		return nil
+	}
+	rows := make([]models.ServiceEvent, 0, len(events))
+	for _, e := range events {
+		level := e.Level
+		switch level {
+		case "debug", "info", "warn", "error":
+		case "warning":
+			level = "warn"
+		default:
+			level = "info"
+		}
+		ts := e.Ts
+		if ts <= 0 {
+			ts = now.UnixMilli()
+		} else if ts < 1e12 {
+			ts *= 1000
+		}
+		rows = append(rows, models.ServiceEvent{
+			Service: service, Instance: instance, Level: level,
+			Message: truncate(e.Message, 1024), Fields: jsonString(e.Fields), TS: ts,
+		})
+	}
+	return initializers.DB.CreateInBatches(&rows, 100).Error
+}
+
+// Event records a single event from the API itself.
+func Event(level, message string, fields map[string]any) {
+	err := RecordEvents(MainService, Instance(), []models.TelemetryEvent{{Level: level, Message: message, Fields: fields}}, time.Now())
+	if err != nil {
+		fmt.Printf("metrics: failed to record event: %v\n", err)
+	}
+}
+
+// Online reports whether an instance has sent a heartbeat recently enough.
+func Online(inst *models.ServiceInstance, now time.Time) bool {
+	interval := inst.HeartbeatIntervalSec
+	if interval <= 0 {
+		interval = defaultHeartbeat
+	}
+	return now.Unix()-inst.LastSeen < int64(3*interval)
+}
+
+var (
+	upMu    sync.Mutex
+	upState = map[string]bool{} // "service/instance" -> online
+)
+
+// ServiceUp returns the last known online state of every instance.
+func ServiceUp() map[[2]string]bool {
+	upMu.Lock()
+	defer upMu.Unlock()
+	out := make(map[[2]string]bool, len(upState))
+	for k, v := range upState {
+		var svc, inst string
+		for i := 0; i < len(k); i++ {
+			if k[i] == '/' {
+				svc, inst = k[:i], k[i+1:]
+				break
+			}
+		}
+		out[[2]string{svc, inst}] = v
+	}
+	return out
+}
+
+// checkServices records online/offline transitions as events.
+func checkServices(now time.Time) {
+	var rows []models.ServiceInstance
+	if err := initializers.DB.Find(&rows).Error; err != nil {
+		return
+	}
+	upMu.Lock()
+	known := make(map[string]bool, len(rows))
+	type change struct {
+		inst   models.ServiceInstance
+		online bool
+	}
+	var changes []change
+	for _, r := range rows {
+		k := r.Service + "/" + r.Instance
+		known[k] = true
+		online := Online(&r, now)
+		prev, seen := upState[k]
+		upState[k] = online
+		if seen && prev != online {
+			changes = append(changes, change{r, online})
+		}
+	}
+	for k := range upState {
+		if !known[k] {
+			delete(upState, k)
+		}
+	}
+	upMu.Unlock()
+
+	for _, ch := range changes {
+		level, msg := "info", "back online"
+		if !ch.online {
+			level, msg = "warn", "heartbeat lost"
+		}
+		_ = RecordEvents(ch.inst.Service, ch.inst.Instance, []models.TelemetryEvent{{
+			Level: level, Message: msg,
+			Fields: map[string]any{"kind": "status", "lastSeen": ch.inst.LastSeen},
+		}}, now)
+	}
+}
+
+var (
+	buildMu      sync.Mutex
+	buildVersion = "dev"
+	startedAt    = time.Now()
+)
+
+// SetVersion sets the version reported in the API's own heartbeat.
+func SetVersion(v string) {
+	buildMu.Lock()
+	buildVersion = v
+	buildMu.Unlock()
+}
+
+// StartedAt is when this process started.
+func StartedAt() time.Time { return startedAt }
+
+func selfHeartbeat(now time.Time) {
+	buildMu.Lock()
+	v := buildVersion
+	buildMu.Unlock()
+	_ = Heartbeat(&models.TelemetryBatch{
+		Service: MainService, Instance: Instance(), Version: v,
+		StartedAt: startedAt.Unix(), HeartbeatIntervalSec: 30, Status: "ok",
+	}, "", now)
+}

metrics/store.go

@@ -0,0 +1,371 @@
+package metrics
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/models"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+const (
+	resMinute = 60
+	resHour   = 3600
+	// points that cannot be written for this long are dropped
+	maxRetryAge = 15 * 60
+)
+
+var (
+	idMu     sync.Mutex
+	seriesID = map[*Series]uint{}
+
+	// rolledUntil is the start of the first hour without an hourly rollup.
+	rolledUntil atomic.Int64
+	// flushMu serializes writers of metric_points (flush, rollup, retention).
+	flushMu sync.Mutex
+)
+
+func pointToAgg(p *models.MetricPoint) *Agg {
+	a := &Agg{Count: p.Count, Sum: p.Sum, Min: p.Min, Max: p.Max, Last: p.Last, LastTs: p.Bucket}
+	if p.Hist != "" {
+		_ = json.Unmarshal([]byte(p.Hist), &a.Hist)
+	}
+	return a
+}
+
+func aggToPoint(id uint, res int, bucket int64, a *Agg) models.MetricPoint {
+	return models.MetricPoint{
+		SeriesID: id, Resolution: res, Bucket: bucket,
+		Count: a.Count, Sum: a.Sum, Min: a.Min, Max: a.Max, Last: a.Last, Hist: a.histJSON(),
+	}
+}
+
+// resolveIDs makes sure every series has a metric_series row.
+func resolveIDs(list []*Series) (map[*Series]uint, error) {
+	out := make(map[*Series]uint, len(list))
+	var missing []*Series
+	idMu.Lock()
+	for _, s := range list {
+		if id, ok := seriesID[s]; ok {
+			out[s] = id
+		} else {
+			missing = append(missing, s)
+		}
+	}
+	idMu.Unlock()
+
+	for _, s := range missing {
+		bounds := ""
+		if len(s.Bounds) > 0 {
+			b, _ := json.Marshal(s.Bounds)
+			bounds = string(b)
+		}
+		row := models.MetricSeries{
+			Service: s.Service, Instance: s.Instance, Name: s.Name, Labels: s.LabelsJSON,
+			Kind: string(s.Kind), Bounds: bounds, Help: s.Help, LastSeen: time.Now().Unix(),
+		}
+		if err := initializers.DB.Clauses(clause.OnConflict{DoNothing: true}).Create(&row).Error; err != nil {
+			return nil, err
+		}
+		if row.ID == 0 {
+			if err := initializers.DB.
+				Where("service = ? AND instance = ? AND name = ? AND labels = ?", s.Service, s.Instance, s.Name, s.LabelsJSON).
+				First(&row).Error; err != nil {
+				return nil, err
+			}
+		}
+		idMu.Lock()
+		seriesID[s] = row.ID
+		idMu.Unlock()
+		out[s] = row.ID
+	}
+	return out, nil
+}
+
+type pointID struct {
+	id     uint
+	bucket int64
+}
+
+// persist merges the given buckets into existing rows and upserts them. The
+// read-merge-write runs in one transaction; this process is the only writer.
+func persist(tx *gorm.DB, res int, points map[pointID]*Agg) error {
+	if len(points) == 0 {
+		return nil
+	}
+	keys := make([]pointID, 0, len(points))
+	for k := range points {
+		keys = append(keys, k)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		if keys[i].id != keys[j].id {
+			return keys[i].id < keys[j].id
+		}
+		return keys[i].bucket < keys[j].bucket
+	})
+
+	for start := 0; start < len(keys); start += 400 {
+		end := min(start+400, len(keys))
+		chunk := keys[start:end]
+		idSet, bucketSet := map[uint]bool{}, map[int64]bool{}
+		for _, k := range chunk {
+			idSet[k.id] = true
+			bucketSet[k.bucket] = true
+		}
+		ids := make([]uint, 0, len(idSet))
+		for id := range idSet {
+			ids = append(ids, id)
+		}
+		buckets := make([]int64, 0, len(bucketSet))
+		for b := range bucketSet {
+			buckets = append(buckets, b)
+		}
+
+		var existing []models.MetricPoint
+		if err := tx.Where("resolution = ? AND series_id IN ? AND bucket IN ?", res, ids, buckets).Find(&existing).Error; err != nil {
+			return err
+		}
+		prev := make(map[pointID]*Agg, len(existing))
+		for i := range existing {
+			prev[pointID{existing[i].SeriesID, existing[i].Bucket}] = pointToAgg(&existing[i])
+		}
+
+		rows := make([]models.MetricPoint, 0, len(chunk))
+		for _, k := range chunk {
+			a := points[k]
+			if old := prev[k]; old != nil {
+				old.merge(a)
+				a = old
+			}
+			rows = append(rows, aggToPoint(k.id, res, k.bucket, a))
+		}
+		err := tx.Clauses(clause.OnConflict{
+			Columns:   []clause.Column{{Name: "series_id"}, {Name: "resolution"}, {Name: "bucket"}},
+			DoUpdates: clause.AssignmentColumns([]string{"count", "sum", "min", "max", "last", "hist"}),
+		}).CreateInBatches(&rows, 200).Error
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+type pipelineStatus struct {
+	mu            sync.Mutex
+	LastFlush     int64  `json:"lastFlush"`
+	LastFlushMs   int64  `json:"lastFlushMs"`
+	LastFlushErr  string `json:"lastFlushError"`
+	LastRollup    int64  `json:"lastRollup"`
+	LastRetention int64  `json:"lastRetention"`
+	Dropped       int64  `json:"droppedPoints"`
+}
+
+var status pipelineStatus
+
+// flush writes every bucket that started before `before`.
+func flush(before int64) error {
+	flushMu.Lock()
+	defer flushMu.Unlock()
+
+	started := time.Now()
+	points := defaultAgg.takeClosed(before)
+	if len(points) == 0 {
+		return nil
+	}
+
+	fail := func(err error) error {
+		now := time.Now().Unix()
+		keep := make(map[pointKey]*Agg)
+		var dropped int64
+		for k, a := range points {
+			if now-k.bucket <= maxRetryAge {
+				keep[k] = a
+			} else {
+				dropped++
+			}
+		}
+		defaultAgg.putBack(keep)
+		status.mu.Lock()
+		status.LastFlushErr = err.Error()
+		status.Dropped += dropped
+		status.mu.Unlock()
+		if dropped > 0 {
+			PointsDropped.Add(float64(dropped))
+		}
+		return err
+	}
+
+	set := map[*Series]bool{}
+	for k := range points {
+		set[k.s] = true
+	}
+	list := make([]*Series, 0, len(set))
+	for s := range set {
+		list = append(list, s)
+	}
+	ids, err := resolveIDs(list)
+	if err != nil {
+		return fail(fmt.Errorf("resolve series: %w", err))
+	}
+
+	rows := make(map[pointID]*Agg, len(points))
+	for k, a := range points {
+		rows[pointID{ids[k.s], k.bucket}] = a
+	}
+	err = initializers.DB.Transaction(func(tx *gorm.DB) error {
+		return persist(tx, resMinute, rows)
+	})
+	if err != nil {
+		return fail(fmt.Errorf("write points: %w", err))
+	}
+
+	// last_seen lets the catalog hide series that stopped reporting
+	idList := make([]uint, 0, len(ids))
+	for _, id := range ids {
+		idList = append(idList, id)
+	}
+	for start := 0; start < len(idList); start += 500 {
+		end := min(start+500, len(idList))
+		initializers.DB.Model(&models.MetricSeries{}).Where("id IN ?", idList[start:end]).
+			Update("last_seen", time.Now().Unix())
+	}
+
+	status.mu.Lock()
+	status.LastFlush = time.Now().Unix()
+	status.LastFlushMs = time.Since(started).Milliseconds()
+	status.LastFlushErr = ""
+	status.mu.Unlock()
+	return nil
+}
+
+func hourOf(ts int64) int64 { return ts - ts%resHour }
+
+// rollup aggregates complete hours of minute data into hourly points. The last
+// rolled hour is redone each time to pick up late telemetry; it is idempotent.
+func rollup(now int64) error {
+	flushMu.Lock()
+	defer flushMu.Unlock()
+
+	// the minute data of an hour is complete ~1 flush after the hour ends
+	lastFull := hourOf(now - 2*resMinute)
+	start := rolledUntil.Load()
+	if start == 0 {
+		var maxHour, minMinute *int64
+		initializers.DB.Model(&models.MetricPoint{}).Where("resolution = ?", resHour).Select("MAX(bucket)").Scan(&maxHour)
+		if maxHour != nil {
+			start = *maxHour + resHour
+		} else {
+			initializers.DB.Model(&models.MetricPoint{}).Where("resolution = ?", resMinute).Select("MIN(bucket)").Scan(&minMinute)
+			if minMinute == nil {
+				status.mu.Lock()
+				status.LastRollup = now
+				status.mu.Unlock()
+				return nil
+			}
+			start = hourOf(*minMinute)
+		}
+	}
+	from := start - resHour
+	if earliest := hourOf(now - int64(initializers.MetricsMinuteRetentionDays)*86400); from < earliest {
+		from = earliest
+	}
+
+	for h := from; h < lastFull; h += resHour {
+		var rows []models.MetricPoint
+		err := initializers.DB.Where("resolution = ? AND bucket >= ? AND bucket < ?", resMinute, h, h+resHour).
+			Order("bucket asc").Find(&rows).Error
+		if err != nil {
+			return err
+		}
+		if len(rows) == 0 {
+			continue
+		}
+		merged := map[pointID]*Agg{}
+		for i := range rows {
+			k := pointID{rows[i].SeriesID, h}
+			a := merged[k]
+			if a == nil {
+				a = &Agg{}
+				merged[k] = a
+			}
+			a.merge(pointToAgg(&rows[i]))
+		}
+		// replace (not merge) so re-rolling an hour is idempotent
+		out := make([]models.MetricPoint, 0, len(merged))
+		for k, a := range merged {
+			out = append(out, aggToPoint(k.id, resHour, k.bucket, a))
+		}
+		err = initializers.DB.Clauses(clause.OnConflict{
+			Columns:   []clause.Column{{Name: "series_id"}, {Name: "resolution"}, {Name: "bucket"}},
+			DoUpdates: clause.AssignmentColumns([]string{"count", "sum", "min", "max", "last", "hist"}),
+		}).CreateInBatches(&out, 200).Error
+		if err != nil {
+			return err
+		}
+	}
+	if lastFull > rolledUntil.Load() {
+		rolledUntil.Store(lastFull)
+	}
+	status.mu.Lock()
+	status.LastRollup = now
+	status.mu.Unlock()
+	return nil
+}
+
+// retention deletes data older than the configured windows.
+func retention(now time.Time) error {
+	flushMu.Lock()
+	defer flushMu.Unlock()
+
+	day := int64(86400)
+	ts := now.Unix()
+	steps := []struct {
+		what string
+		run  func() error
+	}{
+		{"minute points", func() error {
+			return initializers.DB.Where("resolution = ? AND bucket < ?", resMinute, ts-int64(initializers.MetricsMinuteRetentionDays)*day).
+				Delete(&models.MetricPoint{}).Error
+		}},
+		{"hour points", func() error {
+			return initializers.DB.Where("resolution = ? AND bucket < ?", resHour, ts-int64(initializers.MetricsHourRetentionDays)*day).
+				Delete(&models.MetricPoint{}).Error
+		}},
+		{"events", func() error {
+			return initializers.DB.Where("ts < ?", (ts-int64(initializers.EventsRetentionDays)*day)*1000).
+				Delete(&models.ServiceEvent{}).Error
+		}},
+		{"top items", func() error {
+			cutoff := now.UTC().AddDate(0, 0, -initializers.TopItemsRetentionDays).Format("2006-01-02")
+			return initializers.DB.Where("day < ?", cutoff).Delete(&models.DailyTopItem{}).Error
+		}},
+		{"stale series", func() error {
+			stale := ts - 30*day
+			res := initializers.DB.Where("last_seen < ? AND id NOT IN (SELECT DISTINCT series_id FROM metric_points)", stale).
+				Delete(&models.MetricSeries{})
+			if res.Error == nil && res.RowsAffected > 0 {
+				// forget cached ids so a revived series gets a new row
+				idMu.Lock()
+				seriesID = map[*Series]uint{}
+				idMu.Unlock()
+			}
+			return res.Error
+		}},
+	}
+	for _, step := range steps {
+		if err := step.run(); err != nil {
+			log.Printf("metrics retention (%s): %v", step.what, err)
+		}
+	}
+	status.mu.Lock()
+	status.LastRetention = now.Unix()
+	status.mu.Unlock()
+	return nil
+}

metrics/traffic.go

@@ -0,0 +1,328 @@
+package metrics
+
+import (
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/binary"
+	"log"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
+	"gitea.tbdevent.eu/TBD/reforger_crawler_main/models"
+	"github.com/gin-gonic/gin"
+	"gorm.io/gorm/clause"
+)
+
+// Middleware records latency and traffic for every request except static admin
+// UI files and health checks.
+func Middleware() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		path := c.Request.URL.Path
+		if c.Request.Method == "OPTIONS" || path == "/health" || strings.HasPrefix(path, "/admin/ui") {
+			c.Next()
+			return
+		}
+		start := time.Now()
+		c.Next()
+		ms := float64(time.Since(start).Microseconds()) / 1000
+
+		route := c.FullPath()
+		if route == "" {
+			route = "(unmatched)"
+		}
+		status := c.Writer.Status()
+		group := routeGroup(path)
+		client := "direct"
+		if c.GetHeader("Origin") != "" || c.GetHeader("Referer") != "" {
+			client = "web"
+		}
+		HTTPDuration.Observe(ms, c.Request.Method, route, strconv.Itoa(status/100)+"xx", group, client)
+		if size := c.Writer.Size(); size > 0 {
+			HTTPResponseBytes.Add(float64(size), group)
+		}
+
+		if group == "v1" && status < 500 {
+			visitors.add(c.ClientIP(), c.Request.UserAgent())
+		}
+		if status == 200 {
+			switch route {
+			case "/v1/addon/:id":
+				Top.Add("addon", c.Param("id"))
+			case "/v1/getPossible":
+				Top.Add("search", normalizeSearch(c.Query("q")))
+			}
+		}
+	}
+}
+
+func routeGroup(path string) string {
+	for _, g := range []string{"v1", "back", "admin", "metrics"} {
+		if path == "/"+g || strings.HasPrefix(path, "/"+g+"/") {
+			return g
+		}
+	}
+	return "other"
+}
+
+func normalizeSearch(q string) string {
+	q = strings.Join(strings.Fields(strings.ToLower(q)), " ")
+	if len(q) > 64 {
+		q = q[:64]
+	}
+	return q
+}
+
+// visitorSet counts distinct (IP, user agent) pairs per UTC day. Only salted
+// 64-bit hashes are kept and the salt never leaves memory.
+type visitorSet struct {
+	mu     sync.Mutex
+	day    string
+	salt   []byte
+	humans map[uint64]struct{}
+	bots   map[uint64]struct{}
+}
+
+var visitors = &visitorSet{}
+
+const maxVisitors = 2_000_000
+
+func (v *visitorSet) rotate(day string) {
+	v.day = day
+	v.salt = make([]byte, 32)
+	_, _ = rand.Read(v.salt)
+	v.humans = make(map[uint64]struct{})
+	v.bots = make(map[uint64]struct{})
+}
+
+func isBot(ua string) bool {
+	ua = strings.ToLower(ua)
+	if ua == "" {
+		return true
+	}
+	for _, s := range []string{"bot", "spider", "crawl", "curl", "wget", "python", "go-http", "headless", "preview", "monitor"} {
+		if strings.Contains(ua, s) {
+			return true
+		}
+	}
+	return false
+}
+
+func (v *visitorSet) add(ip, ua string) {
+	day := time.Now().UTC().Format("2006-01-02")
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	if v.day != day {
+		v.rotate(day)
+	}
+	h := sha256.New()
+	h.Write(v.salt)
+	h.Write([]byte(ip))
+	h.Write([]byte{0})
+	h.Write([]byte(ua))
+	key := binary.BigEndian.Uint64(h.Sum(nil)[:8])
+	set := v.humans
+	if isBot(ua) {
+		set = v.bots
+	}
+	if len(set) < maxVisitors {
+		set[key] = struct{}{}
+	}
+}
+
+// Visitors returns today's approximate unique human and bot visitors. The set
+// lives in memory, so after a restart the day's highest recorded value is used
+// until the new set catches up.
+func Visitors() (humans, bots int) {
+	now := time.Now().UTC()
+	day := now.Format("2006-01-02")
+	visitors.mu.Lock()
+	if visitors.day == day {
+		humans, bots = len(visitors.humans), len(visitors.bots)
+	}
+	visitors.mu.Unlock()
+
+	dayStart := now.Truncate(24 * time.Hour).Unix()
+	if dayStart < startedAt.Unix() {
+		if t, err := Totals(Selector{Service: MainService, Name: "visitors_unique", GroupBy: []string{"kind"}}, dayStart, now.Unix()); err == nil {
+			for _, g := range t {
+				if g.Max == nil {
+					continue
+				}
+				switch g.Labels["kind"] {
+				case "human":
+					humans = max(humans, int(*g.Max))
+				case "bot":
+					bots = max(bots, int(*g.Max))
+				}
+			}
+		}
+	}
+	return humans, bots
+}
+
+// liveVisitors is the in-memory count only, used when recording the gauge.
+func liveVisitors() (humans, bots int) {
+	day := time.Now().UTC().Format("2006-01-02")
+	visitors.mu.Lock()
+	defer visitors.mu.Unlock()
+	if visitors.day != day {
+		return 0, 0
+	}
+	return len(visitors.humans), len(visitors.bots)
+}
+
+// topCounter counts requested items per UTC day and persists the most popular.
+type topCounter struct {
+	mu      sync.Mutex
+	day     string
+	counts  map[string]map[string]int64
+	pending []topSnapshot // previous days not yet written
+}
+
+type topSnapshot struct {
+	day    string
+	counts map[string]map[string]int64
+}
+
+// Top tracks the most looked-up addons and search queries.
+var Top = &topCounter{counts: map[string]map[string]int64{}}
+
+const (
+	maxTopKeys    = 50_000
+	topPersistMax = 500
+)
+
+func (t *topCounter) Add(kind, item string) {
+	if item == "" {
+		return
+	}
+	if len(item) > 128 {
+		item = item[:128]
+	}
+	day := time.Now().UTC().Format("2006-01-02")
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.day != day {
+		if t.day != "" {
+			t.pending = append(t.pending, topSnapshot{t.day, t.counts})
+		}
+		t.day = day
+		t.counts = map[string]map[string]int64{}
+	}
+	m := t.counts[kind]
+	if m == nil {
+		m = map[string]int64{}
+		t.counts[kind] = m
+	}
+	if _, ok := m[item]; !ok && len(m) >= maxTopKeys {
+		for k, v := range m {
+			if v <= 1 {
+				delete(m, k)
+			}
+		}
+		if len(m) >= maxTopKeys {
+			return
+		}
+	}
+	m[item]++
+}
+
+// Today returns today's in-memory counts for a kind.
+func (t *topCounter) Today(kind string) (string, map[string]int64) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	out := make(map[string]int64, len(t.counts[kind]))
+	for k, v := range t.counts[kind] {
+		out[k] = v
+	}
+	return t.day, out
+}
+
+// seed loads today's persisted counts so a restart does not reset them.
+func (t *topCounter) seed() {
+	day := time.Now().UTC().Format("2006-01-02")
+	var rows []models.DailyTopItem
+	if err := initializers.DB.Where("day = ?", day).Find(&rows).Error; err != nil {
+		log.Printf("metrics: failed to load top items: %v", err)
+		return
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.day != day {
+		t.day = day
+		t.counts = map[string]map[string]int64{}
+	}
+	for _, r := range rows {
+		if t.counts[r.Kind] == nil {
+			t.counts[r.Kind] = map[string]int64{}
+		}
+		t.counts[r.Kind][r.Item] += r.Count
+	}
+}
+
+func topRows(snap topSnapshot) []models.DailyTopItem {
+	var rows []models.DailyTopItem
+	for kind, m := range snap.counts {
+		type kv struct {
+			k string
+			v int64
+		}
+		list := make([]kv, 0, len(m))
+		for k, v := range m {
+			list = append(list, kv{k, v})
+		}
+		sort.Slice(list, func(i, j int) bool { return list[i].v > list[j].v })
+		if len(list) > topPersistMax {
+			list = list[:topPersistMax]
+		}
+		for _, e := range list {
+			rows = append(rows, models.DailyTopItem{Day: snap.day, Kind: kind, Item: e.k, Count: e.v})
+		}
+	}
+	return rows
+}
+
+func (t *topCounter) flush() error {
+	t.mu.Lock()
+	snaps := append([]topSnapshot{}, t.pending...)
+	t.pending = nil
+	cur := topSnapshot{day: t.day, counts: map[string]map[string]int64{}}
+	for kind, m := range t.counts {
+		cp := make(map[string]int64, len(m))
+		for k, v := range m {
+			cp[k] = v
+		}
+		cur.counts[kind] = cp
+	}
+	t.mu.Unlock()
+	if cur.day != "" {
+		snaps = append(snaps, cur)
+	}
+
+	for i, snap := range snaps {
+		rows := topRows(snap)
+		if len(rows) == 0 {
+			continue
+		}
+		err := initializers.DB.Clauses(clause.OnConflict{
+			Columns:   []clause.Column{{Name: "day"}, {Name: "kind"}, {Name: "item"}},
+			DoUpdates: clause.AssignmentColumns([]string{"count"}),
+		}).CreateInBatches(&rows, 200).Error
+		if err != nil {
+			// keep finished days for the next attempt; today is re-read from memory
+			t.mu.Lock()
+			for _, old := range snaps[i:] {
+				if old.day != cur.day {
+					t.pending = append(t.pending, old)
+				}
+			}
+			t.mu.Unlock()
+			return err
+		}
+	}
+	return nil
+}