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 = ®istry{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) } } }