modify(metricstore): more verbose logging to DataNonAlignedIssue

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)
+//	findBuffers([]Selector{{Any: true}, {String: "cpu0"}}, metricOffset, callback, nil)
-func (l *Level) findBuffers(selector util.Selector, offset int, f func(b *buffer) error) error {
+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 {
+		for key, lvl := range l.children {
-			err := lvl.findBuffers(nil, offset, f)
+			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 {
+		for key, lvl := range l.children {
-			if err := lvl.findBuffers(selector[1:], offset, f); err != nil {
+			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
+}

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",
+				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, n, minfo.Frequency, dataFrom, dataTo, cfrom, cto, missingfront))
+					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",
+				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, n, minfo.Frequency, dataFrom, dataTo, cfrom, cto, missingback))
+					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]",
+				cclog.Warnf("%s", fmt.Errorf("%w: metric=%s node=%s buf#%d expected[%d,%d] actual[%d,%d]",
-					ErrDataDoesNotAlignDataLenMismatch, metric, n, dataFrom, dataTo, cfrom, cto))
+					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

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]",
+			return fmt.Errorf("%w: metric=%s node=%s buf#%d expected[%d,%d] actual[%d,%d]",
-				ErrDataDoesNotAlignMissingFront, metric, n, from, to, cfrom, cto)
+				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]",
+			return fmt.Errorf("%w: metric=%s node=%s buf#%d expected[%d,%d] actual[%d,%d]",
-				ErrDataDoesNotAlignMissingBack, metric, n, from, to, cfrom, cto)
+				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
 	}