diff --git a/pkg/metricstore/checkpoint.go b/pkg/metricstore/checkpoint.go index c2b86415..700bf7dc 100644 --- a/pkg/metricstore/checkpoint.go +++ b/pkg/metricstore/checkpoint.go @@ -397,9 +397,25 @@ func enqueueCheckpointHosts(dir string, work chan<- [2]string) error { // FromCheckpoint loads checkpoint files from disk into memory in parallel. // -// Uses worker pool to load cluster/host combinations. Returns number of files -// loaded and any errors. +// Uses worker pool to load cluster/host combinations. Hosts that the +// NodeProvider reports as in use by running jobs load ALL their checkpoint +// files (a job that started before `from` needs its full history in memory); +// all other hosts load only files satisfying the `from` cutoff. A nil +// provider or a provider error falls back to cutoff-only loading with a +// warning. Returns number of files loaded and any errors. func (m *MemoryStore) FromCheckpoint(dir string, from int64) (int, error) { + var usedNodes map[string][]string + if m.nodeProvider != nil { + un, err := m.nodeProvider.GetUsedNodes(from) + if err != nil { + cclog.Warnf("[METRICSTORE]> GetUsedNodes failed, loading checkpoints with retention cutoff only: %s", err.Error()) + } else { + usedNodes = un + } + } else { + cclog.Warnf("[METRICSTORE]> no NodeProvider set, loading checkpoints with retention cutoff only") + } + var wg sync.WaitGroup work := make(chan [2]string, Keys.NumWorkers*4) n, errs := int32(0), int32(0) @@ -409,8 +425,14 @@ func (m *MemoryStore) FromCheckpoint(dir string, from int64) (int, error) { go func() { defer wg.Done() for host := range work { + // A host with a running job that started before the cutoff + // needs its full history in memory: load all its files. + loadFrom := from + if isNodeUsed(usedNodes, host[0], host[1]) { + loadFrom = 0 + } lvl := m.root.findLevelOrCreate(host[:], len(m.Metrics)) - nn, err := lvl.fromCheckpoint(m, filepath.Join(dir, host[0], host[1]), from) + nn, err := lvl.fromCheckpoint(m, filepath.Join(dir, host[0], host[1]), loadFrom) if err != nil { cclog.Errorf("[METRICSTORE]> error while loading checkpoints for %s/%s: %s", host[0], host[1], err.Error()) atomic.AddInt32(&errs, 1) diff --git a/pkg/metricstore/nodeprovider_test.go b/pkg/metricstore/nodeprovider_test.go index c8f0e5b0..7071b167 100644 --- a/pkg/metricstore/nodeprovider_test.go +++ b/pkg/metricstore/nodeprovider_test.go @@ -6,7 +6,14 @@ package metricstore import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" "testing" + + "github.com/ClusterCockpit/cc-lib/v2/schema" ) // fakeNodeProvider implements NodeProvider for tests. @@ -19,6 +26,8 @@ func (f *fakeNodeProvider) GetUsedNodes(ts int64) (map[string][]string, error) { return f.nodes, f.err } +var errTestProvider = errors.New("provider failure") + func TestIsNodeUsed(t *testing.T) { used := map[string][]string{"fritz": {"node001", "node003"}} @@ -44,3 +53,111 @@ func TestIsNodeUsed(t *testing.T) { t.Error("empty map must report false") } } + +// newTestStore builds a minimal MemoryStore without touching the singleton. +func newTestStore() *MemoryStore { + return &MemoryStore{ + Metrics: map[string]MetricConfig{ + "cpu_load": {Frequency: 60, offset: 0}, + }, + root: Level{ + metrics: make([]*buffer, 1), + children: make(map[string]*Level), + }, + } +} + +// writeTestCheckpoint writes a valid JSON checkpoint file .json into dir. +func writeTestCheckpoint(t *testing.T, dir string, ts int64) { + t.Helper() + cf := &CheckpointFile{ + From: ts, + To: ts + 120, + Metrics: map[string]*CheckpointMetrics{ + "cpu_load": {Frequency: 60, Start: ts, Data: []schema.Float{1.0, 2.0, 3.0}}, + }, + Children: make(map[string]*CheckpointFile), + } + data, err := json.Marshal(cf) + if err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, fmt.Sprintf("%d.json", ts)), data, 0o644); err != nil { + t.Fatal(err) + } +} + +// writeThreeCheckpointsPerHost creates ts 1000/2000/5000 for node001+node002. +func writeThreeCheckpointsPerHost(t *testing.T, dir string) { + t.Helper() + for _, host := range []string{"node001", "node002"} { + hostDir := filepath.Join(dir, "fritz", host) + writeTestCheckpoint(t, hostDir, 1000) + writeTestCheckpoint(t, hostDir, 2000) + writeTestCheckpoint(t, hostDir, 5000) + } +} + +func TestFromCheckpointLoadsAllFilesForUsedNodes(t *testing.T) { + oldWorkers := Keys.NumWorkers + Keys.NumWorkers = 2 + t.Cleanup(func() { Keys.NumWorkers = oldWorkers }) + + dir := t.TempDir() + writeThreeCheckpointsPerHost(t, dir) + + ms := newTestStore() + ms.SetNodeProvider(&fakeNodeProvider{nodes: map[string][]string{"fritz": {"node001"}}}) + + n, err := ms.FromCheckpoint(dir, 3000) + if err != nil { + t.Fatal(err) + } + // node001 (used): all 3 files. node002: bridge 2000.json + 5000.json = 2. + if n != 5 { + t.Fatalf("expected 5 files loaded, got %d", n) + } +} + +func TestFromCheckpointWithoutProviderKeepsCutoff(t *testing.T) { + oldWorkers := Keys.NumWorkers + Keys.NumWorkers = 2 + t.Cleanup(func() { Keys.NumWorkers = oldWorkers }) + + dir := t.TempDir() + writeThreeCheckpointsPerHost(t, dir) + + ms := newTestStore() + + n, err := ms.FromCheckpoint(dir, 3000) + if err != nil { + t.Fatal(err) + } + // 2 files per host, no provider set. + if n != 4 { + t.Fatalf("expected 4 files loaded, got %d", n) + } +} + +func TestFromCheckpointProviderErrorFallsBack(t *testing.T) { + oldWorkers := Keys.NumWorkers + Keys.NumWorkers = 2 + t.Cleanup(func() { Keys.NumWorkers = oldWorkers }) + + dir := t.TempDir() + writeThreeCheckpointsPerHost(t, dir) + + ms := newTestStore() + ms.SetNodeProvider(&fakeNodeProvider{err: errTestProvider}) + + n, err := ms.FromCheckpoint(dir, 3000) + if err != nil { + t.Fatal(err) + } + if n != 4 { + t.Fatalf("expected fallback to cutoff load (4 files), got %d", n) + } +}