From b79567623de3783148cc6738e30be47b332672d9 Mon Sep 17 00:00:00 2001
From: Thomas Roehl <Thomas.Roehl@googlemail.com>
Date: Wed, 2 Feb 2022 17:03:10 +0100
Subject: [PATCH] Split MetricRouter and MetricAggregator
---
collectors/likwidMetric.go | 18 ++++++-------
.../metricAggregator.go | 10 ++++----
internal/metricAggregator/metricAggregator.md | 0
.../metricAggregatorFunctions.go | 2 +-
internal/metricRouter/metricRouter.go | 25 ++++++++++---------
5 files changed, 28 insertions(+), 27 deletions(-)
rename internal/{metricRouter => metricAggregator}/metricAggregator.go (98%)
create mode 100644 internal/metricAggregator/metricAggregator.md
rename internal/{metricRouter => metricAggregator}/metricAggregatorFunctions.go (99%)
diff --git a/collectors/likwidMetric.go b/collectors/likwidMetric.go
index 54f025a..3eaa590 100644
--- a/collectors/likwidMetric.go
+++ b/collectors/likwidMetric.go
@@ -24,7 +24,7 @@ import (
cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
topo "github.com/ClusterCockpit/cc-metric-collector/internal/ccTopology"
- mr "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
+ agg "github.com/ClusterCockpit/cc-metric-collector/internal/metricAggregator"
)
type MetricScope string
@@ -70,10 +70,10 @@ func GetAllMetricScopes() []MetricScope {
}
type LikwidCollectorMetricConfig struct {
- Name string `json:"name"` // Name of the metric
- Calc string `json:"calc"` // Calculation for the metric using
- Aggr string `json:"aggregation"` // if scope unequal to LIKWID metric scope, the values are combined (sum, min, max, mean or avg, median)
- Scope MetricScope `json:"scope"` // scope for calculation. subscopes are aggregated using the 'aggregation' function
+ Name string `json:"name"` // Name of the metric
+ Calc string `json:"calc"` // Calculation for the metric using
+ //Aggr string `json:"aggregation"` // if scope unequal to LIKWID metric scope, the values are combined (sum, min, max, mean or avg, median)
+ Scope MetricScope `json:"scope"` // scope for calculation. subscopes are aggregated using the 'aggregation' function
Publish bool `json:"publish"`
granulatity MetricScope
}
@@ -316,7 +316,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
}
for _, metric := range evset.Metrics {
// Try to evaluate the metric
- _, err := mr.EvalFloat64Condition(metric.Calc, params)
+ _, err := agg.EvalFloat64Condition(metric.Calc, params)
if err != nil {
cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
continue
@@ -345,7 +345,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
}
for _, metric := range m.config.Metrics {
// Try to evaluate the global metric
- _, err := mr.EvalFloat64Condition(metric.Calc, globalParams)
+ _, err := agg.EvalFloat64Condition(metric.Calc, globalParams)
if err != nil {
cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
continue
@@ -430,7 +430,7 @@ func (m *LikwidCollector) calcEventsetMetrics(group int, interval time.Duration,
scopemap := m.scopeRespTids[metric.Scope]
for domain, tid := range scopemap {
if tid >= 0 {
- value, err := mr.EvalFloat64Condition(metric.Calc, m.results[group][tid])
+ value, err := agg.EvalFloat64Condition(metric.Calc, m.results[group][tid])
if err != nil {
cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
continue
@@ -467,7 +467,7 @@ func (m *LikwidCollector) calcGlobalMetrics(interval time.Duration, output chan
}
}
// Evaluate the metric
- value, err := mr.EvalFloat64Condition(metric.Calc, params)
+ value, err := agg.EvalFloat64Condition(metric.Calc, params)
if err != nil {
cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
continue
diff --git a/internal/metricRouter/metricAggregator.go b/internal/metricAggregator/metricAggregator.go
similarity index 98%
rename from internal/metricRouter/metricAggregator.go
rename to internal/metricAggregator/metricAggregator.go
index e3303e4..a05f061 100644
--- a/internal/metricRouter/metricAggregator.go
+++ b/internal/metricAggregator/metricAggregator.go
@@ -1,4 +1,4 @@
-package metricRouter
+package metricAggregator
import (
"context"
@@ -16,7 +16,7 @@ import (
"github.com/PaesslerAG/gval"
)
-type metricAggregatorIntervalConfig struct {
+type MetricAggregatorIntervalConfig struct {
Name string `json:"name"` // Metric name for the new metric
Function string `json:"function"` // Function to apply on the metric
Condition string `json:"if"` // Condition for applying function
@@ -27,7 +27,7 @@ type metricAggregatorIntervalConfig struct {
}
type metricAggregator struct {
- functions []*metricAggregatorIntervalConfig
+ functions []*MetricAggregatorIntervalConfig
constants map[string]interface{}
language gval.Language
output chan lp.CCMetric
@@ -65,7 +65,7 @@ var metricCacheLanguage = gval.NewLanguage(
func (c *metricAggregator) Init(output chan lp.CCMetric) error {
c.output = output
- c.functions = make([]*metricAggregatorIntervalConfig, 0)
+ c.functions = make([]*MetricAggregatorIntervalConfig, 0)
c.constants = make(map[string]interface{})
// add constants like hostname, numSockets, ... to constants list
@@ -246,7 +246,7 @@ func (c *metricAggregator) AddAggregation(name, function, condition string, tags
return nil
}
}
- var agg metricAggregatorIntervalConfig
+ var agg MetricAggregatorIntervalConfig
agg.Name = name
agg.Condition = newcond
agg.gvalCond = gvalCond
diff --git a/internal/metricAggregator/metricAggregator.md b/internal/metricAggregator/metricAggregator.md
new file mode 100644
index 0000000..e69de29
diff --git a/internal/metricRouter/metricAggregatorFunctions.go b/internal/metricAggregator/metricAggregatorFunctions.go
similarity index 99%
rename from internal/metricRouter/metricAggregatorFunctions.go
rename to internal/metricAggregator/metricAggregatorFunctions.go
index f00479d..1fbef65 100644
--- a/internal/metricRouter/metricAggregatorFunctions.go
+++ b/internal/metricAggregator/metricAggregatorFunctions.go
@@ -1,4 +1,4 @@
-package metricRouter
+package metricAggregator
import (
"errors"
diff --git a/internal/metricRouter/metricRouter.go b/internal/metricRouter/metricRouter.go
index 6d63e15..a31f2a6 100644
--- a/internal/metricRouter/metricRouter.go
+++ b/internal/metricRouter/metricRouter.go
@@ -10,6 +10,7 @@ import (
cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+ agg "github.com/ClusterCockpit/cc-metric-collector/internal/metricAggregator"
mct "github.com/ClusterCockpit/cc-metric-collector/internal/multiChanTicker"
)
@@ -22,15 +23,15 @@ type metricRouterTagConfig struct {
// Metric router configuration
type metricRouterConfig struct {
- 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 []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
- dropMetrics map[string]bool // Internal map for O(1) lookup
+ 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
+ dropMetrics map[string]bool // Internal map for O(1) lookup
}
// Metric router data structure
@@ -161,7 +162,7 @@ func (r *metricRouter) DoAddTags(point lp.CCMetric) {
conditionMatches = true
} else {
var err error
- conditionMatches, err = EvalBoolCondition(m.Condition, getParamMap(point))
+ conditionMatches, err = agg.EvalBoolCondition(m.Condition, getParamMap(point))
if err != nil {
cclog.ComponentError("MetricRouter", err.Error())
conditionMatches = false
@@ -182,7 +183,7 @@ func (r *metricRouter) DoDelTags(point lp.CCMetric) {
conditionMatches = true
} else {
var err error
- conditionMatches, err = EvalBoolCondition(m.Condition, getParamMap(point))
+ conditionMatches, err = agg.EvalBoolCondition(m.Condition, getParamMap(point))
if err != nil {
cclog.ComponentError("MetricRouter", err.Error())
conditionMatches = false
@@ -202,7 +203,7 @@ func (r *metricRouter) dropMetric(point lp.CCMetric) bool {
}
// Checking the dropping conditions
for _, m := range r.config.DropMetricsIf {
- conditionMatches, err := EvalBoolCondition(m, getParamMap(point))
+ conditionMatches, err := agg.EvalBoolCondition(m, getParamMap(point))
if conditionMatches || err != nil {
return true
}