mirror of
https://github.com/ClusterCockpit/cc-backend
synced 2026-08-07 22:07:14 +02:00
feat(metricstore): prune empty node levels in retention free path
This commit is contained in:
@@ -535,88 +535,13 @@ func Free(ms *MemoryStore, t time.Time) (int, error) {
|
|||||||
case 0:
|
case 0:
|
||||||
return ms.Free(nil, t.Unix())
|
return ms.Free(nil, t.Unix())
|
||||||
|
|
||||||
// Else formulate selectors, exclude those from the map
|
// Else free every cluster/node except the used ones, pruning node Levels
|
||||||
// and free the rest of the selectors
|
// that become empty in the same locked traversal.
|
||||||
default:
|
default:
|
||||||
selectors := GetSelectors(ms, excludeSelectors)
|
return ms.root.freeExcludingUsed(t.Unix(), excludeSelectors)
|
||||||
return FreeSelected(ms, selectors, t)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// FreeSelected frees buffers for specific selectors while preserving others.
|
|
||||||
//
|
|
||||||
// This function is used when we want to retain some specific nodes beyond the retention time.
|
|
||||||
// It iterates through the provided selectors and frees their associated buffers.
|
|
||||||
//
|
|
||||||
// Parameters:
|
|
||||||
// - ms: The MemoryStore instance
|
|
||||||
// - selectors: List of selector paths to free (e.g., [["cluster1", "node1"], ["cluster2", "node2"]])
|
|
||||||
// - t: Time threshold for freeing buffers
|
|
||||||
//
|
|
||||||
// Returns the total number of buffers freed and any error encountered.
|
|
||||||
func FreeSelected(ms *MemoryStore, selectors [][]string, t time.Time) (int, error) {
|
|
||||||
freed := 0
|
|
||||||
|
|
||||||
for _, selector := range selectors {
|
|
||||||
|
|
||||||
freedBuffers, err := ms.Free(selector, t.Unix())
|
|
||||||
if err != nil {
|
|
||||||
cclog.Errorf("error while freeing selected buffers: %#v", err)
|
|
||||||
}
|
|
||||||
freed += freedBuffers
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
return freed, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetSelectors returns all selectors at depth 2 (cluster/node level) that are NOT in the exclusion map.
|
|
||||||
//
|
|
||||||
// This function generates a list of selectors whose buffers should be freed by excluding
|
|
||||||
// selectors that correspond to nodes currently in use by running jobs.
|
|
||||||
//
|
|
||||||
// Parameters:
|
|
||||||
// - ms: The MemoryStore instance
|
|
||||||
// - excludeSelectors: Map of cluster names to node hostnames that should NOT be freed
|
|
||||||
//
|
|
||||||
// Returns a list of selectors ([]string paths) that can be safely freed.
|
|
||||||
//
|
|
||||||
// Example:
|
|
||||||
//
|
|
||||||
// If the tree has paths ["emmy", "node001"] and ["emmy", "node002"],
|
|
||||||
// and excludeSelectors contains {"emmy": ["node001"]},
|
|
||||||
// then only [["emmy", "node002"]] is returned.
|
|
||||||
func GetSelectors(ms *MemoryStore, excludeSelectors map[string][]string) [][]string {
|
|
||||||
allSelectors := ms.GetPaths(2)
|
|
||||||
|
|
||||||
filteredSelectors := make([][]string, 0, len(allSelectors))
|
|
||||||
|
|
||||||
for _, path := range allSelectors {
|
|
||||||
if len(path) < 2 {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
key := path[0] // The "Key" (Level 1)
|
|
||||||
value := path[1] // The "Value" (Level 2)
|
|
||||||
|
|
||||||
exclude := false
|
|
||||||
|
|
||||||
// Check if the key exists in our exclusion map
|
|
||||||
if excludedValues, exists := excludeSelectors[key]; exists {
|
|
||||||
// The key exists, now check if the specific value is in the exclusion list
|
|
||||||
if slices.Contains(excludedValues, value) {
|
|
||||||
exclude = true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if !exclude {
|
|
||||||
filteredSelectors = append(filteredSelectors, path)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return filteredSelectors
|
|
||||||
}
|
|
||||||
|
|
||||||
// isNodeUsed reports whether cluster/host appears in the used-nodes map
|
// isNodeUsed reports whether cluster/host appears in the used-nodes map
|
||||||
// returned by NodeProvider.GetUsedNodes. Host lists are sorted per the
|
// returned by NodeProvider.GetUsedNodes. Host lists are sorted per the
|
||||||
// interface contract, so lookup is a binary search. A nil map means no
|
// interface contract, so lookup is a binary search. A nil map means no
|
||||||
|
|||||||
@@ -7,7 +7,9 @@ package metricstore
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"slices"
|
"slices"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/ClusterCockpit/cc-lib/v2/schema"
|
"github.com/ClusterCockpit/cc-lib/v2/schema"
|
||||||
)
|
)
|
||||||
@@ -96,3 +98,59 @@ func TestFreeExcludingUsedPrunesSubLevelNode(t *testing.T) {
|
|||||||
t.Error("node must be pruned once all descendant buffers are freed")
|
t.Error("node must be pruned once all descendant buffers are freed")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestFreeViaProviderPrunesDeadNodes(t *testing.T) {
|
||||||
|
ms := newTestStore()
|
||||||
|
const freq int64 = 60
|
||||||
|
thr := time.Unix(100000, 0)
|
||||||
|
|
||||||
|
writeNode(ms, "fritz", "dead", nil, freq, 1000, 2000) // stale, not used -> pruned
|
||||||
|
writeNode(ms, "fritz", "busy", nil, freq, 1000, 2000) // stale, but used -> kept
|
||||||
|
writeNode(ms, "fritz", "live", nil, freq, 100000, 100600) // fresh -> kept
|
||||||
|
|
||||||
|
ms.SetNodeProvider(&fakeNodeProvider{nodes: map[string][]string{"fritz": {"busy"}}})
|
||||||
|
|
||||||
|
if _, err := Free(ms, thr); err != nil {
|
||||||
|
t.Fatalf("Free: %v", err)
|
||||||
|
}
|
||||||
|
if hasNode(ms, "fritz", "dead") {
|
||||||
|
t.Error("stale non-used node must be pruned")
|
||||||
|
}
|
||||||
|
if !hasNode(ms, "fritz", "busy") {
|
||||||
|
t.Error("used node must survive")
|
||||||
|
}
|
||||||
|
if !hasNode(ms, "fritz", "live") {
|
||||||
|
t.Error("fresh node must survive")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestFreeExcludingUsedConcurrentReadNoRace(t *testing.T) {
|
||||||
|
ms := newTestStore()
|
||||||
|
const freq, thr = int64(60), int64(100000)
|
||||||
|
for _, n := range []string{"a", "b", "c", "d"} {
|
||||||
|
writeNode(ms, "fritz", n, nil, freq, 1000, 2000)
|
||||||
|
}
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
stop := make(chan struct{})
|
||||||
|
for i := 0; i < 4; i++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-stop:
|
||||||
|
return
|
||||||
|
default:
|
||||||
|
_ = ms.ListChildren([]string{"fritz"})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := ms.root.freeExcludingUsed(thr, nil); err != nil {
|
||||||
|
t.Fatalf("freeExcludingUsed: %v", err)
|
||||||
|
}
|
||||||
|
close(stop)
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user