ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2025-08-14 06:52:58 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into use_metric_pointers

Browse Source
This commit is contained in:
Thomas Gruber
2022-02-02 15:53:26 +01:00
committed by GitHub
parent 7f4de77de1 1222f7a32f
commit 2611b1e301
10 changed files with 996 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
internal/metricRouter/README.md
View File

@@ -6,6 +6,8 @@ The CCMetric router sits in between the collectors and the sinks and can be used
```json
{
"num_cache_intervals" : 1,
"interval_timestamp" : true,
"add_tags" : [
{
"key" : "cluster",
@@ -25,16 +27,58 @@ The CCMetric router sits in between the collectors and the sinks and can be used
"if" : "*"
}
],
"interval_timestamp" : true
"interval_aggregates" : [
{
"name" : "temp_cores_avg",
"if" : "match('temp_core_%d+', metric.Name())",
"function" : "avg(values)",
"tags" : {
"type" : "node"
},
"meta" : {
"group": "IPMI",
"unit": "degC",
"source": "TempCollector"
}
}
],
"drop_metrics" : [
"not_interesting_metric_at_all"
],
"drop_metrics_if" : [
"match('temp_core_%d+', metric.Name())"
],
"rename_metrics" : {
"metric_12345" : "mymetric"
}
}
```
There are three main options `add_tags`, `delete_tags` and `interval_timestamp`. `add_tags` and `delete_tags` are lists consisting of dicts with `key`, `value` and `if`. The `value` can be omitted in the `delete_tags` part as it only uses the `key` for removal. The `interval_timestamp` setting means that a unique timestamp is applied to all metrics traversing the router during an interval.
# The `interval_timestamp` option
# Conditional manipulation of tags
The collectors' `Read()` functions are not called simultaneously and therefore the metrics gathered in an interval can have different timestamps. If you want to avoid that and have a common timestamp (the beginning of the interval), set this option to `true` and the MetricRouter sets the time.
The `if` setting allows conditional testing of a single metric like in the example:
# The `num_cache_intervals` option
If the MetricRouter should buffer metrics of intervals in a MetricCache, this option specifies the number of past intervals that should be kept. If `num_cache_intervals = 0`, the cache is disabled. With `num_cache_intervals = 1`, only the metrics of the last interval are buffered.
A `num_cache_intervals > 0` is required to use the `interval_aggregates` option.
# The `rename_metrics` option
In the ClusterCockpit world we specified a set of standard metrics. Since some collectors determine the metric names based on files, execuables and libraries, they might change from system to system (or installation to installtion, OS to OS, ...). In order to get the common names, you can rename incoming metrics before sending them to the sink. If the metric name matches the `oldname`, it is changed to `newname`
```json
{
"oldname" : "newname",
"clock_mhz" : "clock"
}
```
# Conditional manipulation of tags (`add_tags` and `del_tags`)
Common config format:
```json
{
"key" : "test",
@@ -43,8 +87,131 @@ The `if` setting allows conditional testing of a single metric like in the examp
}
```
If the CCMetric name is equal to 'temp_package_id_0', it adds an additional tag `test=testing` to the metric.
## The `del_tags` option
In order to match all metrics, you can use `*`, so in order to add a flag per default, like the `cluster=testcluster` tag in the example.
The collectors are free to add whatever `key=value` pair to the metric tags (although the usage of tags should be minimized). If you want to delete a tag afterwards, you can do that. When the `if` condition matches on a metric, the `key` is removed from the metric's tags.
If you want to remove a tag for all metrics, use the condition wildcard `*`. The `value` field can be omitted in the `del_tags` case.
Never delete tags:
- `hostname`
- `type`
- `type-id`
## The `add_tags` option
In some cases, metrics should be tagged or an existing tag changed based on some condition. This can be done in the `add_tags` section. When the `if` condition evaluates to `true`, the tag `key` is added or gets changed to the new `value`.
If the CCMetric name is equal to `temp_package_id_0`, it adds an additional tag `test=testing` to the metric.
For this metric, a more useful example would be:
```json
[
{
"key" : "type",
"value" : "socket",
"if" : "name == 'temp_package_id_0'"
},
{
"key" : "type-id",
"value" : "0",
"if" : "name == 'temp_package_id_0'"
},
]
```
The metric `temp_package_id_0` corresponds to the tempature of the first CPU socket (=package). With the above configuration, the tags would reflect that because commonly the [TempCollector](../../collectors/tempMetric.md) submits only `node` metrics.
In order to match all metrics, you can use `*`, so in order to add a flag per default. This is useful to attached system-specific tags like `cluster=testcluster`:
```json
{
"key" : "cluster",
"value" : "testcluster",
"if" : "*"
}
```
# Dropping metrics
In some cases, you want to drop a metric and don't get it forwarded to the sinks. There are two options based on the required specification:
- Based only on the metric name -> `drop_metrics` section
- An evaluable condition with more overhead -> `drop_metrics_if` section
## The `drop_metrics` section
The argument is a list of metric names. No futher checks are performed, only a comparison of the metric name
```json
{
"drop_metrics" : [
"drop_metric_1",
"drop_metric_2"
]
}
```
The example drops all metrics with the name `drop_metric_1` and `drop_metric_2`.
## The `drop_metrics_if` section
This option takes a list of evaluable conditions and performs them one after the other on **all** metrics incoming from the collectors and the metric cache (aka `interval_aggregates`).
```json
{
"drop_metrics_if" : [
"match('drop_metric_%d+', name)",
"match('cpu', type) && type-id == 0"
]
}
```
The first line is comparable with the example in `drop_metrics`, it drops all metrics starting with `drop_metric_` and ending with a number. The second line drops all metrics of the first hardware thread (**not** recommended)
# Aggregate metric values of the current interval with the `interval_aggregates` option
**Note:** `interval_aggregates` works only if `num_cache_intervals` > 0
In some cases, you need to derive new metrics based on the metrics arriving during an interval. This can be done in the `interval_aggregates` section. The logic is similar to the other metric manipulation and filtering options. A cache stores all metrics that arrive during an interval. At the beginning of the *next* interval, the list of metrics is submitted to the MetricAggregator. It derives new metrics and submits them back to the MetricRouter, so they are sent in the next interval but have the timestamp of the previous interval beginning.
```json
"interval_aggregates" : [
{
"name" : "new_metric_name",
"if" : "match('sub_metric_%d+', metric.Name())",
"function" : "avg(values)",
"tags" : {
"key" : "value",
"type" : "node"
},
"meta" : {
"key" : "value",
"group": "IPMI",
"unit": "<copy>",
}
}
]
```
The above configuration, collects all metric values for metrics evaluating `if` to `true`. Afterwards it calculates the average `avg` of the `values` (list of all metrics' field `value`) and creates a new CCMetric with the name `new_metric_name` and adds the tags in `tags` and the meta information in `meta`. The special value `<copy>` searches the input metrics and copies the value of the first match of `key` to the new CCMetric.
If you are not interested in the input metrics `sub_metric_%d+` at all, you can add the same condition used here to the `drop_metrics_if` section to drop them.
Use cases for `interval_aggregates`:
- Combine multiple metrics of the a collector to a new one like the [MemstatCollector](../../collectors/memstatMetric.md) does it for `mem_used`)):
```json
{
"name" : "mem_used",
"if" : "source == 'MemstatCollector'",
"function" : "sum(mem_total) - (sum(mem_free) + sum(mem_buffers) + sum(mem_cached))",
"tags" : {
"type" : "node"
},
"meta" : {
"group": "<copy>",
"unit": "<copy>",
"source": "<copy>"
}
}
```

83
internal/metricRouter/metricAggregator.go
View File

@@ -3,6 +3,7 @@ package metricRouter
import (
"context"
"fmt"
"math"
"os"
"strings"
"time"
@@ -84,7 +85,7 @@ func (c *metricAggregator) Init(output chan *lp.CCMetric) error {
c.constants["smtWidth"] = cinfo.SMTWidth
c.language = gval.NewLanguage(
gval.Base(),
gval.Full(),
metricCacheLanguage,
)
@@ -283,6 +284,86 @@ func (c *metricAggregator) AddFunction(name string, function func(args ...interf
c.language = gval.NewLanguage(c.language, gval.Function(name, function))
}
func EvalBoolCondition(condition string, params map[string]interface{}) (bool, error) {
newcond := strings.ReplaceAll(condition, "'", "\"")
newcond = strings.ReplaceAll(newcond, "%", "\\")
language := gval.NewLanguage(
gval.Full(),
metricCacheLanguage,
)
value, err := gval.Evaluate(newcond, params, language)
if err != nil {
return false, err
}
var endResult bool = false
err = nil
switch r := value.(type) {
case bool:
endResult = r
case float64:
if r != 0.0 {
endResult = true
}
case float32:
if r != 0.0 {
endResult = true
}
case int:
if r != 0 {
endResult = true
}
case int64:
if r != 0 {
endResult = true
}
case int32:
if r != 0 {
endResult = true
}
default:
err = fmt.Errorf("cannot evaluate '%s' to bool", newcond)
}
return endResult, err
}
func EvalFloat64Condition(condition string, params map[string]interface{}) (float64, error) {
var endResult float64 = math.NaN()
newcond := strings.ReplaceAll(condition, "'", "\"")
newcond = strings.ReplaceAll(newcond, "%", "\\")
language := gval.NewLanguage(
gval.Full(),
metricCacheLanguage,
)
value, err := gval.Evaluate(newcond, params, language)
if err != nil {
cclog.ComponentDebug("MetricRouter", condition, " = ", err.Error())
return endResult, err
}
err = nil
switch r := value.(type) {
case bool:
if r {
endResult = 1.0
} else {
endResult = 0.0
}
case float64:
endResult = r
case float32:
endResult = float64(r)
case int:
endResult = float64(r)
case int64:
endResult = float64(r)
case int32:
endResult = float64(r)
default:
err = fmt.Errorf("cannot evaluate '%s' to float64", newcond)
}
return endResult, err
}
func NewAggregator(output chan *lp.CCMetric) (MetricAggregator, error) {
a := new(metricAggregator)
err := a.Init(output)

118
internal/metricRouter/metricRouter.go
View File

@@ -11,7 +11,6 @@ import (
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
mct "github.com/ClusterCockpit/cc-metric-collector/internal/multiChanTicker"
"gopkg.in/Knetic/govaluate.v2"
)
// Metric router tag configuration
@@ -26,8 +25,12 @@ 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
}
// Metric router data structure
@@ -92,17 +95,19 @@ func (r *metricRouter) Init(ticker mct.MultiChanTicker, wg *sync.WaitGroup, rout
cclog.ComponentError("MetricRouter", err.Error())
return err
}
numIntervals := r.config.NumCacheIntervals
if numIntervals <= 0 {
numIntervals = 1
if r.config.NumCacheIntervals >= 0 {
r.cache, err = NewCache(r.cache_input, r.ticker, &r.cachewg, r.config.NumCacheIntervals)
if err != nil {
cclog.ComponentError("MetricRouter", "MetricCache initialization failed:", err.Error())
return err
}
for _, agg := range r.config.IntervalAgg {
r.cache.AddAggregation(agg.Name, agg.Function, agg.Condition, agg.Tags, agg.Meta)
}
}
r.cache, err = NewCache(r.cache_input, r.ticker, &r.cachewg, numIntervals)
if err != nil {
cclog.ComponentError("MetricRouter", "MetricCache initialization failed:", err.Error())
return err
}
for _, agg := range r.config.IntervalAgg {
r.cache.AddAggregation(agg.Name, agg.Function, agg.Condition, agg.Tags, agg.Meta)
r.config.dropMetrics = make(map[string]bool)
for _, mname := range r.config.DropMetrics {
r.config.dropMetrics[mname] = true
}
return nil
}
@@ -130,48 +135,34 @@ func (r *metricRouter) StartTimer() {
cclog.ComponentDebug("MetricRouter", "TIMER START")
}
// EvalCondition evaluates condition cond for metric data from point
func (r *metricRouter) EvalCondition(cond string, pptr *lp.CCMetric) (bool, error) {
point := *pptr
expression, err := govaluate.NewEvaluableExpression(cond)
if err != nil {
cclog.ComponentDebug("MetricRouter", cond, " = ", err.Error())
return false, err
}
// Add metric name, tags, meta data, fields and timestamp to the parameter list
func getParamMap(point lp.CCMetric) map[string]interface{} {
params := make(map[string]interface{})
params["metric"] = point
params["name"] = point.Name()
for _, t := range point.TagList() {
params[t.Key] = t.Value
for key, value := range point.Tags() {
params[key] = value
}
for _, m := range point.MetaList() {
params[m.Key] = m.Value
for key, value := range point.Meta() {
params[key] = value
}
for _, f := range point.FieldList() {
params[f.Key] = f.Value
}
params["timestamp"] = point.Time()
// evaluate condition
result, err := expression.Evaluate(params)
if err != nil {
cclog.ComponentDebug("MetricRouter", cond, " = ", err.Error())
return false, err
}
return bool(result.(bool)), err
return params
}
// DoAddTags adds a tag when condition is fullfiled
func (r *metricRouter) DoAddTags(point *lp.CCMetric) {
for _, m := range r.config.AddTags {
var conditionMatches bool
var conditionMatches bool = false
if m.Condition == "*" {
conditionMatches = true
} else {
var err error
conditionMatches, err = r.EvalCondition(m.Condition, point)
conditionMatches, err = EvalBoolCondition(m.Condition, getParamMap(point))
if err != nil {
cclog.ComponentError("MetricRouter", err.Error())
conditionMatches = false
@@ -186,13 +177,13 @@ func (r *metricRouter) DoAddTags(point *lp.CCMetric) {
// DoDelTags removes a tag when condition is fullfiled
func (r *metricRouter) DoDelTags(point *lp.CCMetric) {
for _, m := range r.config.DelTags {
var conditionMatches bool
var conditionMatches bool = false
if m.Condition == "*" {
conditionMatches = true
} else {
var err error
conditionMatches, err = r.EvalCondition(m.Condition, point)
conditionMatches, err = EvalBoolCondition(m.Condition, getParamMap(point))
if err != nil {
cclog.ComponentError("MetricRouter", err.Error())
conditionMatches = false
@@ -204,9 +195,24 @@ func (r *metricRouter) DoDelTags(point *lp.CCMetric) {
}
}
// Conditional test whether a metric should be dropped
func (r *metricRouter) dropMetric(point *lp.CCMetric) bool {
// Simple drop check
if _, ok := r.config.dropMetrics[(*point).Name()]; ok {
return true
}
// Checking the dropping conditions
for _, m := range r.config.DropMetricsIf {
conditionMatches, err := EvalBoolCondition(m, getParamMap((*point)))
if conditionMatches || err != nil {
return true
}
}
return false
}
// Start starts the metric router
func (r *metricRouter) Start() {
// start timer if configured
r.timestamp = time.Now()
if r.config.IntervalStamp {
@@ -225,13 +231,21 @@ func (r *metricRouter) Start() {
cclog.ComponentDebug("MetricRouter", "FORWARD", *point)
r.DoAddTags(point)
r.DoDelTags(point)
if new, ok := r.config.RenameMetrics[point.Name()]; ok {
point.SetName(new)
}
r.DoAddTags(point)
r.DoDelTags(point)
for _, o := range r.outputs {
o <- point
}
}
// Start Metric Cache
r.cache.Start()
if r.config.NumCacheIntervals > 0 {
r.cache.Start()
}
r.wg.Add(1)
go func() {
@@ -248,20 +262,30 @@ func (r *metricRouter) Start() {
if r.config.IntervalStamp {
(*p).SetTime(r.timestamp)
}
forward(p)
r.cache.Add(p)
if !r.dropMetric(p) {
forward(p)
}
// even if the metric is dropped, it is stored in the cache for
// aggregations
if r.config.NumCacheIntervals > 0 {
r.cache.Add(p)
}
case p := <-r.recv_input:
// receive from receive manager
if r.config.IntervalStamp {
(*p).SetTime(r.timestamp)
}
forward(p)
if !r.dropMetric(p) {
forward(p)
}
case p := <-r.cache_input:
// receive from metric cache and aggregator
(*p).AddTag("hostname", r.hostname)
forward(p)
// receive from metric collector
if !r.dropMetric(p) {
(*p).AddTag("hostname", r.hostname)
forward(p)
}
}
}
}()
@@ -295,8 +319,10 @@ func (r *metricRouter) Close() {
// wait for close of channel r.timerdone
<-r.timerdone
}
r.cache.Close()
r.cachewg.Wait()
if r.config.NumCacheIntervals > 0 {
r.cache.Close()
r.cachewg.Wait()
}
}
// New creates a new initialized metric router