From de75f7a090e60ade8bc4ad8432f9c617a46c4e72 Mon Sep 17 00:00:00 2001 From: Daniel Simmons Date: Sat, 11 Jul 2026 09:49:11 +0200 Subject: [PATCH 1/2] Aggregate metrics into per-minute rollups for fast, small stats The metrics table grew to ~7GB (one raw row per tick, ~11.9M/month kept for 90 days) and `hal stats` took minutes because p99 was computed by loading every matching row into Go and sorting it. Store per-minute rollups (count, sum, and a mergeable log-scale histogram) and query those for the hour/day/month windows, keeping raw points only for the high-resolution "last minute" view: - store: add MetricRollup model + a DDSketch-style histogram (fixed bucket boundaries, so per-minute histograms sum exactly and a quantile over any window is a true quantile accurate to ~2% relative error). - metrics: roll up completed minutes each minute; one-time Go backfill of historical raw (chunked); prune raw to 24h and rollups to 90d, gating raw pruning on backfill completion so history is never lost. - stats: last minute from raw (exact); longer windows merge rollup histograms and compute the p99 of the merged distribution. Stays on SQLite / the pure-static CGO_ENABLED=0 build (DuckDB would require CGO, incompatible with the scratch-based deployment image). Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/hal/commands/stats.go | 79 ++++++++--- cmd/hal/commands/stats_test.go | 61 +++++--- metrics/service.go | 249 +++++++++++++++++++++++++++++---- metrics/service_test.go | 158 +++++++++++++++++---- store/histogram.go | 81 +++++++++++ store/histogram_test.go | 56 ++++++++ store/models.go | 22 ++- store/sqlite.go | 2 +- 8 files changed, 613 insertions(+), 95 deletions(-) create mode 100644 store/histogram.go create mode 100644 store/histogram_test.go diff --git a/cmd/hal/commands/stats.go b/cmd/hal/commands/stats.go index 95dc601..e4fdb76 100644 --- a/cmd/hal/commands/stats.go +++ b/cmd/hal/commands/stats.go @@ -87,43 +87,82 @@ func runStatsCommand(dbPath string) error { return printTable(summaries) } +// sumMetrics returns the total count of a counter metric over the given window. +// The "last minute" window is served from raw points (exact, high resolution); +// longer windows are served from pre-aggregated per-minute rollups, which keeps +// the query fast regardless of how much history exists. func sumMetrics(db *gorm.DB, metricType store.MetricType, duration time.Duration) int64 { - since := time.Now().Add(-duration) var result struct { Total int64 } - db.Model(&store.Metric{}). - Select("COALESCE(SUM(value), 0) as total"). - Where("metric_type = ? AND timestamp > ?", metricType, since). - Scan(&result) + if duration <= time.Minute { + since := time.Now().Add(-duration) + db.Model(&store.Metric{}). + Select("COALESCE(SUM(value), 0) as total"). + Where("metric_type = ? AND timestamp > ?", metricType, since). + Scan(&result) + } else { + since := time.Now().Add(-duration).Unix() + db.Model(&store.MetricRollup{}). + Select("COALESCE(SUM(count), 0) as total"). + Where("metric_type = ? AND bucket_start > ?", metricType, since). + Scan(&result) + } return result.Total } +// calculateP99 returns the p99 of a timer metric over the given window. The +// "last minute" window computes an exact p99 from raw points; longer windows +// merge the per-minute rollup histograms and compute the p99 of the merged +// distribution. Because the histograms share fixed bucket boundaries, the merge +// is exact and the result is a true p99 accurate to a bounded relative error. func calculateP99(db *gorm.DB, metricType store.MetricType, duration time.Duration) string { - since := time.Now().Add(-duration) - var values []int64 + if duration <= time.Minute { + since := time.Now().Add(-duration) + var values []int64 - db.Model(&store.Metric{}). - Select("value"). - Where("metric_type = ? AND timestamp > ?", metricType, since). - Scan(&values) + db.Model(&store.Metric{}). + Select("value"). + Where("metric_type = ? AND timestamp > ?", metricType, since). + Scan(&values) - if len(values) == 0 { - return "0ms" + if len(values) == 0 { + return "0ms" + } + + sort.Slice(values, func(i, j int) bool { + return values[i] < values[j] + }) + + index := int(math.Ceil(float64(len(values))*0.99)) - 1 + if index < 0 { + index = 0 + } + + return formatDuration(time.Duration(values[index])) } - sort.Slice(values, func(i, j int) bool { - return values[i] < values[j] - }) + since := time.Now().Add(-duration).Unix() + var rollups []store.MetricRollup + + db.Model(&store.MetricRollup{}). + Select("histogram"). + Where("metric_type = ? AND bucket_start > ?", metricType, since). + Find(&rollups) - index := int(math.Ceil(float64(len(values))*0.99)) - 1 - if index < 0 { - index = 0 + merged := make(map[int32]int64) + for _, r := range rollups { + merged = store.MergeHistograms(merged, r.Histogram) + } + + p99 := store.HistogramQuantile(merged, 0.99) + if p99 == 0 { + return "0ms" } - return formatDuration(time.Duration(values[index])) + return formatDuration(time.Duration(p99)) } func formatDuration(d time.Duration) string { diff --git a/cmd/hal/commands/stats_test.go b/cmd/hal/commands/stats_test.go index ab397f4..544a287 100644 --- a/cmd/hal/commands/stats_test.go +++ b/cmd/hal/commands/stats_test.go @@ -190,38 +190,61 @@ func TestSumMetrics(t *testing.T) { db := setupTestDBForStats(t) now := time.Now() - // Insert metrics within and outside the time window - metrics := []store.Metric{ - { - Timestamp: now.Add(-30 * time.Second), // Within 1 minute - MetricType: store.MetricTypeAutomationTriggered, - Value: 1, - }, - { - Timestamp: now.Add(-45 * time.Second), // Within 1 minute + // Raw points drive the high-resolution "last minute" window. + for _, ts := range []time.Time{now.Add(-30 * time.Second), now.Add(-45 * time.Second)} { + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: ts, MetricType: store.MetricTypeAutomationTriggered, Value: 1, - }, - { - Timestamp: now.Add(-2 * time.Minute), // Outside 1 minute window - MetricType: store.MetricTypeAutomationTriggered, - Value: 1, - }, + }).Error) } - for _, metric := range metrics { - assert.NilError(t, db.Create(&metric).Error) + // Rollups drive longer windows: two minute-buckets within the last 5 minutes + // totalling 3. + rollups := []store.MetricRollup{ + {MetricType: store.MetricTypeAutomationTriggered, BucketStart: now.Add(-2 * time.Minute).Truncate(time.Minute).Unix(), Count: 2, Sum: 2, Histogram: map[int32]int64{0: 2}}, + {MetricType: store.MetricTypeAutomationTriggered, BucketStart: now.Add(-3 * time.Minute).Truncate(time.Minute).Unix(), Count: 1, Sum: 1, Histogram: map[int32]int64{0: 1}}, + } + for _, r := range rollups { + assert.NilError(t, db.Create(&r).Error) } - // Test sum for last minute (should be 2) + // Last minute reads raw: 2 points. result := sumMetrics(db.DB, store.MetricTypeAutomationTriggered, time.Minute) assert.Equal(t, result, int64(2)) - // Test sum for last 5 minutes (should be 3) + // Last 5 minutes reads rollups: 2 + 1 = 3. result = sumMetrics(db.DB, store.MetricTypeAutomationTriggered, 5*time.Minute) assert.Equal(t, result, int64(3)) } +func TestCalculateP99FromRollups(t *testing.T) { + db := setupTestDBForStats(t) + + now := time.Now() + tenMs := (10 * time.Millisecond).Nanoseconds() + ninetyMs := (90 * time.Millisecond).Nanoseconds() + + // Two minutes of samples: 99 fast (10ms) and 100 slow (90ms). Merged, the p99 + // of the 199 samples (nearest-rank 198th) falls in the 90ms bucket, so the + // window p99 is a true quantile of the underlying distribution. + rollups := []store.MetricRollup{ + {MetricType: store.MetricTypeTickProcessingTime, BucketStart: now.Add(-2 * time.Minute).Truncate(time.Minute).Unix(), Count: 99, Sum: tenMs * 99, Histogram: map[int32]int64{store.HistogramBucket(tenMs): 99}}, + {MetricType: store.MetricTypeTickProcessingTime, BucketStart: now.Add(-3 * time.Minute).Truncate(time.Minute).Unix(), Count: 100, Sum: ninetyMs * 100, Histogram: map[int32]int64{store.HistogramBucket(ninetyMs): 100}}, + } + for _, r := range rollups { + assert.NilError(t, db.Create(&r).Error) + } + + expected := formatDuration(time.Duration(store.HistogramValue(store.HistogramBucket(ninetyMs)))) + result := calculateP99(db.DB, store.MetricTypeTickProcessingTime, 5*time.Minute) + assert.Equal(t, result, expected) + + // No rollups for this metric type -> "0ms". + none := calculateP99(db.DB, store.MetricTypeAutomationTriggered, 5*time.Minute) + assert.Equal(t, none, "0ms") +} + func TestCalculateP99(t *testing.T) { db := setupTestDBForStats(t) diff --git a/metrics/service.go b/metrics/service.go index 0446fea..bb0f605 100644 --- a/metrics/service.go +++ b/metrics/service.go @@ -1,35 +1,56 @@ package metrics import ( + "errors" "time" "github.com/dansimau/hal/logger" "github.com/dansimau/hal/store" "gorm.io/gorm" + "gorm.io/gorm/clause" ) -// Service handles metrics collection and pruning -// Uses async write queue for non-blocking performance +const ( + // rollupInterval is how often completed minutes are aggregated into rollups. + rollupInterval = time.Minute + // rawRetention is how long individual raw metric points are kept. Raw points + // are only needed for the high-resolution "last minute" view; everything + // longer is served from rollups, so this can be short. + rawRetention = 24 * time.Hour + // rollupRetention is how long per-minute rollups are kept. + rollupRetention = 90 * 24 * time.Hour + // pruneInterval is how often old raw points and rollups are deleted. + pruneInterval = time.Hour + // rollupLookback is how far back each incremental rollup pass reprocesses + // completed minutes. Reprocessing is idempotent (upsert) and guards against + // gaps if a pass is delayed or samples arrive slightly late. + rollupLookback = 5 * time.Minute +) + +// Service handles metrics collection, rollup and pruning. +// Raw points are written via the async write queue for non-blocking performance, +// then periodically aggregated into per-minute rollups for fast long-range queries. type Service struct { - db *store.Store - pruneInterval time.Duration // How often to prune old metrics (default: daily) - retentionTime time.Duration // How long to keep metrics (default: 3 months) - stopChan chan struct{} + db *store.Store + stopChan chan struct{} + // backfillDone is closed once the historical raw data has been rolled up. Raw + // pruning is held off until then so that un-rolled history is never deleted. + backfillDone chan struct{} } // NewService creates a new metrics service func NewService(db *store.Store) *Service { return &Service{ - db: db, - pruneInterval: 24 * time.Hour, // Prune daily - retentionTime: 90 * 24 * time.Hour, // Keep 3 months of metrics - stopChan: make(chan struct{}), + db: db, + stopChan: make(chan struct{}), + backfillDone: make(chan struct{}), } } -// Start begins the metrics pruning goroutine +// Start begins the rollup and pruning goroutines. func (s *Service) Start() { - go s.pruneMetrics() + go s.rollupLoop() + go s.pruneLoop() logger.Info("Metrics service started", "") } @@ -40,7 +61,6 @@ func (s *Service) Stop() { } // RecordCounter records a counter metric (value = 1) -// Writes directly to SQLite, leveraging WAL mode for performance func (s *Service) RecordCounter(metricType store.MetricType, entityID, automationName string) { metric := store.Metric{ Timestamp: time.Now(), @@ -56,7 +76,6 @@ func (s *Service) RecordCounter(metricType store.MetricType, entityID, automatio } // RecordTimer records a timer metric (value = duration in nanoseconds) -// Writes directly to SQLite, leveraging WAL mode for performance func (s *Service) RecordTimer(metricType store.MetricType, duration time.Duration, entityID, automationName string) { metric := store.Metric{ Timestamp: time.Now(), @@ -71,9 +90,20 @@ func (s *Service) RecordTimer(metricType store.MetricType, duration time.Duratio }) } -// pruneMetrics runs in a goroutine to periodically remove old metrics -func (s *Service) pruneMetrics() { - ticker := time.NewTicker(s.pruneInterval) +// rollupLoop performs a one-time backfill of any historical raw data, then +// incrementally rolls up completed minutes. +func (s *Service) rollupLoop() { + // Roll up all historical raw data so long-range queries have data immediately + // (rather than waiting for it to accumulate minute by minute). Runs once at + // startup. Only on success do we allow raw pruning to proceed, so that raw + // history is never deleted before it has been captured in rollups. + if err := backfillRollups(s.db.DB, time.Now()); err != nil { + logger.Error("Failed to backfill metric rollups", "", "error", err) + } else { + close(s.backfillDone) + } + + ticker := time.NewTicker(rollupInterval) defer ticker.Stop() for { @@ -81,16 +111,183 @@ func (s *Service) pruneMetrics() { case <-s.stopChan: return case <-ticker.C: - cutoffTime := time.Now().Add(-s.retentionTime) s.db.EnqueueWrite(func(db *gorm.DB) error { - result := db.Where("timestamp < ?", cutoffTime).Delete(&store.Metric{}) - if result.Error != nil { - logger.Error("Failed to prune old metrics", "", "error", result.Error) - } else if result.RowsAffected > 0 { - logger.Info("Pruned old metrics", "", "count", result.RowsAffected, "cutoff", cutoffTime) - } - return result.Error + return rollupRecent(db, time.Now()) }) } } -} \ No newline at end of file +} + +// pruneLoop periodically deletes old raw points and old rollups. +func (s *Service) pruneLoop() { + ticker := time.NewTicker(pruneInterval) + defer ticker.Stop() + + for { + select { + case <-s.stopChan: + return + case <-ticker.C: + now := time.Now() + + // Only prune raw once the historical backfill has completed. + pruneRaw := false + select { + case <-s.backfillDone: + pruneRaw = true + default: + } + + s.db.EnqueueWrite(func(db *gorm.DB) error { + return pruneOldData(db, now, pruneRaw) + }) + } + } +} + +// backfillHistogram is the raw-data chunk size processed per iteration during +// backfill. Chunking bounds memory when rolling up a large historical backlog. +const backfillChunk = 24 * time.Hour + +// backfillRollups rolls up all historical raw metrics (older than the current +// minute) into per-minute rollups. Histograms cannot be built in SQL, so raw +// data is read in day-sized chunks and aggregated in Go. Existing rollups are +// left untouched (ON CONFLICT DO NOTHING), so freshly computed incremental +// rollups always take precedence over the backfill. +func backfillRollups(db *gorm.DB, now time.Time) error { + upper := now.Truncate(time.Minute) // exclude the in-progress minute + + // Find the earliest raw point via the ORM so its timestamp deserialises + // correctly (a raw MIN() aggregate comes back as a string). + var earliest store.Metric + if err := db.Select("timestamp").Order("timestamp ASC").First(&earliest).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil // no raw data to back-fill + } + + return err + } + + for chunkStart := earliest.Timestamp.Truncate(time.Minute); chunkStart.Before(upper); chunkStart = chunkStart.Add(backfillChunk) { + chunkEnd := chunkStart.Add(backfillChunk) + if chunkEnd.After(upper) { + chunkEnd = upper + } + + if err := rollupWindow(db, chunkStart, chunkEnd, true); err != nil { + return err + } + } + + return nil +} + +// rollupRecent recomputes rollups for recently completed minutes from raw data. +// It is idempotent: existing rollups for the same bucket are overwritten with +// the freshly computed values. +func rollupRecent(db *gorm.DB, now time.Time) error { + upper := now.Truncate(time.Minute) // exclude the in-progress minute + return rollupWindow(db, upper.Add(-rollupLookback), upper, false) +} + +// rollupWindow aggregates raw metrics in [lower, upper) into per-minute rollups +// and upserts them. When keepExisting is true, existing rollups are preserved +// (used by backfill); otherwise they are overwritten (used by incremental +// rollups, which are authoritative for recent minutes). +func rollupWindow(db *gorm.DB, lower, upper time.Time, keepExisting bool) error { + var raw []store.Metric + if err := db. + Select("metric_type", "timestamp", "value"). + Where("timestamp >= ? AND timestamp < ?", lower, upper). + Find(&raw).Error; err != nil { + return err + } + + rollups := aggregate(raw) + if len(rollups) == 0 { + return nil + } + + onConflict := clause.OnConflict{ + Columns: []clause.Column{{Name: "metric_type"}, {Name: "bucket_start"}}, + DoUpdates: clause.AssignmentColumns([]string{"count", "sum", "histogram"}), + } + if keepExisting { + onConflict = clause.OnConflict{ + Columns: []clause.Column{{Name: "metric_type"}, {Name: "bucket_start"}}, + DoNothing: true, + } + } + + return db.Clauses(onConflict).Create(&rollups).Error +} + +// aggregate groups raw metrics into per-minute rollups, building a count, sum and +// log-scale histogram for each bucket. The histogram lets true quantiles be +// computed later over any range of merged rollups. +func aggregate(raw []store.Metric) []store.MetricRollup { + type key struct { + metricType store.MetricType + bucket int64 + } + + type accumulator struct { + count int64 + sum int64 + histogram map[int32]int64 + } + + accs := make(map[key]*accumulator) + for _, m := range raw { + k := key{m.MetricType, m.Timestamp.Truncate(time.Minute).Unix()} + + acc := accs[k] + if acc == nil { + acc = &accumulator{histogram: make(map[int32]int64)} + accs[k] = acc + } + + acc.count++ + acc.sum += m.Value + acc.histogram[store.HistogramBucket(m.Value)]++ + } + + rollups := make([]store.MetricRollup, 0, len(accs)) + for k, acc := range accs { + rollups = append(rollups, store.MetricRollup{ + MetricType: k.metricType, + BucketStart: k.bucket, + Count: acc.count, + Sum: acc.sum, + Histogram: acc.histogram, + }) + } + + return rollups +} + +// pruneOldData deletes rollups older than rollupRetention and, when pruneRaw is +// true, raw points older than rawRetention. Raw pruning is gated on the caller +// having confirmed the historical backfill completed, so that raw history is +// never deleted before it has been captured in rollups. +func pruneOldData(db *gorm.DB, now time.Time, pruneRaw bool) error { + if pruneRaw { + rawCutoff := now.Add(-rawRetention) + if result := db.Where("timestamp < ?", rawCutoff).Delete(&store.Metric{}); result.Error != nil { + logger.Error("Failed to prune old raw metrics", "", "error", result.Error) + return result.Error + } else if result.RowsAffected > 0 { + logger.Info("Pruned old raw metrics", "", "count", result.RowsAffected, "cutoff", rawCutoff) + } + } + + rollupCutoff := now.Add(-rollupRetention).Unix() + if result := db.Where("bucket_start < ?", rollupCutoff).Delete(&store.MetricRollup{}); result.Error != nil { + logger.Error("Failed to prune old metric rollups", "", "error", result.Error) + return result.Error + } else if result.RowsAffected > 0 { + logger.Info("Pruned old metric rollups", "", "count", result.RowsAffected, "cutoff", rollupCutoff) + } + + return nil +} diff --git a/metrics/service_test.go b/metrics/service_test.go index be4e58b..2013a53 100644 --- a/metrics/service_test.go +++ b/metrics/service_test.go @@ -1,6 +1,7 @@ package metrics import ( + "math" "testing" "time" @@ -28,8 +29,6 @@ func TestNewService(t *testing.T) { service := NewService(db) assert.Assert(t, service != nil) - assert.Equal(t, service.pruneInterval, 24*time.Hour) - assert.Equal(t, service.retentionTime, 90*24*time.Hour) // 3 months } func TestRecordCounter(t *testing.T) { @@ -99,48 +98,151 @@ func TestMultipleMetrics(t *testing.T) { assert.Equal(t, count, int64(3)) } -func TestPruneOldMetrics(t *testing.T) { +func TestPruneOldData(t *testing.T) { db := setupTestDB(t) - service := NewService(db) - // Create old and new metrics (100 days old vs 1 hour old) - oldTime := time.Now().Add(-100 * 24 * time.Hour) // 100 days old (older than 90 day retention) - newTime := time.Now().Add(-1 * time.Hour) // 1 hour old + now := time.Now() - oldMetric := store.Metric{ - Timestamp: oldTime, + // Raw metrics: one older than rawRetention (24h), one within it. + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: now.Add(-48 * time.Hour), MetricType: store.MetricTypeAutomationTriggered, Value: 1, - } + }).Error) + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: now.Add(-1 * time.Hour), + MetricType: store.MetricTypeAutomationTriggered, + Value: 1, + }).Error) + + // Rollups: one older than rollupRetention (90d), one within it. + assert.NilError(t, db.Create(&store.MetricRollup{ + MetricType: store.MetricTypeAutomationTriggered, + BucketStart: now.Add(-100 * 24 * time.Hour).Truncate(time.Minute).Unix(), + Count: 1, Sum: 1, Histogram: map[int32]int64{0: 1}, + }).Error) + assert.NilError(t, db.Create(&store.MetricRollup{ + MetricType: store.MetricTypeAutomationTriggered, + BucketStart: now.Add(-1 * time.Hour).Truncate(time.Minute).Unix(), + Count: 1, Sum: 1, Histogram: map[int32]int64{0: 1}, + }).Error) + + assert.NilError(t, pruneOldData(db.DB, now, true)) + + var rawCount, rollupCount int64 + db.Model(&store.Metric{}).Count(&rawCount) + db.Model(&store.MetricRollup{}).Count(&rollupCount) + + // Only the recent raw point and recent rollup should remain. + assert.Equal(t, rawCount, int64(1)) + assert.Equal(t, rollupCount, int64(1)) +} + +func TestPruneOldDataSkipsRawUntilBackfilled(t *testing.T) { + db := setupTestDB(t) - newMetric := store.Metric{ - Timestamp: newTime, + now := time.Now() + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: now.Add(-48 * time.Hour), MetricType: store.MetricTypeAutomationTriggered, Value: 1, + }).Error) + + // With pruneRaw=false, old raw points must be retained. + assert.NilError(t, pruneOldData(db.DB, now, false)) + + var rawCount int64 + db.Model(&store.Metric{}).Count(&rawCount) + assert.Equal(t, rawCount, int64(1)) +} + +func TestAggregate(t *testing.T) { + base := time.Date(2026, 7, 11, 12, 30, 0, 0, time.UTC) + + var raw []store.Metric + // 100 samples in one minute with values 1..100 (ns). True p99 (nearest-rank, + // ceil(100*0.99)=99, index 98) is 99. + for i := int64(1); i <= 100; i++ { + raw = append(raw, store.Metric{ + Timestamp: base.Add(time.Duration(i) * time.Millisecond), + MetricType: store.MetricTypeTickProcessingTime, + Value: i, + }) + } + + rollups := aggregate(raw) + assert.Equal(t, len(rollups), 1) + + r := rollups[0] + assert.Equal(t, r.BucketStart, base.Unix()) + assert.Equal(t, r.Count, int64(100)) + assert.Equal(t, r.Sum, int64(5050)) + + // The histogram-derived p99 should be within the bucketing's relative accuracy + // of the true value (99). + p99 := store.HistogramQuantile(r.Histogram, 0.99) + assert.Assert(t, math.Abs(float64(p99)-99)/99 <= 0.05, "p99 = %d, want ~99", p99) +} + +func TestBackfillRollups(t *testing.T) { + db := setupTestDB(t) + + base := time.Date(2026, 7, 11, 12, 30, 0, 0, time.UTC) + + // Two samples in minute 12:30, one in 12:31. + for _, ts := range []time.Time{base, base.Add(30 * time.Second), base.Add(90 * time.Second)} { + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: ts, + MetricType: store.MetricTypeAutomationTriggered, + Value: 1, + }).Error) } - // Insert metrics directly into database - assert.NilError(t, db.Create(&oldMetric).Error) - assert.NilError(t, db.Create(&newMetric).Error) + // Backfill as if "now" is well after the samples so both minutes are complete. + assert.NilError(t, backfillRollups(db.DB, base.Add(10*time.Minute))) + + var rollups []store.MetricRollup + assert.NilError(t, db.Order("bucket_start ASC").Find(&rollups).Error) + assert.Equal(t, len(rollups), 2) + assert.Equal(t, rollups[0].BucketStart, base.Unix()) + assert.Equal(t, rollups[0].Count, int64(2)) + assert.Equal(t, rollups[1].BucketStart, base.Add(time.Minute).Unix()) + assert.Equal(t, rollups[1].Count, int64(1)) - // Verify both metrics exist + // Backfill must not overwrite existing rollups (ON CONFLICT DO NOTHING). + assert.NilError(t, backfillRollups(db.DB, base.Add(10*time.Minute))) var count int64 - db.Model(&store.Metric{}).Count(&count) + db.Model(&store.MetricRollup{}).Count(&count) assert.Equal(t, count, int64(2)) +} - // Manually trigger pruning with the default retention time (90 days) - cutoffTime := time.Now().Add(-service.retentionTime) - result := db.Where("timestamp < ?", cutoffTime).Delete(&store.Metric{}) - assert.NilError(t, result.Error) +func TestRollupRecent(t *testing.T) { + db := setupTestDB(t) - // Only new metric should remain - db.Model(&store.Metric{}).Count(&count) - assert.Equal(t, count, int64(1)) + now := time.Now().Truncate(time.Minute) + // Put samples in the previous, now-completed minute. + bucket := now.Add(-time.Minute) + for i := int64(1); i <= 10; i++ { + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: bucket.Add(time.Duration(i) * time.Second), + MetricType: store.MetricTypeTickProcessingTime, + Value: i * 1000, + }).Error) + } - // Verify it's the new metric - var remaining store.Metric - db.First(&remaining) - assert.Assert(t, remaining.Timestamp.After(cutoffTime)) + assert.NilError(t, rollupRecent(db.DB, now)) + + var r store.MetricRollup + assert.NilError(t, db.Where("bucket_start = ?", bucket.Unix()).First(&r).Error) + assert.Equal(t, r.Count, int64(10)) + assert.Equal(t, r.Sum, int64(55000)) + + // Histogram must survive the round-trip through the JSON serializer. + var histTotal int64 + for _, c := range r.Histogram { + histTotal += c + } + assert.Equal(t, histTotal, int64(10)) } func TestServiceStartStop(t *testing.T) { diff --git a/store/histogram.go b/store/histogram.go new file mode 100644 index 0000000..e045a4a --- /dev/null +++ b/store/histogram.go @@ -0,0 +1,81 @@ +package store + +import ( + "math" + "slices" +) + +// HistogramGamma is the ratio between consecutive histogram bucket boundaries. +// Bucket i covers the range (gamma^(i-1), gamma^i], so the representative value +// of any bucket is within ~2% of every sample it contains. This is the DDSketch +// bucketing scheme: because the boundaries are fixed for all time (independent of +// the data), per-minute histograms can be summed exactly, and a quantile computed +// over any set of merged histograms is a true quantile of the underlying samples, +// accurate to within that relative error. +// +// gamma = (1+a)/(1-a) with a = 0.02 gives a relative accuracy of ~2%. +const HistogramGamma = 1.0408163265306123 + +var histogramLogGamma = math.Log(HistogramGamma) + +// HistogramBucket returns the histogram bucket index for a value. Values below 1 +// are clamped to bucket 0 (they are not meaningful for the nanosecond timings we +// record). +func HistogramBucket(value int64) int32 { + if value < 1 { + return 0 + } + + return int32(math.Ceil(math.Log(float64(value)) / histogramLogGamma)) +} + +// HistogramValue returns a representative value for a bucket index: the midpoint +// of the bucket's range, which is within HistogramGamma-relative error of every +// value the bucket contains. +func HistogramValue(bucket int32) int64 { + return int64(2 * math.Pow(HistogramGamma, float64(bucket)) / (HistogramGamma + 1)) +} + +// MergeHistograms sums a set of per-bucket histograms into a single histogram. +func MergeHistograms(histograms ...map[int32]int64) map[int32]int64 { + merged := make(map[int32]int64) + for _, h := range histograms { + for bucket, count := range h { + merged[bucket] += count + } + } + + return merged +} + +// HistogramQuantile returns the q-th quantile (0 < q <= 1) of the samples +// summarised by a merged histogram, using the nearest-rank method. The result is +// accurate to within HistogramGamma relative error of the true quantile. +func HistogramQuantile(counts map[int32]int64, q float64) int64 { + var total int64 + for _, c := range counts { + total += c + } + + if total == 0 { + return 0 + } + + buckets := make([]int32, 0, len(counts)) + for b := range counts { + buckets = append(buckets, b) + } + slices.Sort(buckets) + + rank := int64(math.Ceil(float64(total) * q)) + + var cumulative int64 + for _, b := range buckets { + cumulative += counts[b] + if cumulative >= rank { + return HistogramValue(b) + } + } + + return HistogramValue(buckets[len(buckets)-1]) +} diff --git a/store/histogram_test.go b/store/histogram_test.go new file mode 100644 index 0000000..250a73f --- /dev/null +++ b/store/histogram_test.go @@ -0,0 +1,56 @@ +package store_test + +import ( + "math" + "testing" + + "github.com/dansimau/hal/store" + "gotest.tools/v3/assert" +) + +func TestHistogramQuantileAccuracy(t *testing.T) { + t.Parallel() + + // A uniform distribution of 1..10000. The true p99 (nearest-rank) is 9900. + counts := make(map[int32]int64) + for i := int64(1); i <= 10000; i++ { + counts[store.HistogramBucket(i)]++ + } + + got := store.HistogramQuantile(counts, 0.99) + + // The result must be within the bucketing's relative accuracy of the true + // value (allowing a little slack for the nearest-rank bucket boundary). + relErr := math.Abs(float64(got)-9900) / 9900 + assert.Assert(t, relErr <= 0.05, "p99 = %d, relative error %.3f", got, relErr) +} + +func TestHistogramQuantileMergesExactly(t *testing.T) { + t.Parallel() + + // Splitting the same samples across separate per-minute histograms and merging + // them must yield the same quantile as summarising them all at once. + whole := make(map[int32]int64) + minuteA := make(map[int32]int64) + minuteB := make(map[int32]int64) + for i := int64(1); i <= 10000; i++ { + b := store.HistogramBucket(i) + whole[b]++ + if i%2 == 0 { + minuteA[b]++ + } else { + minuteB[b]++ + } + } + + merged := store.MergeHistograms(minuteA, minuteB) + + assert.Equal(t, store.HistogramQuantile(merged, 0.99), store.HistogramQuantile(whole, 0.99)) + assert.Equal(t, store.HistogramQuantile(merged, 0.50), store.HistogramQuantile(whole, 0.50)) +} + +func TestHistogramQuantileEmpty(t *testing.T) { + t.Parallel() + + assert.Equal(t, store.HistogramQuantile(map[int32]int64{}, 0.99), int64(0)) +} diff --git a/store/models.go b/store/models.go index 125bc37..e922380 100644 --- a/store/models.go +++ b/store/models.go @@ -14,7 +14,7 @@ type Model struct { type Entity struct { Model - ID string `gorm:"primaryKey"` + ID string `gorm:"primaryKey"` State *homeassistant.State `gorm:"serializer:json"` } @@ -37,6 +37,26 @@ type Metric struct { AutomationName string `gorm:"size:100"` // Optional: which automation was involved } +// MetricRollup represents a pre-aggregated, per-minute summary of raw metrics. +// Rollups are kept far longer than raw metrics (which are pruned aggressively), +// keeping long-range queries (last day/month) fast and storage small. +type MetricRollup struct { + ID uint `gorm:"primaryKey;autoIncrement"` + MetricType MetricType `gorm:"uniqueIndex:idx_rollup_type_bucket,priority:1;not null;size:50"` + // BucketStart is the Unix timestamp (seconds) of the start of the one-minute + // bucket. Using an integer minute boundary makes bucketing identical whether + // computed in Go or in SQL, so upserts conflict-match reliably. + BucketStart int64 `gorm:"uniqueIndex:idx_rollup_type_bucket,priority:2;not null"` + Count int64 `gorm:"not null"` // number of raw samples in the bucket + Sum int64 `gorm:"not null"` // sum of raw values in the bucket + // Histogram is a log-scale histogram of the raw values in the bucket, keyed by + // HistogramBucket index (see store/histogram.go). Because bucket boundaries are + // fixed across all time, per-minute histograms sum exactly, so a quantile + // computed over any range of rollups is a true quantile of the underlying + // samples, accurate to a bounded relative error. + Histogram map[int32]int64 `gorm:"serializer:json"` +} + // Log represents a single log entry type Log struct { ID uint `gorm:"primaryKey;autoIncrement"` diff --git a/store/sqlite.go b/store/sqlite.go index 134dcaf..600d5af 100644 --- a/store/sqlite.go +++ b/store/sqlite.go @@ -27,7 +27,7 @@ func Open(path string) (*Store, error) { return nil, err } - if err := db.AutoMigrate(&Entity{}, &Metric{}, &Log{}); err != nil { + if err := db.AutoMigrate(&Entity{}, &Metric{}, &MetricRollup{}, &Log{}); err != nil { return nil, err } From 254c1f7b08e44e207df346171f1e079902b32406 Mon Sep 17 00:00:00 2001 From: Daniel Simmons Date: Sat, 11 Jul 2026 10:14:39 +0200 Subject: [PATCH 2/2] Address review: watermark-based rollups + raw-tail stats Fixes the issues Codex flagged on the rollup approach: - Rolling up now resumes from a watermark (the latest existing rollup) up to the current minute, in day chunks. This single mechanism does the initial backfill, fills gaps after downtime, and does steady-state per-minute rollups, so a long backfill can no longer leave permanent holes between the backfill window and the incremental lookback. - A failed rollup pass changes nothing and is retried on the next tick (the watermark doesn't advance), instead of permanently disabling raw pruning via a one-shot flag. - Raw pruning is gated on the watermark reaching the raw cut-off, so raw is never deleted before it has been captured in rollups, and pruning resumes automatically once rollups catch up. - stats now serves hour/day/month from rollups PLUS the not-yet-rolled raw tail, and falls back to raw entirely when no rollups cover the window, so standalone/upgraded `hal stats` and the current minute are reported correctly rather than as 0. - Add PRAGMA busy_timeout so the rollup loop, async writer, and separate CLI processes wait for the write lock instead of erroring. Co-Authored-By: Claude Opus 4.8 (1M context) --- cmd/hal/commands/stats.go | 86 ++++++++++++---- cmd/hal/commands/stats_test.go | 55 +++++++++- metrics/service.go | 179 +++++++++++++++++---------------- metrics/service_test.go | 91 +++++++++++------ store/sqlite.go | 6 +- 5 files changed, 275 insertions(+), 142 deletions(-) diff --git a/cmd/hal/commands/stats.go b/cmd/hal/commands/stats.go index e4fdb76..8650447 100644 --- a/cmd/hal/commands/stats.go +++ b/cmd/hal/commands/stats.go @@ -88,36 +88,70 @@ func runStatsCommand(dbPath string) error { } // sumMetrics returns the total count of a counter metric over the given window. -// The "last minute" window is served from raw points (exact, high resolution); -// longer windows are served from pre-aggregated per-minute rollups, which keeps -// the query fast regardless of how much history exists. +// The "last minute" window is served entirely from raw points (exact, high +// resolution). Longer windows are served from pre-aggregated rollups plus the raw +// "tail" that has not been rolled up yet (the current minute and anything the +// rollup loop has not caught up on), so the result stays correct and current even +// when rollups are missing or lagging. func sumMetrics(db *gorm.DB, metricType store.MetricType, duration time.Duration) int64 { - var result struct { + since := time.Now().Add(-duration) + + if duration <= time.Minute { + return rawSum(db, metricType, since) + } + + var rollup struct { Total int64 } + db.Model(&store.MetricRollup{}). + Select("COALESCE(SUM(count), 0) as total"). + Where("metric_type = ? AND bucket_start > ?", metricType, since.Unix()). + Scan(&rollup) - if duration <= time.Minute { - since := time.Now().Add(-duration) - db.Model(&store.Metric{}). - Select("COALESCE(SUM(value), 0) as total"). - Where("metric_type = ? AND timestamp > ?", metricType, since). - Scan(&result) - } else { - since := time.Now().Add(-duration).Unix() - db.Model(&store.MetricRollup{}). - Select("COALESCE(SUM(count), 0) as total"). - Where("metric_type = ? AND bucket_start > ?", metricType, since). - Scan(&result) + return rollup.Total + rawSum(db, metricType, rawTailStart(db, metricType, since)) +} + +// rawSum returns the sum of raw metric values at or after the given time. +func rawSum(db *gorm.DB, metricType store.MetricType, since time.Time) int64 { + var result struct { + Total int64 } + db.Model(&store.Metric{}). + Select("COALESCE(SUM(value), 0) as total"). + Where("metric_type = ? AND timestamp >= ?", metricType, since). + Scan(&result) return result.Total } +// rawTailStart returns the time from which raw points should supplement rollups +// for a window beginning at `since`: the end of the latest rolled-up minute +// within the window, or `since` itself when no rollups cover the window (so the +// window falls back to raw entirely, e.g. before the first backfill has run). +// Because rolled minutes are contiguous up to the watermark, the raw tail never +// overlaps a rollup, so the two never double-count. +func rawTailStart(db *gorm.DB, metricType store.MetricType, since time.Time) time.Time { + var result struct { + Max *int64 + } + db.Model(&store.MetricRollup{}). + Select("MAX(bucket_start) as max"). + Where("metric_type = ? AND bucket_start > ?", metricType, since.Unix()). + Scan(&result) + + if result.Max == nil { + return since + } + + return time.Unix(*result.Max, 0).Add(time.Minute) +} + // calculateP99 returns the p99 of a timer metric over the given window. The // "last minute" window computes an exact p99 from raw points; longer windows -// merge the per-minute rollup histograms and compute the p99 of the merged -// distribution. Because the histograms share fixed bucket boundaries, the merge -// is exact and the result is a true p99 accurate to a bounded relative error. +// merge the per-minute rollup histograms with the not-yet-rolled raw tail and +// compute the p99 of the merged distribution. Because the histograms share fixed +// bucket boundaries, the merge is exact and the result is a true p99 accurate to +// a bounded relative error. func calculateP99(db *gorm.DB, metricType store.MetricType, duration time.Duration) string { if duration <= time.Minute { since := time.Now().Add(-duration) @@ -144,12 +178,12 @@ func calculateP99(db *gorm.DB, metricType store.MetricType, duration time.Durati return formatDuration(time.Duration(values[index])) } - since := time.Now().Add(-duration).Unix() + since := time.Now().Add(-duration) var rollups []store.MetricRollup db.Model(&store.MetricRollup{}). Select("histogram"). - Where("metric_type = ? AND bucket_start > ?", metricType, since). + Where("metric_type = ? AND bucket_start > ?", metricType, since.Unix()). Find(&rollups) merged := make(map[int32]int64) @@ -157,6 +191,16 @@ func calculateP99(db *gorm.DB, metricType store.MetricType, duration time.Durati merged = store.MergeHistograms(merged, r.Histogram) } + // Supplement with raw points not yet captured in a rollup. + var rawValues []int64 + db.Model(&store.Metric{}). + Select("value"). + Where("metric_type = ? AND timestamp >= ?", metricType, rawTailStart(db, metricType, since)). + Scan(&rawValues) + for _, v := range rawValues { + merged[store.HistogramBucket(v)]++ + } + p99 := store.HistogramQuantile(merged, 0.99) if p99 == 0 { return "0ms" diff --git a/cmd/hal/commands/stats_test.go b/cmd/hal/commands/stats_test.go index 544a287..e202ea8 100644 --- a/cmd/hal/commands/stats_test.go +++ b/cmd/hal/commands/stats_test.go @@ -199,8 +199,7 @@ func TestSumMetrics(t *testing.T) { }).Error) } - // Rollups drive longer windows: two minute-buckets within the last 5 minutes - // totalling 3. + // Rollups cover older minutes within the window: two buckets totalling 3. rollups := []store.MetricRollup{ {MetricType: store.MetricTypeAutomationTriggered, BucketStart: now.Add(-2 * time.Minute).Truncate(time.Minute).Unix(), Count: 2, Sum: 2, Histogram: map[int32]int64{0: 2}}, {MetricType: store.MetricTypeAutomationTriggered, BucketStart: now.Add(-3 * time.Minute).Truncate(time.Minute).Unix(), Count: 1, Sum: 1, Histogram: map[int32]int64{0: 1}}, @@ -213,8 +212,27 @@ func TestSumMetrics(t *testing.T) { result := sumMetrics(db.DB, store.MetricTypeAutomationTriggered, time.Minute) assert.Equal(t, result, int64(2)) - // Last 5 minutes reads rollups: 2 + 1 = 3. + // Last 5 minutes = rolled buckets (2 + 1 = 3) plus the not-yet-rolled raw tail + // (the 2 recent points) = 5. result = sumMetrics(db.DB, store.MetricTypeAutomationTriggered, 5*time.Minute) + assert.Equal(t, result, int64(5)) +} + +func TestSumMetricsFallsBackToRawWithoutRollups(t *testing.T) { + db := setupTestDBForStats(t) + + now := time.Now() + // No rollups exist (e.g. before the first backfill). A longer window must + // still count the raw points it can see rather than reporting 0. + for _, ts := range []time.Time{now.Add(-30 * time.Second), now.Add(-2 * time.Minute), now.Add(-4 * time.Minute)} { + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: ts, + MetricType: store.MetricTypeAutomationTriggered, + Value: 1, + }).Error) + } + + result := sumMetrics(db.DB, store.MetricTypeAutomationTriggered, 5*time.Minute) assert.Equal(t, result, int64(3)) } @@ -245,6 +263,37 @@ func TestCalculateP99FromRollups(t *testing.T) { assert.Equal(t, none, "0ms") } +func TestCalculateP99IncludesRawTail(t *testing.T) { + db := setupTestDBForStats(t) + + now := time.Now() + fastMs := (5 * time.Millisecond).Nanoseconds() + slowMs := (80 * time.Millisecond).Nanoseconds() + + // An older rolled-up minute of fast samples... + assert.NilError(t, db.Create(&store.MetricRollup{ + MetricType: store.MetricTypeTickProcessingTime, + BucketStart: now.Add(-3 * time.Minute).Truncate(time.Minute).Unix(), + Count: 100, Sum: fastMs * 100, + Histogram: map[int32]int64{store.HistogramBucket(fastMs): 100}, + }).Error) + + // ...plus recent raw samples not yet rolled up. A slow one in the raw tail must + // still influence the window p99. + for range 3 { + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: now.Add(-10 * time.Second), + MetricType: store.MetricTypeTickProcessingTime, + Value: slowMs, + }).Error) + } + + // 100 fast + 3 slow = 103 samples; nearest-rank p99 (102nd) is a slow sample. + expected := formatDuration(time.Duration(store.HistogramValue(store.HistogramBucket(slowMs)))) + result := calculateP99(db.DB, store.MetricTypeTickProcessingTime, 5*time.Minute) + assert.Equal(t, result, expected) +} + func TestCalculateP99(t *testing.T) { db := setupTestDBForStats(t) diff --git a/metrics/service.go b/metrics/service.go index bb0f605..9203e39 100644 --- a/metrics/service.go +++ b/metrics/service.go @@ -21,10 +21,9 @@ const ( rollupRetention = 90 * 24 * time.Hour // pruneInterval is how often old raw points and rollups are deleted. pruneInterval = time.Hour - // rollupLookback is how far back each incremental rollup pass reprocesses - // completed minutes. Reprocessing is idempotent (upsert) and guards against - // gaps if a pass is delayed or samples arrive slightly late. - rollupLookback = 5 * time.Minute + // backfillChunk is the raw-data window processed per iteration when catching + // up. Chunking bounds memory when rolling up a large historical backlog. + backfillChunk = 24 * time.Hour ) // Service handles metrics collection, rollup and pruning. @@ -33,17 +32,13 @@ const ( type Service struct { db *store.Store stopChan chan struct{} - // backfillDone is closed once the historical raw data has been rolled up. Raw - // pruning is held off until then so that un-rolled history is never deleted. - backfillDone chan struct{} } // NewService creates a new metrics service func NewService(db *store.Store) *Service { return &Service{ - db: db, - stopChan: make(chan struct{}), - backfillDone: make(chan struct{}), + db: db, + stopChan: make(chan struct{}), } } @@ -90,30 +85,23 @@ func (s *Service) RecordTimer(metricType store.MetricType, duration time.Duratio }) } -// rollupLoop performs a one-time backfill of any historical raw data, then -// incrementally rolls up completed minutes. +// rollupLoop rolls up completed minutes from raw data. The first pass catches up +// all historical raw (the initial backfill); subsequent passes roll up each newly +// completed minute. A single mechanism handles backfill, gap-filling after +// downtime, and steady state (see rollupPending). func (s *Service) rollupLoop() { - // Roll up all historical raw data so long-range queries have data immediately - // (rather than waiting for it to accumulate minute by minute). Runs once at - // startup. Only on success do we allow raw pruning to proceed, so that raw - // history is never deleted before it has been captured in rollups. - if err := backfillRollups(s.db.DB, time.Now()); err != nil { - logger.Error("Failed to backfill metric rollups", "", "error", err) - } else { - close(s.backfillDone) - } - ticker := time.NewTicker(rollupInterval) defer ticker.Stop() for { + if err := s.rollupPending(time.Now()); err != nil { + logger.Error("Failed to roll up metrics", "", "error", err) + } + select { case <-s.stopChan: return case <-ticker.C: - s.db.EnqueueWrite(func(db *gorm.DB) error { - return rollupRecent(db, time.Now()) - }) } } } @@ -128,53 +116,38 @@ func (s *Service) pruneLoop() { case <-s.stopChan: return case <-ticker.C: - now := time.Now() - - // Only prune raw once the historical backfill has completed. - pruneRaw := false - select { - case <-s.backfillDone: - pruneRaw = true - default: + if err := pruneOldData(s.db.DB, time.Now()); err != nil { + logger.Error("Failed to prune metrics", "", "error", err) } - - s.db.EnqueueWrite(func(db *gorm.DB) error { - return pruneOldData(db, now, pruneRaw) - }) } } } -// backfillHistogram is the raw-data chunk size processed per iteration during -// backfill. Chunking bounds memory when rolling up a large historical backlog. -const backfillChunk = 24 * time.Hour - -// backfillRollups rolls up all historical raw metrics (older than the current -// minute) into per-minute rollups. Histograms cannot be built in SQL, so raw -// data is read in day-sized chunks and aggregated in Go. Existing rollups are -// left untouched (ON CONFLICT DO NOTHING), so freshly computed incremental -// rollups always take precedence over the backfill. -func backfillRollups(db *gorm.DB, now time.Time) error { +// rollupPending rolls up every completed minute that has not been rolled up yet: +// from just after the latest existing rollup (or the earliest raw point if none +// exist) up to the current minute, in day-sized chunks. This one mechanism does +// the initial historical backfill, closes gaps left by downtime or a slow +// backfill, and performs steady-state per-minute rollups. Because the resume +// point is derived from the rollups themselves, a failed pass changes nothing and +// is simply retried on the next tick. +func (s *Service) rollupPending(now time.Time) error { upper := now.Truncate(time.Minute) // exclude the in-progress minute - // Find the earliest raw point via the ORM so its timestamp deserialises - // correctly (a raw MIN() aggregate comes back as a string). - var earliest store.Metric - if err := db.Select("timestamp").Order("timestamp ASC").First(&earliest).Error; err != nil { - if errors.Is(err, gorm.ErrRecordNotFound) { - return nil // no raw data to back-fill - } - + lower, ok, err := rollupResumePoint(s.db.DB) + if err != nil { return err } + if !ok { + return nil // no raw data to roll up + } - for chunkStart := earliest.Timestamp.Truncate(time.Minute); chunkStart.Before(upper); chunkStart = chunkStart.Add(backfillChunk) { + for chunkStart := lower; chunkStart.Before(upper); chunkStart = chunkStart.Add(backfillChunk) { chunkEnd := chunkStart.Add(backfillChunk) if chunkEnd.After(upper) { chunkEnd = upper } - if err := rollupWindow(db, chunkStart, chunkEnd, true); err != nil { + if err := rollupWindow(s.db.DB, chunkStart, chunkEnd); err != nil { return err } } @@ -182,19 +155,55 @@ func backfillRollups(db *gorm.DB, now time.Time) error { return nil } -// rollupRecent recomputes rollups for recently completed minutes from raw data. -// It is idempotent: existing rollups for the same bucket are overwritten with -// the freshly computed values. -func rollupRecent(db *gorm.DB, now time.Time) error { - upper := now.Truncate(time.Minute) // exclude the in-progress minute - return rollupWindow(db, upper.Add(-rollupLookback), upper, false) +// rollupResumePoint returns the minute from which rolling up should resume: just +// after the latest existing rollup, or the earliest raw point if none exist. The +// bool is false when there is no raw data at all. +func rollupResumePoint(db *gorm.DB) (time.Time, bool, error) { + watermark, ok, err := latestRolledMinute(db) + if err != nil { + return time.Time{}, false, err + } + if ok { + // Resume at the minute after the latest rolled-up one. + return time.Unix(watermark, 0).Add(time.Minute), true, nil + } + + // No rollups yet: start from the earliest raw point. Fetch it via the ORM so + // the timestamp deserialises correctly (a raw MIN() aggregate comes back as a + // string). + var earliest store.Metric + if err := db.Select("timestamp").Order("timestamp ASC").First(&earliest).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return time.Time{}, false, nil + } + + return time.Time{}, false, err + } + + return earliest.Timestamp.Truncate(time.Minute), true, nil +} + +// latestRolledMinute returns the BucketStart (Unix seconds) of the most recent +// rollup, or ok=false when there are no rollups. +func latestRolledMinute(db *gorm.DB) (int64, bool, error) { + var result struct { + Max *int64 + } + if err := db.Model(&store.MetricRollup{}).Select("MAX(bucket_start) as max").Scan(&result).Error; err != nil { + return 0, false, err + } + if result.Max == nil { + return 0, false, nil + } + + return *result.Max, true, nil } // rollupWindow aggregates raw metrics in [lower, upper) into per-minute rollups -// and upserts them. When keepExisting is true, existing rollups are preserved -// (used by backfill); otherwise they are overwritten (used by incremental -// rollups, which are authoritative for recent minutes). -func rollupWindow(db *gorm.DB, lower, upper time.Time, keepExisting bool) error { +// and upserts them, overwriting any existing rollup for the same bucket. Since +// rollups are recomputed contiguously from the watermark, overwriting is +// idempotent. +func rollupWindow(db *gorm.DB, lower, upper time.Time) error { var raw []store.Metric if err := db. Select("metric_type", "timestamp", "value"). @@ -208,18 +217,10 @@ func rollupWindow(db *gorm.DB, lower, upper time.Time, keepExisting bool) error return nil } - onConflict := clause.OnConflict{ + return db.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "metric_type"}, {Name: "bucket_start"}}, DoUpdates: clause.AssignmentColumns([]string{"count", "sum", "histogram"}), - } - if keepExisting { - onConflict = clause.OnConflict{ - Columns: []clause.Column{{Name: "metric_type"}, {Name: "bucket_start"}}, - DoNothing: true, - } - } - - return db.Clauses(onConflict).Create(&rollups).Error + }).Create(&rollups).Error } // aggregate groups raw metrics into per-minute rollups, building a count, sum and @@ -266,19 +267,29 @@ func aggregate(raw []store.Metric) []store.MetricRollup { return rollups } -// pruneOldData deletes rollups older than rollupRetention and, when pruneRaw is -// true, raw points older than rawRetention. Raw pruning is gated on the caller -// having confirmed the historical backfill completed, so that raw history is -// never deleted before it has been captured in rollups. -func pruneOldData(db *gorm.DB, now time.Time, pruneRaw bool) error { - if pruneRaw { - rawCutoff := now.Add(-rawRetention) +// pruneOldData deletes rollups older than rollupRetention and raw points older +// than rawRetention. Raw is only pruned once it has been captured in rollups: the +// rollup watermark must have advanced past the raw cut-off. This means an +// incomplete or failed catch-up simply defers raw pruning (no data is lost), and +// pruning resumes automatically once the rollups catch up. +func pruneOldData(db *gorm.DB, now time.Time) error { + rawCutoff := now.Add(-rawRetention) + + watermark, ok, err := latestRolledMinute(db) + if err != nil { + return err + } + + switch { + case ok && watermark >= rawCutoff.Unix(): if result := db.Where("timestamp < ?", rawCutoff).Delete(&store.Metric{}); result.Error != nil { logger.Error("Failed to prune old raw metrics", "", "error", result.Error) return result.Error } else if result.RowsAffected > 0 { logger.Info("Pruned old raw metrics", "", "count", result.RowsAffected, "cutoff", rawCutoff) } + default: + logger.Info("Deferring raw metric prune until rollups catch up", "", "cutoff", rawCutoff) } rollupCutoff := now.Add(-rollupRetention).Unix() diff --git a/metrics/service_test.go b/metrics/service_test.go index 2013a53..bdf7344 100644 --- a/metrics/service_test.go +++ b/metrics/service_test.go @@ -127,18 +127,19 @@ func TestPruneOldData(t *testing.T) { Count: 1, Sum: 1, Histogram: map[int32]int64{0: 1}, }).Error) - assert.NilError(t, pruneOldData(db.DB, now, true)) + assert.NilError(t, pruneOldData(db.DB, now)) var rawCount, rollupCount int64 db.Model(&store.Metric{}).Count(&rawCount) db.Model(&store.MetricRollup{}).Count(&rollupCount) - // Only the recent raw point and recent rollup should remain. + // Only the recent raw point and recent rollup should remain. The watermark + // (latest rollup, 1h ago) is past the 24h raw cut-off, so raw is pruned. assert.Equal(t, rawCount, int64(1)) assert.Equal(t, rollupCount, int64(1)) } -func TestPruneOldDataSkipsRawUntilBackfilled(t *testing.T) { +func TestPruneOldDataDefersRawUntilRolledUp(t *testing.T) { db := setupTestDB(t) now := time.Now() @@ -148,12 +149,25 @@ func TestPruneOldDataSkipsRawUntilBackfilled(t *testing.T) { Value: 1, }).Error) - // With pruneRaw=false, old raw points must be retained. - assert.NilError(t, pruneOldData(db.DB, now, false)) + // No rollups exist yet, so the watermark has not reached the raw cut-off: old + // raw must be retained rather than lost. + assert.NilError(t, pruneOldData(db.DB, now)) var rawCount int64 db.Model(&store.Metric{}).Count(&rawCount) assert.Equal(t, rawCount, int64(1)) + + // A stale rollup whose watermark is still behind the cut-off must also defer + // raw pruning. + assert.NilError(t, db.Create(&store.MetricRollup{ + MetricType: store.MetricTypeAutomationTriggered, + BucketStart: now.Add(-72 * time.Hour).Truncate(time.Minute).Unix(), + Count: 1, Sum: 1, Histogram: map[int32]int64{0: 1}, + }).Error) + + assert.NilError(t, pruneOldData(db.DB, now)) + db.Model(&store.Metric{}).Count(&rawCount) + assert.Equal(t, rawCount, int64(1)) } func TestAggregate(t *testing.T) { @@ -184,8 +198,9 @@ func TestAggregate(t *testing.T) { assert.Assert(t, math.Abs(float64(p99)-99)/99 <= 0.05, "p99 = %d, want ~99", p99) } -func TestBackfillRollups(t *testing.T) { +func TestRollupPendingBackfillsHistory(t *testing.T) { db := setupTestDB(t) + service := NewService(db) base := time.Date(2026, 7, 11, 12, 30, 0, 0, time.UTC) @@ -198,8 +213,8 @@ func TestBackfillRollups(t *testing.T) { }).Error) } - // Backfill as if "now" is well after the samples so both minutes are complete. - assert.NilError(t, backfillRollups(db.DB, base.Add(10*time.Minute))) + // Roll up as if "now" is well after the samples so both minutes are complete. + assert.NilError(t, service.rollupPending(base.Add(10*time.Minute))) var rollups []store.MetricRollup assert.NilError(t, db.Order("bucket_start ASC").Find(&rollups).Error) @@ -209,40 +224,52 @@ func TestBackfillRollups(t *testing.T) { assert.Equal(t, rollups[1].BucketStart, base.Add(time.Minute).Unix()) assert.Equal(t, rollups[1].Count, int64(1)) - // Backfill must not overwrite existing rollups (ON CONFLICT DO NOTHING). - assert.NilError(t, backfillRollups(db.DB, base.Add(10*time.Minute))) + // Re-running is idempotent: contiguous re-rollup yields the same rollups. + assert.NilError(t, service.rollupPending(base.Add(10*time.Minute))) var count int64 db.Model(&store.MetricRollup{}).Count(&count) assert.Equal(t, count, int64(2)) + + // Histograms must survive the round-trip through the JSON serializer. + var r store.MetricRollup + assert.NilError(t, db.Where("bucket_start = ?", base.Unix()).First(&r).Error) + var histTotal int64 + for _, c := range r.Histogram { + histTotal += c + } + assert.Equal(t, histTotal, int64(2)) } -func TestRollupRecent(t *testing.T) { +func TestRollupPendingResumesFromWatermarkWithoutGaps(t *testing.T) { db := setupTestDB(t) + service := NewService(db) - now := time.Now().Truncate(time.Minute) - // Put samples in the previous, now-completed minute. - bucket := now.Add(-time.Minute) - for i := int64(1); i <= 10; i++ { - assert.NilError(t, db.Create(&store.Metric{ - Timestamp: bucket.Add(time.Duration(i) * time.Second), - MetricType: store.MetricTypeTickProcessingTime, - Value: i * 1000, - }).Error) - } + // Use local-tz times (as production does via time.Now()) so the stored + // timestamps and the watermark-derived resume bound compare consistently. + base := time.Now().Truncate(time.Minute).Add(-10 * time.Minute) - assert.NilError(t, rollupRecent(db.DB, now)) + // First pass rolls up the first minute. + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: base.Add(10 * time.Second), + MetricType: store.MetricTypeTickProcessingTime, + Value: 1000, + }).Error) + assert.NilError(t, service.rollupPending(base.Add(time.Minute))) - var r store.MetricRollup - assert.NilError(t, db.Where("bucket_start = ?", bucket.Unix()).First(&r).Error) - assert.Equal(t, r.Count, int64(10)) - assert.Equal(t, r.Sum, int64(55000)) + // Later, data appears two minutes on (the minute in between is a gap with no + // data). The next pass must still roll up that later minute by resuming from + // the watermark rather than a fixed short lookback, so a long backfill can + // never leave permanent holes. + assert.NilError(t, db.Create(&store.Metric{ + Timestamp: base.Add(2*time.Minute + 10*time.Second), + MetricType: store.MetricTypeTickProcessingTime, + Value: 2000, + }).Error) + assert.NilError(t, service.rollupPending(base.Add(3*time.Minute))) - // Histogram must survive the round-trip through the JSON serializer. - var histTotal int64 - for _, c := range r.Histogram { - histTotal += c - } - assert.Equal(t, histTotal, int64(10)) + var buckets []int64 + assert.NilError(t, db.Model(&store.MetricRollup{}).Order("bucket_start ASC").Pluck("bucket_start", &buckets).Error) + assert.DeepEqual(t, buckets, []int64{base.Unix(), base.Add(2 * time.Minute).Unix()}) } func TestServiceStartStop(t *testing.T) { diff --git a/store/sqlite.go b/store/sqlite.go index 600d5af..b27fbc4 100644 --- a/store/sqlite.go +++ b/store/sqlite.go @@ -12,8 +12,10 @@ type Store struct { } func Open(path string) (*Store, error) { - // Add auto_vacuum pragma to DSN - must be set before database is created - dsn := path + "?_pragma=auto_vacuum(FULL)" + // auto_vacuum must be set before the database is created. busy_timeout lets + // concurrent writers (the async writer, the rollup loop, and separate CLI + // processes) wait for the lock instead of failing with SQLITE_BUSY. + dsn := path + "?_pragma=auto_vacuum(FULL)&_pragma=busy_timeout(5000)" db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{}) if err != nil { return nil, err