Aggregate metrics into per-minute rollups for fast, small stats

cmd/hal/commands/stats.go

@@ -87,43 +87,126 @@ 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 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 {
 	since := time.Now().Add(-duration)
-	var result struct {
+
+	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)
 
+	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).
+		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 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)
+		var values []int64
+
+		db.Model(&store.Metric{}).
+			Select("value").
+			Where("metric_type = ? AND timestamp > ?", metricType, since).
+			Scan(&values)
+
+		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]))
+	}
+
 	since := time.Now().Add(-duration)
-	var values []int64
+	var rollups []store.MetricRollup
 
-	db.Model(&store.Metric{}).
-		Select("value").
-		Where("metric_type = ? AND timestamp > ?", metricType, since).
-		Scan(&values)
+	db.Model(&store.MetricRollup{}).
+		Select("histogram").
+		Where("metric_type = ? AND bucket_start > ?", metricType, since.Unix()).
+		Find(&rollups)
 
-	if len(values) == 0 {
-		return "0ms"
+	merged := make(map[int32]int64)
+	for _, r := range rollups {
+		merged = store.MergeHistograms(merged, r.Histogram)
 	}
 
-	sort.Slice(values, func(i, j int) bool {
-		return values[i] < values[j]
-	})
+	// 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)]++
+	}
 
-	index := int(math.Ceil(float64(len(values))*0.99)) - 1
-	if index < 0 {
-		index = 0
+	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 {

cmd/hal/commands/stats_test.go

@@ -190,38 +190,110 @@ 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 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}},
+	}
+	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 = 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))
 }
 
+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 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)