mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2026-07-22 06:50:38 +02:00
f83f8afa86
recomputeStats() wrote statSum/statSamples/statMin/statMax/statsValid from inside the stats() fast path, which runs under only a shared RLock via MemoryStore.Stats -> Level.findBuffers. Two concurrent queries on the same invalid buffer could both enter the fast path and race on those fields. Require statsValid in the fast-path guard instead of recomputing inline; invalid buffers now fall through to the existing point-by-point scan, which only reads b.data. recomputeStats() is unchanged and still used by checkpoint loadFile at single-threaded load time. Updated stats_test.go: overwritten buffers now stay statsValid=false after a query (documenting non-mutating reads); TestStatsMultiBufferChain and TestStatsFastPathThenPartialTail now recompute stats after building bare buffers so the fast-path cache fold is still covered; added TestStatsConcurrentQueriesNoRace (race-clean under -race for both valid and post-overwrite buffers) and TestStatsGappedChain (real inter-buffer gap correctly excluded from Samples/Min/Max/Avg). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
413 lines
12 KiB
Go
413 lines
12 KiB
Go
// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
|
|
// All rights reserved. This file is part of cc-backend.
|
|
// Use of this source code is governed by a MIT-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package metricstore
|
|
|
|
import (
|
|
"math"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/ClusterCockpit/cc-lib/v2/schema"
|
|
)
|
|
|
|
func TestNewBufferStatsInitialized(t *testing.T) {
|
|
b := newBuffer(100, 10)
|
|
if b.statSamples != 0 {
|
|
t.Errorf("statSamples = %d, want 0", b.statSamples)
|
|
}
|
|
if b.statSum != 0 {
|
|
t.Errorf("statSum = %v, want 0", b.statSum)
|
|
}
|
|
if b.statMin != math.MaxFloat32 {
|
|
t.Errorf("statMin = %v, want MaxFloat32 sentinel", b.statMin)
|
|
}
|
|
if b.statMax != -math.MaxFloat32 {
|
|
t.Errorf("statMax = %v, want -MaxFloat32 sentinel", b.statMax)
|
|
}
|
|
if !b.statsValid {
|
|
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")
|
|
}
|
|
}
|
|
|
|
func TestRecomputeStatsRebuildsAfterOverwrite(t *testing.T) {
|
|
b := newBuffer(100, 10)
|
|
b.write(100, schema.Float(1.0))
|
|
b.write(110, schema.Float(2.0))
|
|
b.write(100, schema.Float(99.0)) // overwrite -> statsValid false, min stale
|
|
|
|
b.recomputeStats()
|
|
|
|
if !b.statsValid {
|
|
t.Error("statsValid = false, want true after recompute")
|
|
}
|
|
if b.statSamples != 2 {
|
|
t.Errorf("statSamples = %d, want 2", b.statSamples)
|
|
}
|
|
if b.statSum != 101.0 {
|
|
t.Errorf("statSum = %v, want 101.0", b.statSum)
|
|
}
|
|
if b.statMin != 2.0 {
|
|
t.Errorf("statMin = %v, want 2.0 (1.0 was overwritten)", b.statMin)
|
|
}
|
|
if b.statMax != 99.0 {
|
|
t.Errorf("statMax = %v, want 99.0", b.statMax)
|
|
}
|
|
}
|
|
|
|
func TestRecomputeStatsEmptyBuffer(t *testing.T) {
|
|
b := newBuffer(100, 10)
|
|
b.recomputeStats()
|
|
if b.statSamples != 0 || b.statSum != 0 {
|
|
t.Errorf("empty buffer stats = samples %d sum %v, want 0/0", b.statSamples, b.statSum)
|
|
}
|
|
if b.statMin != math.MaxFloat32 || b.statMax != -math.MaxFloat32 {
|
|
t.Errorf("empty buffer min/max = %v/%v, want sentinels", b.statMin, b.statMax)
|
|
}
|
|
}
|
|
|
|
func TestStatsFullBufferMatchesScan(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))
|
|
// Range fully covers the buffer (from before firstWrite, to at/after end).
|
|
s, _, _, err := b.stats(100, 130)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
if s.Samples != 3 {
|
|
t.Errorf("Samples = %d, want 3", s.Samples)
|
|
}
|
|
if s.Min != 1.0 || s.Max != 5.0 {
|
|
t.Errorf("Min/Max = %v/%v, want 1.0/5.0", s.Min, s.Max)
|
|
}
|
|
if s.Avg != 3.0 {
|
|
t.Errorf("Avg = %v, want 3.0", s.Avg)
|
|
}
|
|
}
|
|
|
|
func TestStatsPartialRangeScans(t *testing.T) {
|
|
b := newBuffer(100, 10)
|
|
b.write(100, schema.Float(3.0)) // t=100
|
|
b.write(110, schema.Float(1.0)) // t=110
|
|
b.write(120, schema.Float(5.0)) // t=120
|
|
// Partial range: only t=110 and t=120 (from excludes t=100).
|
|
s, _, _, err := b.stats(110, 130)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
if s.Samples != 2 {
|
|
t.Errorf("Samples = %d, want 2", s.Samples)
|
|
}
|
|
if s.Min != 1.0 || s.Max != 5.0 {
|
|
t.Errorf("Min/Max = %v/%v, want 1.0/5.0", s.Min, s.Max)
|
|
}
|
|
}
|
|
|
|
func TestStatsOverwriteThenFullQuery(t *testing.T) {
|
|
b := newBuffer(100, 10)
|
|
b.write(100, schema.Float(1.0))
|
|
b.write(110, schema.Float(2.0))
|
|
b.write(120, schema.Float(3.0))
|
|
b.write(100, schema.Float(99.0)) // overwrite the running min -> statsValid false
|
|
|
|
s, _, _, err := b.stats(100, 130)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
if s.Min != 2.0 {
|
|
t.Errorf("Min = %v, want 2.0 (scanned after overwrite)", s.Min)
|
|
}
|
|
if s.Max != 99.0 {
|
|
t.Errorf("Max = %v, want 99.0", s.Max)
|
|
}
|
|
if s.Samples != 3 {
|
|
t.Errorf("Samples = %d, want 3", s.Samples)
|
|
}
|
|
if b.statsValid {
|
|
t.Error("statsValid should still be false: the read path must not mutate buffer state")
|
|
}
|
|
}
|
|
|
|
func TestStatsMultiBufferChain(t *testing.T) {
|
|
// Two full buffers of cap=3 (interior buffers are always full).
|
|
// b1: t=100,110,120 = 1,2,3 ; b2: t=130,140,150 = 4,5,6.
|
|
b1 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 95}
|
|
b1.data = append(b1.data, 1.0, 2.0, 3.0)
|
|
b2 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 125}
|
|
b2.data = append(b2.data, 4.0, 5.0, 6.0)
|
|
b2.prev = b1
|
|
b1.next = b2
|
|
// b1/b2 built bare: statsValid is false (zero value). Recompute here to mirror
|
|
// checkpoint-loaded valid buffers, so the fast-path cache fold is exercised.
|
|
b1.recomputeStats()
|
|
b2.recomputeStats()
|
|
|
|
s, _, _, err := b2.stats(100, 160)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
if s.Samples != 6 {
|
|
t.Errorf("Samples = %d, want 6", s.Samples)
|
|
}
|
|
if s.Min != 1.0 || s.Max != 6.0 {
|
|
t.Errorf("Min/Max = %v/%v, want 1.0/6.0", s.Min, s.Max)
|
|
}
|
|
if s.Avg != 3.5 {
|
|
t.Errorf("Avg = %v, want 3.5", s.Avg)
|
|
}
|
|
}
|
|
|
|
func TestStatsFastPathThenPartialTail(t *testing.T) {
|
|
// Regression test: three buffers (all frequency 10) where the query fast-paths
|
|
// earlier fully-covered buffers, then only PARTIALLY covers the last buffer.
|
|
// firstWrite = start + frequency/2, so with start=95 the first value is at t=100.
|
|
// b1: 1,2,3 at t=100,110,120 (end=130, fully in [100,170) -> fast path folds all 3)
|
|
// b2: 4,5,6 at t=130,140,150 (end=160, fully in [100,170) -> fast path folds all 3)
|
|
// b3: 7,8,9 at t=160,170,180 (end=190; not fully covered -> scans; only t=160 < 170)
|
|
// Query [100, 170) includes t=100,110,120,130,140,150,160 = values 1..7 (7 points).
|
|
// t=170 is excluded since 170 >= to. Expected: Samples=7, Sum=28, Min=1, Max=7, Avg=4.0.
|
|
// Buffers built bare: statsValid is false (zero value). Recompute on b1/b2 to mirror
|
|
// checkpoint-loaded valid buffers so the fast path fires for them; b3 is only
|
|
// partially covered by the query so it still scans regardless of validity,
|
|
// giving intended mixed fast-path-then-scan coverage.
|
|
|
|
b1 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 95}
|
|
b1.data = append(b1.data, 1.0, 2.0, 3.0)
|
|
|
|
b2 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 125}
|
|
b2.data = append(b2.data, 4.0, 5.0, 6.0)
|
|
|
|
b3 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 155}
|
|
b3.data = append(b3.data, 7.0, 8.0, 9.0)
|
|
|
|
b1.next = b2
|
|
b2.prev = b1
|
|
b2.next = b3
|
|
b3.prev = b2
|
|
|
|
b1.recomputeStats()
|
|
b2.recomputeStats()
|
|
b3.recomputeStats()
|
|
|
|
s, _, _, err := b3.stats(100, 170)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
if s.Samples != 7 {
|
|
t.Errorf("Samples = %d, want 7", s.Samples)
|
|
}
|
|
if s.Min != 1.0 || s.Max != 7.0 {
|
|
t.Errorf("Min/Max = %v/%v, want 1.0/7.0", s.Min, s.Max)
|
|
}
|
|
if s.Avg != 4.0 {
|
|
t.Errorf("Avg = %v, want 4.0", s.Avg)
|
|
}
|
|
}
|
|
|
|
func TestCheckpointLoadedBufferStatsEager(t *testing.T) {
|
|
// Mirror how loadFile constructs a buffer: bare struct, data assigned
|
|
// directly with cap==len, then stats computed eagerly.
|
|
data := []schema.Float{2.0, 4.0, 6.0}
|
|
n := len(data)
|
|
b := &buffer{
|
|
frequency: 10,
|
|
start: 95,
|
|
data: data[0:n:n],
|
|
archived: true,
|
|
}
|
|
b.recomputeStats() // the eager call loadFile will perform
|
|
|
|
if !b.statsValid {
|
|
t.Error("statsValid = false, want true after eager recompute at load")
|
|
}
|
|
if b.statSamples != 3 {
|
|
t.Errorf("statSamples = %d, want 3", b.statSamples)
|
|
}
|
|
if b.statMin != 2.0 || b.statMax != 6.0 {
|
|
t.Errorf("statMin/statMax = %v/%v, want 2.0/6.0", b.statMin, b.statMax)
|
|
}
|
|
if b.statSum != 12.0 {
|
|
t.Errorf("statSum = %v, want 12.0", b.statSum)
|
|
}
|
|
}
|
|
|
|
// TestStatsConcurrentQueriesNoRace guards against the data race fixed by making
|
|
// the stats() read path non-mutating: previously an invalid buffer's stats()
|
|
// call would run recomputeStats() (a write) under only the caller's shared
|
|
// RLock (see MemoryStore.Stats -> Level.findBuffers), so concurrent readers on
|
|
// the same buffer could race on statSum/statSamples/statMin/statMax/statsValid.
|
|
// Run with -race; both subtests must be race-clean.
|
|
func TestStatsConcurrentQueriesNoRace(t *testing.T) {
|
|
const goroutines = 50
|
|
|
|
t.Run("valid buffer", func(t *testing.T) {
|
|
b := newBuffer(100, 10)
|
|
vals := []schema.Float{3.0, 1.0, 5.0, 2.0, 4.0}
|
|
for i, v := range vals {
|
|
ts := int64(100 + i*10)
|
|
if _, err := b.write(ts, v); err != nil {
|
|
t.Fatalf("write(%d) error = %v", ts, err)
|
|
}
|
|
}
|
|
if !b.statsValid {
|
|
t.Fatal("precondition: statsValid should be true for an append-only buffer")
|
|
}
|
|
|
|
from, to := int64(100), int64(150)
|
|
expected, _, _, err := b.stats(from, to)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < goroutines; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
s, _, _, err := b.stats(from, to)
|
|
if err != nil {
|
|
t.Errorf("stats() error = %v", err)
|
|
return
|
|
}
|
|
if s != expected {
|
|
t.Errorf("stats() = %+v, want %+v", s, expected)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
})
|
|
|
|
t.Run("invalid buffer after overwrite", func(t *testing.T) {
|
|
b := newBuffer(100, 10)
|
|
if _, err := b.write(100, schema.Float(1.0)); err != nil {
|
|
t.Fatalf("write error = %v", err)
|
|
}
|
|
if _, err := b.write(110, schema.Float(2.0)); err != nil {
|
|
t.Fatalf("write error = %v", err)
|
|
}
|
|
if _, err := b.write(120, schema.Float(3.0)); err != nil {
|
|
t.Fatalf("write error = %v", err)
|
|
}
|
|
if _, err := b.write(100, schema.Float(99.0)); err != nil { // overwrite -> statsValid false
|
|
t.Fatalf("write error = %v", err)
|
|
}
|
|
if b.statsValid {
|
|
t.Fatal("precondition: statsValid should be false after overwrite")
|
|
}
|
|
|
|
from, to := int64(100), int64(130)
|
|
expected, _, _, err := b.stats(from, to)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < goroutines; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
s, _, _, err := b.stats(from, to)
|
|
if err != nil {
|
|
t.Errorf("stats() error = %v", err)
|
|
return
|
|
}
|
|
if s != expected {
|
|
t.Errorf("stats() = %+v, want %+v", s, expected)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
})
|
|
}
|
|
|
|
// TestStatsGappedChain covers a multi-buffer chain with a real time gap between
|
|
// buffers: b1's last real point is at t=120, and b2's first real point is at
|
|
// t=170, so t=130,140,150,160 have no data anywhere (unlike an in-buffer,
|
|
// NaN-filled gap). The query spans the gap; only the real points on either
|
|
// side should be counted.
|
|
func TestStatsGappedChain(t *testing.T) {
|
|
b1 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 95}
|
|
b1.data = append(b1.data, 1.0, 2.0, 3.0) // t=100,110,120 (end=130)
|
|
|
|
b2 := &buffer{data: make([]schema.Float, 0, 3), frequency: 10, start: 165}
|
|
b2.data = append(b2.data, 4.0, 5.0, 6.0) // t=170,180,190 (end=200)
|
|
|
|
b1.next = b2
|
|
b2.prev = b1
|
|
|
|
b1.recomputeStats()
|
|
b2.recomputeStats()
|
|
|
|
s, _, _, err := b2.stats(100, 200)
|
|
if err != nil {
|
|
t.Fatalf("stats() error = %v", err)
|
|
}
|
|
if s.Samples != 6 {
|
|
t.Errorf("Samples = %d, want 6 (gap contributes no samples)", s.Samples)
|
|
}
|
|
if s.Min != 1.0 || s.Max != 6.0 {
|
|
t.Errorf("Min/Max = %v/%v, want 1.0/6.0", s.Min, s.Max)
|
|
}
|
|
if s.Avg != 3.5 {
|
|
t.Errorf("Avg = %v, want 3.5", s.Avg)
|
|
}
|
|
}
|