mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2026-07-22 06:50:38 +02:00
feat(metricstore): maintain running buffer stats on write
This commit is contained in:
@@ -207,6 +207,23 @@ func newBuffer(ts, freq int64) *buffer {
|
|||||||
return b
|
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.
|
// write appends a timestamped value to the buffer chain.
|
||||||
//
|
//
|
||||||
// Returns the head buffer (which may be newly created if capacity was reached).
|
// 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
|
// Overwriting value or writing value from past
|
||||||
if idx < len(b.data) {
|
if idx < len(b.data) {
|
||||||
b.data[idx] = value
|
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
|
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.data = append(b.data, value)
|
||||||
|
b.updateStats(value)
|
||||||
return b, nil
|
return b, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -8,6 +8,8 @@ package metricstore
|
|||||||
import (
|
import (
|
||||||
"math"
|
"math"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/ClusterCockpit/cc-lib/v2/schema"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestNewBufferStatsInitialized(t *testing.T) {
|
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")
|
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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user