From 27507a47ad74e92b3e097b96bea3cc66fcd496de Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 27 Mar 2026 12:16:31 +0000 Subject: [PATCH] perf(metrics): keep downsampled series ordered --- .../subsystems/performance-and-scalability.md | 6 +++ pkg/metrics/store.go | 49 +++++++++++++++++-- pkg/metrics/store_slo_test.go | 7 +++ pkg/metrics/store_test.go | 33 +++++++++++++ 4 files changed, 91 insertions(+), 4 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index f75d816c2..f21029a8d 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -204,6 +204,12 @@ and `frontend-modern/src/components/Dashboard/useDashboardState.ts` must keep filters, stats, and table visibility driven by the workload route's own REST-backed health so websocket churn does not hide already-fetched workloads or swap the protected hot path into a false disconnected shell. +The same performance ownership now applies to the downsampled +`pkg/metrics/store.go` batched query path. `QueryAllBatch` may drop the global +SQLite `ORDER BY resource_id, metric_type, bucket_ts` sort from the grouped +query only if the runtime preserves ascending timestamps within every returned +resource/metric series and the hot-path proof keeps both latency and ordering +guarded together in the explicit metrics SLO surface. That runtime is now intentionally split by concern: `frontend-modern/src/components/Dashboard/useDashboardState.ts` owns top-level dashboard orchestration, workload loading, and composition across diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index 081135778..c391664d6 100644 --- a/pkg/metrics/store.go +++ b/pkg/metrics/store.go @@ -8,6 +8,7 @@ import ( "net/url" "os" "path/filepath" + "sort" "strconv" "strings" "sync" @@ -890,7 +891,6 @@ func (s *Store) queryAllBatchWithTier(resourceType string, resourceIDs []string, WHERE resource_type = ? AND resource_id IN (%s) AND tier = ? AND timestamp >= ? AND timestamp <= ? GROUP BY resource_id, metric_type, bucket_ts - ORDER BY resource_id, metric_type, bucket_ts ASC `, inClause) } else { params = make([]interface{}, 0, len(resourceIDs)+4) @@ -925,7 +925,8 @@ func (s *Store) queryAllBatchWithTier(resourceType string, resourceIDs []string, } defer rows.Close() - result := make(map[string]map[string][]MetricPoint) + result := make(map[string]map[string][]MetricPoint, len(resourceIDs)) + seriesCapacity := estimateQueryAllBatchSeriesCapacity(start, end, stepSecs) for rows.Next() { var resourceID, metricType string var ts int64 @@ -937,12 +938,52 @@ func (s *Store) queryAllBatchWithTier(resourceType string, resourceIDs []string, p.Timestamp = time.Unix(ts, 0) if _, exists := result[resourceID]; !exists { - result[resourceID] = make(map[string][]MetricPoint) + result[resourceID] = make(map[string][]MetricPoint, 8) + } + if _, exists := result[resourceID][metricType]; !exists { + result[resourceID][metricType] = make([]MetricPoint, 0, seriesCapacity) } result[resourceID][metricType] = append(result[resourceID][metricType], p) } - return result, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + if stepSecs > 1 { + sortMetricSeriesPoints(result) + } + + return result, nil +} + +func estimateQueryAllBatchSeriesCapacity(start, end time.Time, stepSecs int64) int { + if stepSecs <= 1 { + return 0 + } + if !end.After(start) { + return 1 + } + + stepDuration := time.Duration(stepSecs) * time.Second + buckets := int(end.Sub(start)/stepDuration) + 2 + if buckets < 1 { + return 1 + } + return buckets +} + +func sortMetricSeriesPoints(result map[string]map[string][]MetricPoint) { + for _, metricMap := range result { + for metricType, points := range metricMap { + if len(points) < 2 { + continue + } + sort.Slice(points, func(i, j int) bool { + return points[i].Timestamp.Before(points[j].Timestamp) + }) + metricMap[metricType] = points + } + } } // selectTier chooses the appropriate data tier based on time range diff --git a/pkg/metrics/store_slo_test.go b/pkg/metrics/store_slo_test.go index 668f40e14..a5066079c 100644 --- a/pkg/metrics/store_slo_test.go +++ b/pkg/metrics/store_slo_test.go @@ -385,6 +385,13 @@ func TestSLO_QueryAllBatch(t *testing.T) { if len(result[id]) != len(metricTypes) { t.Fatalf("sanity: expected %d metric types for %s, got %d", len(metricTypes), id, len(result[id])) } + for metricType, points := range result[id] { + for i := 1; i < len(points); i++ { + if points[i].Timestamp.Before(points[i-1].Timestamp) { + t.Fatalf("sanity: expected ascending timestamps for %s/%s, got %v before %v", id, metricType, points[i], points[i-1]) + } + } + } } latencies := measureLatencies(t, func() { diff --git a/pkg/metrics/store_test.go b/pkg/metrics/store_test.go index b916c733c..b3356b860 100644 --- a/pkg/metrics/store_test.go +++ b/pkg/metrics/store_test.go @@ -565,6 +565,39 @@ func TestQueryAllBatch(t *testing.T) { } }) + t.Run("downsampled results stay ordered per metric series", func(t *testing.T) { + downsampledStoreDir := t.TempDir() + downsampledCfg := DefaultConfig(downsampledStoreDir) + downsampledCfg.DBPath = filepath.Join(downsampledStoreDir, "metrics-batch-downsampled.db") + downsampledCfg.FlushInterval = time.Hour + + downsampledStore, err := NewStore(downsampledCfg) + if err != nil { + t.Fatalf("NewStore(downsampled): %v", err) + } + defer downsampledStore.Close() + + base := time.Unix(10_000, 0) + downsampledStore.writeBatch([]bufferedMetric{ + {resourceType: "disk", resourceID: "disk-ordered", metricType: "smart_temp", value: 33, timestamp: base.Add(130 * time.Second), tier: TierRaw}, + {resourceType: "disk", resourceID: "disk-ordered", metricType: "smart_temp", value: 31, timestamp: base.Add(10 * time.Second), tier: TierRaw}, + {resourceType: "disk", resourceID: "disk-ordered", metricType: "smart_temp", value: 32, timestamp: base.Add(70 * time.Second), tier: TierRaw}, + }) + + batch, err := downsampledStore.QueryAllBatch("disk", []string{"disk-ordered"}, base, base.Add(3*time.Minute), 60) + if err != nil { + t.Fatalf("QueryAllBatch(downsampled): %v", err) + } + + points := batch["disk-ordered"]["smart_temp"] + if len(points) != 3 { + t.Fatalf("expected 3 bucketed points, got %d", len(points)) + } + if !points[0].Timestamp.Before(points[1].Timestamp) || !points[1].Timestamp.Before(points[2].Timestamp) { + t.Fatalf("expected ascending timestamps, got %+v", points) + } + }) + t.Run("chunked resource lists return complete results beyond sqlite parameter threshold", func(t *testing.T) { chunkedStoreDir := t.TempDir() chunkedCfg := DefaultConfig(chunkedStoreDir)