reforger_crawler_main/metrics/metrics_test.go

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)
		}
	}
}