From 162e835cf1100d0e8893bc953b4306da55b0f922 Mon Sep 17 00:00:00 2001 From: Aditya Ujeniya Date: Wed, 22 Jul 2026 11:18:06 +0200 Subject: [PATCH] feat(metricstore): add exclusion-aware free that prunes empty node levels --- pkg/metricstore/level.go | 71 +++++++++++++++++++++++++ pkg/metricstore/prune_test.go | 98 +++++++++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 pkg/metricstore/prune_test.go diff --git a/pkg/metricstore/level.go b/pkg/metricstore/level.go index 1d360564..ae060677 100644 --- a/pkg/metricstore/level.go +++ b/pkg/metricstore/level.go @@ -41,6 +41,7 @@ package metricstore import ( + "slices" "sync" "time" "unsafe" @@ -233,6 +234,76 @@ func (l *Level) freeAndCheckEmpty(t int64) (int, bool, error) { return n, empty, nil } +// freeExcludingUsed frees buffers older than t across the whole tree except for +// nodes listed in used (cluster name -> sorted hostnames), and deletes node +// Levels that become empty. The receiver must be the root level. Returns the +// total number of buffers freed. +// +// This is the exclusion-aware counterpart of freeAndCheckEmpty used by the +// retention path when a NodeProvider reports nodes in use by running jobs: +// used nodes are never freed, so never empty, so never pruned. Holding the root +// write lock for the full pass matches the existing root free (ms.Free(nil, t)). +func (l *Level) freeExcludingUsed(t int64, used map[string][]string) (int, error) { + l.lock.Lock() + defer l.lock.Unlock() + + total := 0 + for cluster, clusterLvl := range l.children { + if clusterLvl == nil { + continue + } + n, empty, err := clusterLvl.freeNodesExcludingUsed(t, used[cluster]) + total += n + if err != nil { + return total, err + } + if empty { + delete(l.children, cluster) + } + } + return total, nil +} + +// freeNodesExcludingUsed operates on a cluster level. For each node child whose +// name is not in usedHosts (sorted, per NodeProvider contract), it frees buffers +// older than t and deletes the node when it becomes empty. Returns the number of +// buffers freed and whether the cluster level itself is now empty. +func (l *Level) freeNodesExcludingUsed(t int64, usedHosts []string) (int, bool, error) { + l.lock.Lock() + defer l.lock.Unlock() + + n := 0 + for node, nodeLvl := range l.children { + if nodeLvl == nil { + continue + } + if _, found := slices.BinarySearch(usedHosts, node); found { + continue // used node: preserve entirely + } + m, empty, err := nodeLvl.freeAndCheckEmpty(t) + n += m + if err != nil { + return n, false, err + } + if empty { + delete(l.children, node) + } + } + + // Cluster is empty only if it has no children and no buffers of its own + // (inner levels may hold aggregated metrics). + empty := len(l.children) == 0 + if empty { + for _, b := range l.metrics { + if b != nil { + empty = false + break + } + } + } + return n, empty, nil +} + // forceFree removes the oldest buffer from each metric chain in the subtree. // // Unlike free(), which removes based on time threshold, this unconditionally removes diff --git a/pkg/metricstore/prune_test.go b/pkg/metricstore/prune_test.go new file mode 100644 index 00000000..31643753 --- /dev/null +++ b/pkg/metricstore/prune_test.go @@ -0,0 +1,98 @@ +// 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 ( + "slices" + "testing" + + "github.com/ClusterCockpit/cc-lib/v2/schema" +) + +// writeNode creates cluster/node (and optional deeper sub-levels) and fills +// every metric buffer with points from startTs to endTs (inclusive) at the +// store's metric frequency. subPath appends levels below the node (e.g. +// {"socket0"}); pass nil to write buffers directly on the node level. +func writeNode(ms *MemoryStore, cluster, node string, subPath []string, freq, startTs, endTs int64) { + selector := append([]string{cluster, node}, subPath...) + lvl := ms.root.findLevelOrCreate(selector, len(ms.Metrics)) + for i := range lvl.metrics { + lvl.metrics[i] = newBuffer(startTs, freq) + for ts := startTs; ts <= endTs; ts += freq { + lvl.metrics[i].write(ts, schema.Float(1)) + } + } +} + +// hasNode reports whether [cluster, node] still exists in the tree. +func hasNode(ms *MemoryStore, cluster, node string) bool { + return slices.Contains(ms.ListChildren([]string{cluster}), node) +} + +func TestFreeExcludingUsedPrunesDeadNode(t *testing.T) { + ms := newTestStore() // one metric "cpu_load", freq 60 + const freq, thr = int64(60), int64(100000) + + // dead: last data ~2000, well below threshold -> emptied -> pruned + writeNode(ms, "fritz", "dead", nil, freq, 1000, 2000) + // alive: data around/after threshold -> stays + writeNode(ms, "fritz", "alive", nil, freq, thr, thr+600) + + freed, err := ms.root.freeExcludingUsed(thr, nil) + if err != nil { + t.Fatalf("freeExcludingUsed: %v", err) + } + if freed == 0 { + t.Fatal("expected at least one buffer freed") + } + if hasNode(ms, "fritz", "dead") { + t.Error("dead node must be pruned from the tree") + } + if !hasNode(ms, "fritz", "alive") { + t.Error("alive node must be preserved") + } +} + +func TestFreeExcludingUsedPreservesUsedNode(t *testing.T) { + ms := newTestStore() + const freq, thr = int64(60), int64(100000) + + // both would be dead by timestamp, but "used" is excluded + writeNode(ms, "fritz", "used", nil, freq, 1000, 2000) + writeNode(ms, "fritz", "gone", nil, freq, 1000, 2000) + + used := map[string][]string{"fritz": {"used"}} // sorted hostnames + + if _, err := ms.root.freeExcludingUsed(thr, used); err != nil { + t.Fatalf("freeExcludingUsed: %v", err) + } + if !hasNode(ms, "fritz", "used") { + t.Error("used node must be preserved even when stale") + } + if hasNode(ms, "fritz", "gone") { + t.Error("non-used dead node must be pruned") + } + // used node still holds its buffer + lvl := ms.root.findLevel([]string{"fritz", "used"}) + if lvl == nil || lvl.metrics[0] == nil { + t.Error("used node must keep its buffers") + } +} + +func TestFreeExcludingUsedPrunesSubLevelNode(t *testing.T) { + ms := newTestStore() + const freq, thr = int64(60), int64(100000) + + // node holds no direct buffers; only a socket0 child, all stale + writeNode(ms, "fritz", "deep", []string{"socket0"}, freq, 1000, 2000) + + if _, err := ms.root.freeExcludingUsed(thr, nil); err != nil { + t.Fatalf("freeExcludingUsed: %v", err) + } + if hasNode(ms, "fritz", "deep") { + t.Error("node must be pruned once all descendant buffers are freed") + } +}