ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2026-08-12 18:37:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

Simplify cache, calculate with expr

Browse Source
This commit is contained in:
Thomas Gruber
2026-07-31 16:31:04 +02:00
parent 789ba7f4a5
commit 4c1f810f53
7 changed files with 744 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/metricRouter/metricCache.go
+111 -96
View File
@@ -9,6 +9,7 @@ package metricRouter
import (
"fmt"
"math"
"sync"
"time"
@@ -19,57 +20,104 @@ import (
mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
)
type metricCachePeriod struct {
startstamp time.Time
stopstamp time.Time
numMetrics int
sizeMetrics int
metrics []lp.CCMessage
type ccCache struct {
periodIdx int
maxPeriods int
periods [][]lp.CCMessage
periodTimes []struct {
starttime time.Time
endtime time.Time
}
}
type CCCache interface {
Init(numPeriods int) error
Add(msg lp.CCMessage) error
GetPeriod(offset int) (time.Time, time.Time, []lp.CCMessage)
GetAll() []lp.CCMessage
NewPeriod()
}
func (c *ccCache) Init(numPeriods int) error {
c.maxPeriods = numPeriods
c.periodIdx = 0
c.periods = make([][]lp.CCMessage, c.maxPeriods)
c.periodTimes = make([]struct {
starttime time.Time
endtime time.Time
}, c.maxPeriods)
return nil
}
func (c *ccCache) NewPeriod() {
c.periodTimes[c.periodIdx].endtime = time.Now()
c.periodIdx = (c.periodIdx + 1) % c.maxPeriods
fmt.Printf("New period index %d\n", c.periodIdx)
c.periods[c.periodIdx] = c.periods[c.periodIdx][:0]
c.periodTimes[c.periodIdx].starttime = time.Now()
c.periodTimes[c.periodIdx].endtime = c.periodTimes[c.periodIdx].starttime
}
func (c *ccCache) Add(msg lp.CCMessage) error {
c.periods[c.periodIdx] = append(c.periods[c.periodIdx], msg)
c.periodTimes[c.periodIdx].endtime = msg.Time()
return nil
}
func (c *ccCache) GetPeriod(offset int) (time.Time, time.Time, []lp.CCMessage) {
if offset > c.maxPeriods {
offset = offset % c.maxPeriods
}
out := make([]lp.CCMessage, 0)
poff := int(math.Abs(float64(c.periodIdx - offset)))
out = append(out, c.periods[poff%c.maxPeriods]...)
return c.periodTimes[poff%c.maxPeriods].starttime, c.periodTimes[poff%c.maxPeriods].endtime, out
}
func (c *ccCache) GetAll() []lp.CCMessage {
out := make([]lp.CCMessage, 0)
for _, data := range c.periods {
out = append(out, data...)
}
return out
}
// Metric cache data structure
type metricCache struct {
numPeriods int
curPeriod int
lock sync.Mutex
intervals []*metricCachePeriod
cache CCCache
wg *sync.WaitGroup
ticker mct.MultiChanTicker
tickchan chan time.Time
done chan bool
output chan lp.CCMessage
aggEngine agg.MetricAggregator
numPeriods int
started bool
}
type MetricCache interface {
Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) error
Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, interval time.Duration, numPeriods int) error
Start()
Add(metric lp.CCMessage)
GetPeriod(index int) (time.Time, time.Time, []lp.CCMessage)
AddAggregation(name, function, condition string, tags, meta map[string]string) error
DeleteAggregation(name string) error
Close()
}
func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) error {
func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, interval time.Duration, numPeriods int) error {
var err error
c.done = make(chan bool)
c.wg = wg
c.ticker = ticker
c.numPeriods = numPeriods
c.started = false
c.cache = new(ccCache)
c.output = output
c.intervals = make([]*metricCachePeriod, 0)
for i := 0; i < c.numPeriods+1; i++ {
p := new(metricCachePeriod)
p.numMetrics = 0
p.sizeMetrics = 0
p.metrics = make([]lp.CCMessage, 0)
c.intervals = append(c.intervals, p)
}
// Create a new aggregation engine. No separate goroutine at the moment
// The code is executed by the MetricCache goroutine
c.aggEngine, err = agg.NewAggregator(c.output)
c.cache.Init(numPeriods)
c.aggEngine, err = agg.NewAggregatorExpr(c.output)
if err != nil {
return fmt.Errorf("MetricCache: failed to create aggregator: %w", err)
}
@@ -81,69 +129,56 @@ func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker,
func (c *metricCache) Start() {
c.tickchan = make(chan time.Time)
c.ticker.AddChannel(c.tickchan)
// Router cache is done
done := func() {
cclog.ComponentDebug("MetricCache", "DONE")
close(c.done)
}
// Rotate cache interval
rotate := func(timestamp time.Time) int {
oldPeriod := c.curPeriod
c.curPeriod = oldPeriod + 1
if c.curPeriod >= c.numPeriods {
c.curPeriod = 0
}
c.intervals[oldPeriod].numMetrics = 0
c.intervals[oldPeriod].stopstamp = timestamp
c.intervals[c.curPeriod].startstamp = timestamp
c.intervals[c.curPeriod].stopstamp = timestamp
return oldPeriod
}
c.wg.Go(func() {
c.wg.Add(1)
go func() {
for {
select {
case <-c.done:
done()
c.wg.Done()
close(c.done)
cclog.ComponentDebug("MetricCache", "DONE")
return
case tick := <-c.tickchan:
c.lock.Lock()
old := rotate(tick)
// Get the last period and evaluate aggregation metrics
starttime, endtime, metrics := c.GetPeriod(old)
c.lock.Unlock()
if len(metrics) > 0 {
c.aggEngine.Eval(starttime, endtime, metrics)
cclog.ComponentDebug("MetricCache", "Tick", tick)
allmetrics := c.cache.GetAll()
c.cache.NewPeriod()
mintime := tick
maxtime := mintime.AddDate(-1, 0, 0)
for _, metric := range allmetrics {
if metric.Time().Before(mintime) {
mintime = metric.Time()
}
if metric.Time().After(maxtime) {
maxtime = metric.Time()
}
}
if len(allmetrics) > 0 {
cclog.ComponentDebugf("MetricCache", "Evaluate %d metrics from %v to %v", len(allmetrics), mintime.UnixNano(), maxtime.UnixNano())
c.wg.Add(1)
go func() {
c.aggEngine.Eval(mintime, maxtime, allmetrics)
c.wg.Done()
}()
} else {
// This message is also printed in the first interval after startup
cclog.ComponentDebug("MetricCache", "EMPTY INTERVAL?")
}
}
}
})
cclog.ComponentDebug("MetricCache", "START")
}()
cclog.ComponentDebug("MetricCache", "STARTED")
}
// Add a metric to the cache. The interval is defined by the global timer (rotate() in Start())
// The intervals list is used as round-robin buffer and the metric list grows dynamically and
// to avoid reallocations
func (c *metricCache) Add(metric lp.CCMessage) {
if c.curPeriod >= 0 && c.curPeriod < c.numPeriods {
c.lock.Lock()
p := c.intervals[c.curPeriod]
if p.numMetrics < p.sizeMetrics {
p.metrics[p.numMetrics] = metric
p.numMetrics++
p.stopstamp = metric.Time()
} else {
p.metrics = append(p.metrics, metric)
p.numMetrics++
p.sizeMetrics++
p.stopstamp = metric.Time()
}
c.lock.Unlock()
}
c.cache.Add(metric)
}
func (c *metricCache) AddAggregation(name, function, condition string, tags, meta map[string]string) error {
@@ -154,40 +189,20 @@ func (c *metricCache) DeleteAggregation(name string) error {
return c.aggEngine.DeleteAggregation(name)
}
// Get all metrics of a interval. The index is the difference to the current interval, so index=0
// is the current one, index=1 the last interval and so on. Returns and empty array if a wrong index
// is given (negative index, index larger than configured number of total intervals, ...)
func (c *metricCache) GetPeriod(index int) (time.Time, time.Time, []lp.CCMessage) {
start := time.Now()
stop := time.Now()
var metrics []lp.CCMessage
if index >= 0 && index < c.numPeriods {
pindex := c.curPeriod - index
if pindex < 0 {
pindex = c.numPeriods - pindex
}
if pindex >= 0 && pindex < c.numPeriods {
start = c.intervals[pindex].startstamp
stop = c.intervals[pindex].stopstamp
metrics = c.intervals[pindex].metrics
} else {
metrics = make([]lp.CCMessage, 0)
}
} else {
metrics = make([]lp.CCMessage, 0)
}
return start, stop, metrics
}
// Close finishes / stops the metric cache
func (c *metricCache) Close() {
cclog.ComponentDebug("MetricCache", "CLOSE")
c.done <- true
if c.started {
c.done <- true
c.wg.Wait()
}
cclog.ComponentDebug("MetricCache", "CLOSED")
}
func NewCache(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) (MetricCache, error) {
c := new(metricCache)
err := c.Init(output, ticker, wg, numPeriods)
err := c.Init(output, ticker, wg, ticker.GetDuration(), numPeriods)
if err != nil {
return nil, err
}
internal/metricRouter/metricCache_test.go
+69
View File
@@ -0,0 +1,69 @@
// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
// All rights reserved. This file is part of cc-lib.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
// additional authors:
// Holger Obermaier (NHR@KIT)
package metricRouter
import (
"fmt"
"strings"
"sync"
"testing"
"time"
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
)
func TestCache(t *testing.T) {
output := make(chan lp.CCMessage, 2000)
var wg sync.WaitGroup
tickTime := time.Second
testChan := make(chan time.Time)
ticker := mct.NewTicker(tickTime)
ticker.AddChannel(testChan)
maxIntervals := 100
c, err := NewCache(output, ticker, &wg, 4)
if err != nil {
t.Errorf("failed to create new cache: %s", err.Error())
return
}
c.Start()
defer c.Close()
err = c.AddAggregation("ps_input_power", "Avg(values)", "name == 'ps1_input_power' || name == 'ps2_input_power' || name == 'ps3_input_power'", map[string]string{"hostname": "<copy>", "type": "<copy>"}, map[string]string{"unit": "<copy>"})
if err != nil {
t.Errorf("failed to add aggregation for %s", "ps_input_power")
return
}
for range maxIntervals {
timestamp := <-testChan
raw_metrics := []string{
fmt.Sprintf("ps1_input_power,type=node,hostname=myhost,unit=W value=696.0 %d", timestamp.UnixNano()),
fmt.Sprintf("ps2_input_power,type=node,hostname=myhost,unit=W value=732.0 %d", timestamp.UnixNano()),
fmt.Sprintf("ps3_input_power,type=node,hostname=myhost,unit=W value=720.0 %d", timestamp.UnixNano()),
fmt.Sprintf("cpu_load,type=hwthread,type-id=0,hostname=myhost value=45 %d", timestamp.UnixNano()),
}
metrics, err := lp.FromBytes([]byte(strings.Join(raw_metrics, "\n")))
if err != nil {
t.Errorf("failed to generate metrics: %s", err.Error())
}
for _, m := range metrics {
c.Add(m)
}
if len(output) > 0 {
for i := 0; i < len(output); i++ {
m := <-output
t.Log(m.ToLineProtocol(nil))
}
}
}
}
internal/metricRouter/metricRouter.go
+14 -14
View File
@@ -35,19 +35,19 @@ type metricRouterTagConfig struct {
// Metric router configuration
type metricRouterConfig struct {
HostnameTagName string `json:"hostname_tag"` // Key name used when adding the hostname to a metric (default 'hostname')
AddTags []metricRouterTagConfig `json:"add_tags"` // List of tags that are added when the condition is met
DelTags []metricRouterTagConfig `json:"delete_tags"` // List of tags that are removed when the condition is met
IntervalAgg []agg.MetricAggregatorIntervalConfig `json:"interval_aggregates"` // List of aggregation function processed at the end of an interval
DropMetrics []string `json:"drop_metrics"` // List of metric names to drop. For fine-grained dropping use drop_metrics_if
DropMetricsIf []string `json:"drop_metrics_if"` // List of evaluatable terms to drop metrics
RenameMetrics map[string]string `json:"rename_metrics"` // Map to rename metric name from key to value
IntervalStamp bool `json:"interval_timestamp"` // Update timestamp periodically by ticker each interval?
NumCacheIntervals int `json:"num_cache_intervals"` // Number of intervals of cached metrics for evaluation
MaxForward int `json:"max_forward"` // Number of maximal forwarded metrics at one select
NormalizeUnits bool `json:"normalize_units"` // Check unit meta flag and normalize it using cc-units
ChangeUnitPrefix map[string]string `json:"change_unit_prefix"` // Add prefix that should be applied to the metrics
MessageProcessor json.RawMessage `json:"process_messages,omitempty"`
HostnameTagName string `json:"hostname_tag"` // Key name used when adding the hostname to a metric (default 'hostname')
AddTags []metricRouterTagConfig `json:"add_tags"` // List of tags that are added when the condition is met
DelTags []metricRouterTagConfig `json:"delete_tags"` // List of tags that are removed when the condition is met
IntervalAgg []agg.MetricAggregatorExprIntervalConfig `json:"interval_aggregates"` // List of aggregation function processed at the end of an interval
DropMetrics []string `json:"drop_metrics"` // List of metric names to drop. For fine-grained dropping use drop_metrics_if
DropMetricsIf []string `json:"drop_metrics_if"` // List of evaluatable terms to drop metrics
RenameMetrics map[string]string `json:"rename_metrics"` // Map to rename metric name from key to value
IntervalStamp bool `json:"interval_timestamp"` // Update timestamp periodically by ticker each interval?
NumCacheIntervals int `json:"num_cache_intervals"` // Number of intervals of cached metrics for evaluation
MaxForward int `json:"max_forward"` // Number of maximal forwarded metrics at one select
NormalizeUnits bool `json:"normalize_units"` // Check unit meta flag and normalize it using cc-units
ChangeUnitPrefix map[string]string `json:"change_unit_prefix"` // Add prefix that should be applied to the metrics
MessageProcessor json.RawMessage `json:"process_messages,omitempty"`
}
// Metric router data structure
@@ -253,7 +253,7 @@ func (r *metricRouter) Start() {
}
// even if the metric is dropped, it is stored in the cache for
// aggregations
if r.config.NumCacheIntervals > 0 {
if r.config.NumCacheIntervals > 0 && m != nil {
r.cache.Add(m)
}
}