mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2026-08-07 22:07:14 +02:00
feat(metricstore): add exclusion-aware free that prunes empty node levels
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user