From 931a2a6e79ad72fc019ded31ef037b1df698d943 Mon Sep 17 00:00:00 2001 From: Aditya Ujeniya Date: Mon, 20 Jul 2026 10:22:16 +0200 Subject: [PATCH] feat(metricstore): maintain running buffer stats on write --- pkg/metricstore/buffer.go | 21 +++++++++++++ pkg/metricstore/stats_test.go | 56 +++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/pkg/metricstore/buffer.go b/pkg/metricstore/buffer.go index 52e3b287..69b401f3 100644 --- a/pkg/metricstore/buffer.go +++ b/pkg/metricstore/buffer.go @@ -207,6 +207,23 @@ func newBuffer(ts, freq int64) *buffer { return b } +// updateStats folds a single value into the buffer's running statistics. +// NaN values are ignored so they never affect min/max/sum/samples. +func (b *buffer) updateStats(value schema.Float) { + xf := float64(value) + if math.IsNaN(xf) { + return + } + b.statSum += xf + b.statSamples++ + if xf < b.statMin { + b.statMin = xf + } + if xf > b.statMax { + b.statMax = xf + } +} + // write appends a timestamped value to the buffer chain. // // Returns the head buffer (which may be newly created if capacity was reached). @@ -241,6 +258,9 @@ func (b *buffer) write(ts int64, value schema.Float) (*buffer, error) { // Overwriting value or writing value from past if idx < len(b.data) { b.data[idx] = value + // The replaced value may have been the running min/max; the cached + // aggregate can no longer be trusted. Force a recompute on next query. + b.statsValid = false return b, nil } @@ -250,6 +270,7 @@ func (b *buffer) write(ts int64, value schema.Float) (*buffer, error) { } b.data = append(b.data, value) + b.updateStats(value) return b, nil } diff --git a/pkg/metricstore/stats_test.go b/pkg/metricstore/stats_test.go index f437f0bd..f19d6853 100644 --- a/pkg/metricstore/stats_test.go +++ b/pkg/metricstore/stats_test.go @@ -8,6 +8,8 @@ package metricstore import ( "math" "testing" + + "github.com/ClusterCockpit/cc-lib/v2/schema" ) func TestNewBufferStatsInitialized(t *testing.T) { @@ -28,3 +30,57 @@ func TestNewBufferStatsInitialized(t *testing.T) { t.Error("statsValid = false, want true for a fresh empty buffer") } } + +func TestWriteUpdatesRunningStats(t *testing.T) { + b := newBuffer(100, 10) + b.write(100, schema.Float(3.0)) + b.write(110, schema.Float(1.0)) + b.write(120, schema.Float(5.0)) + + if b.statSamples != 3 { + t.Errorf("statSamples = %d, want 3", b.statSamples) + } + if b.statSum != 9.0 { + t.Errorf("statSum = %v, want 9.0", b.statSum) + } + if b.statMin != 1.0 { + t.Errorf("statMin = %v, want 1.0", b.statMin) + } + if b.statMax != 5.0 { + t.Errorf("statMax = %v, want 5.0", b.statMax) + } + if !b.statsValid { + t.Error("statsValid = false, want true after appends only") + } +} + +func TestWriteNaNExcludedFromStats(t *testing.T) { + b := newBuffer(100, 10) + b.write(100, schema.Float(2.0)) + b.write(110, schema.NaN) // explicit NaN + b.write(130, schema.Float(4.0)) // leaves a NaN-filled gap at t=120 + + if b.statSamples != 2 { + t.Errorf("statSamples = %d, want 2 (NaN and gap excluded)", b.statSamples) + } + if b.statSum != 6.0 { + t.Errorf("statSum = %v, want 6.0", b.statSum) + } + if b.statMin != 2.0 || b.statMax != 4.0 { + t.Errorf("statMin/statMax = %v/%v, want 2.0/4.0", b.statMin, b.statMax) + } +} + +func TestWriteOverwriteInvalidatesStats(t *testing.T) { + b := newBuffer(100, 10) + b.write(100, schema.Float(1.0)) + b.write(110, schema.Float(2.0)) + if !b.statsValid { + t.Fatal("precondition: statsValid should be true after appends") + } + + b.write(100, schema.Float(99.0)) // overwrite existing index + if b.statsValid { + t.Error("statsValid = true, want false after overwrite") + } +}