diff --git a/pkg/metricstore/level.go b/pkg/metricstore/level.go index a8b2515b..80df92aa 100644 --- a/pkg/metricstore/level.go +++ b/pkg/metricstore/level.go @@ -381,7 +381,10 @@ func (l *Level) findLevel(selector []string) *Level { // Parameters: // - selector: Pattern to match (consumed recursively) // - offset: Metric index in metrics slice (from MetricConfig.offset) -// - f: Callback invoked on each matching buffer +// - f: Callback invoked on each matching buffer; path holds the matched +// level keys (e.g. ["cluster","node001","cpu0"]) so callers can +// identify which node/component the buffer belongs to +// - path: Accumulated level keys from the root; pass nil at the top level // // Returns: // - error: First error returned by callback, or nil if all succeeded @@ -389,19 +392,19 @@ func (l *Level) findLevel(selector []string) *Level { // Example: // // // Find all cpu0 buffers across all hosts: -// findBuffers([]Selector{{Any: true}, {String: "cpu0"}}, metricOffset, callback) -func (l *Level) findBuffers(selector util.Selector, offset int, f func(b *buffer) error) error { +// findBuffers([]Selector{{Any: true}, {String: "cpu0"}}, metricOffset, callback, nil) +func (l *Level) findBuffers(selector util.Selector, offset int, f func(b *buffer, path []string) error, path []string) error { l.lock.RLock() defer l.lock.RUnlock() if len(selector) == 0 { b := l.metrics[offset] if b != nil { - return f(b) + return f(b, path) } - for _, lvl := range l.children { - err := lvl.findBuffers(nil, offset, f) + for key, lvl := range l.children { + err := lvl.findBuffers(nil, offset, f, appendPath(path, key)) if err != nil { return err } @@ -413,7 +416,7 @@ func (l *Level) findBuffers(selector util.Selector, offset int, f func(b *buffer if len(sel.String) != 0 && l.children != nil { lvl, ok := l.children[sel.String] if ok { - err := lvl.findBuffers(selector[1:], offset, f) + err := lvl.findBuffers(selector[1:], offset, f, appendPath(path, sel.String)) if err != nil { return err } @@ -425,7 +428,7 @@ func (l *Level) findBuffers(selector util.Selector, offset int, f func(b *buffer for _, key := range sel.Group { lvl, ok := l.children[key] if ok { - err := lvl.findBuffers(selector[1:], offset, f) + err := lvl.findBuffers(selector[1:], offset, f, appendPath(path, key)) if err != nil { return err } @@ -435,8 +438,8 @@ func (l *Level) findBuffers(selector util.Selector, offset int, f func(b *buffer } if sel.Any && l.children != nil { - for _, lvl := range l.children { - if err := lvl.findBuffers(selector[1:], offset, f); err != nil { + for key, lvl := range l.children { + if err := lvl.findBuffers(selector[1:], offset, f, appendPath(path, key)); err != nil { return err } } @@ -445,3 +448,13 @@ func (l *Level) findBuffers(selector util.Selector, offset int, f func(b *buffer return nil } + +// appendPath returns a new slice holding path followed by key. It copies rather +// than appending in place so sibling recursions never share/overwrite backing +// storage while the tree is walked concurrently. +func appendPath(path []string, key string) []string { + np := make([]string, len(path)+1) + copy(np, path) + np[len(path)] = key + return np +} diff --git a/pkg/metricstore/metricstore.go b/pkg/metricstore/metricstore.go index d677dfee..7afb7220 100644 --- a/pkg/metricstore/metricstore.go +++ b/pkg/metricstore/metricstore.go @@ -28,6 +28,7 @@ import ( "runtime" "runtime/debug" "slices" + "strings" "sync" "time" @@ -631,7 +632,7 @@ func (m *MemoryStore) Read(selector util.Selector, metric string, from, to, reso n, data := 0, make([]schema.Float, (to-from)/minfo.Frequency+1) var dataFrom, dataTo int64 - err := m.root.findBuffers(selector, minfo.offset, func(b *buffer) error { + err := m.root.findBuffers(selector, minfo.offset, func(b *buffer, path []string) error { cdata, cfrom, cto, err := b.read(from, to, data, false) if err != nil { return err @@ -640,24 +641,27 @@ func (m *MemoryStore) Read(selector util.Selector, metric string, from, to, reso if n == 0 { dataFrom, dataTo = cfrom, cto } else if cfrom != dataFrom || cto != dataTo { + node := strings.Join(path, "/") missingfront, missingback := int((dataFrom-cfrom)/minfo.Frequency), int((dataTo-cto)/minfo.Frequency) switch { case missingfront != 0: - cclog.Warnf("%s", fmt.Errorf("%w: metric=%s buf#%d freq=%d ref[%d,%d] this[%d,%d] missingfront=%d pts", - ErrDataDoesNotAlignMissingFront, metric, n, minfo.Frequency, dataFrom, dataTo, cfrom, cto, missingfront)) + cclog.Warnf("%s", fmt.Errorf("%w: metric=%s node=%s buf#%d freq=%d expected[%d,%d] actual[%d,%d] missingfront=%d pts", + ErrDataDoesNotAlignMissingFront, metric, node, n, minfo.Frequency, dataFrom, dataTo, cfrom, cto, missingfront)) case missingback != 0: - cclog.Warnf("%s", fmt.Errorf("%w: metric=%s buf#%d freq=%d ref[%d,%d] this[%d,%d] missingback=%d pts", - ErrDataDoesNotAlignMissingBack, metric, n, minfo.Frequency, dataFrom, dataTo, cfrom, cto, missingback)) + cclog.Warnf("%s", fmt.Errorf("%w: metric=%s node=%s buf#%d freq=%d expected[%d,%d] actual[%d,%d] missingback=%d pts", + ErrDataDoesNotAlignMissingBack, metric, node, n, minfo.Frequency, dataFrom, dataTo, cfrom, cto, missingback)) default: - cclog.Warnf("%s", fmt.Errorf("%w: metric=%s buf#%d ref[%d,%d] this[%d,%d]", - ErrDataDoesNotAlignDataLenMismatch, metric, n, dataFrom, dataTo, cfrom, cto)) + cclog.Warnf("%s", fmt.Errorf("%w: metric=%s node=%s buf#%d expected[%d,%d] actual[%d,%d]", + ErrDataDoesNotAlignDataLenMismatch, metric, node, n, dataFrom, dataTo, cfrom, cto)) } } + fmt.Printf("Gather data - cto: %d, cfrom: %d, dto: %d, dfrom: %d\n", cto, cfrom, dataTo, dataFrom) + data = cdata n += 1 return nil - }) + }, nil) if err != nil { return nil, 0, 0, 0, err diff --git a/pkg/metricstore/stats.go b/pkg/metricstore/stats.go index 4cc3d861..f873322d 100644 --- a/pkg/metricstore/stats.go +++ b/pkg/metricstore/stats.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "math" + "strings" "github.com/ClusterCockpit/cc-lib/v2/schema" "github.com/ClusterCockpit/cc-lib/v2/util" @@ -134,7 +135,7 @@ func (m *MemoryStore) Stats(selector util.Selector, metric string, from, to int6 n, samples := 0, 0 avg, min, max := schema.Float(0), math.MaxFloat32, -math.MaxFloat32 - err := m.root.findBuffers(selector, minfo.offset, func(b *buffer) error { + err := m.root.findBuffers(selector, minfo.offset, func(b *buffer, path []string) error { stats, cfrom, cto, err := b.stats(from, to) if err != nil { return err @@ -143,11 +144,11 @@ func (m *MemoryStore) Stats(selector util.Selector, metric string, from, to int6 if n == 0 { from, to = cfrom, cto } else if from != cfrom { - return fmt.Errorf("%w: metric=%s buf#%d want[%d,%d] got[%d,%d]", - ErrDataDoesNotAlignMissingFront, metric, n, from, to, cfrom, cto) + return fmt.Errorf("%w: metric=%s node=%s buf#%d expected[%d,%d] actual[%d,%d]", + ErrDataDoesNotAlignMissingFront, metric, strings.Join(path, "/"), n, from, to, cfrom, cto) } else if to != cto { - return fmt.Errorf("%w: metric=%s buf#%d want[%d,%d] got[%d,%d]", - ErrDataDoesNotAlignMissingBack, metric, n, from, to, cfrom, cto) + return fmt.Errorf("%w: metric=%s node=%s buf#%d expected[%d,%d] actual[%d,%d]", + ErrDataDoesNotAlignMissingBack, metric, strings.Join(path, "/"), n, from, to, cfrom, cto) } samples += stats.Samples @@ -156,7 +157,7 @@ func (m *MemoryStore) Stats(selector util.Selector, metric string, from, to int6 max = math.Max(max, float64(stats.Max)) n += 1 return nil - }) + }, nil) if err != nil { return nil, 0, 0, err }