ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-12-25 15:09:05 +01:00
Code Issues Packages Projects Releases Wiki Activity

Make maxForward configurable, save old name in meta in rename metrics and make the hostname tag key configurable

Browse Source
This commit is contained in:
Thomas Roehl 2022-03-09 11:23:17 +01:00
parent e7f7e68095
commit 3cf2f69a07
1 changed files with 13 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
internal/metricRouter/metricRouter.go
View File

@ -25,6 +25,7 @@ type metricRouterTagConfig struct {
// Metric router configuration // Metric router configuration
type metricRouterConfig struct { 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 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 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 IntervalAgg []agg.MetricAggregatorIntervalConfig `json:"interval_aggregates"` // List of aggregation function processed at the end of an interval
@ -33,6 +34,7 @@ type metricRouterConfig struct {
RenameMetrics map[string]string `json:"rename_metrics"` // Map to rename metric name from key to value 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? 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 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
dropMetrics map[string]bool // Internal map for O(1) lookup dropMetrics map[string]bool // Internal map for O(1) lookup
} }
@ -76,7 +78,8 @@ func (r *metricRouter) Init(ticker mct.MultiChanTicker, wg *sync.WaitGroup, rout
r.cache_input = make(chan lp.CCMetric) r.cache_input = make(chan lp.CCMetric)
r.wg = wg r.wg = wg
r.ticker = ticker r.ticker = ticker
r.maxForward = ROUTER_MAX_FORWARD r.config.MaxForward = ROUTER_MAX_FORWARD
r.config.HostnameTagName = "hostname"
// Set hostname // Set hostname
hostname, err := os.Hostname() hostname, err := os.Hostname()
@ -100,6 +103,7 @@ func (r *metricRouter) Init(ticker mct.MultiChanTicker, wg *sync.WaitGroup, rout
cclog.ComponentError("MetricRouter", err.Error()) cclog.ComponentError("MetricRouter", err.Error())
return err return err
} }
r.maxForward = r.config.MaxForward
if r.config.NumCacheIntervals > 0 { if r.config.NumCacheIntervals > 0 {
r.cache, err = NewCache(r.cache_input, r.ticker, &r.cachewg, r.config.NumCacheIntervals) r.cache, err = NewCache(r.cache_input, r.ticker, &r.cachewg, r.config.NumCacheIntervals)
if err != nil { if err != nil {
@ -244,8 +248,10 @@ func (r *metricRouter) Start() {
cclog.ComponentDebug("MetricRouter", "FORWARD", point) cclog.ComponentDebug("MetricRouter", "FORWARD", point)
r.DoAddTags(point) r.DoAddTags(point)
r.DoDelTags(point) r.DoDelTags(point)
if new, ok := r.config.RenameMetrics[point.Name()]; ok { name := point.Name()
if new, ok := r.config.RenameMetrics[name]; ok {
point.SetName(new) point.SetName(new)
point.AddMeta("oldname", name)
} }
r.DoAddTags(point) r.DoAddTags(point)
r.DoDelTags(point) r.DoDelTags(point)
@ -258,7 +264,7 @@ func (r *metricRouter) Start() {
// Foward message received from collector channel // Foward message received from collector channel
coll_forward := func(p lp.CCMetric) { coll_forward := func(p lp.CCMetric) {
// receive from metric collector // receive from metric collector
p.AddTag("hostname", r.hostname) p.AddTag(r.config.HostnameTagName, r.hostname)
if r.config.IntervalStamp { if r.config.IntervalStamp {
p.SetTime(r.timestamp) p.SetTime(r.timestamp)
} }
@ -287,7 +293,7 @@ func (r *metricRouter) Start() {
cache_forward := func(p lp.CCMetric) { cache_forward := func(p lp.CCMetric) {
// receive from metric collector // receive from metric collector
if !r.dropMetric(p) { if !r.dropMetric(p) {
p.AddTag("hostname", r.hostname) p.AddTag(r.config.HostnameTagName, r.hostname)
forward(p) forward(p)
} }
} }
@ -309,19 +315,19 @@ func (r *metricRouter) Start() {
case p := <-r.coll_input: case p := <-r.coll_input:
coll_forward(p) coll_forward(p)
for i := 0; len(r.coll_input) > 0 && i < r.maxForward; i++ { for i := 0; len(r.coll_input) > 0 && i < (r.maxForward-1); i++ {
coll_forward(<-r.coll_input) coll_forward(<-r.coll_input)
} }
case p := <-r.recv_input: case p := <-r.recv_input:
recv_forward(p) recv_forward(p)
for i := 0; len(r.recv_input) > 0 && i < r.maxForward; i++ { for i := 0; len(r.recv_input) > 0 && i < (r.maxForward-1); i++ {
recv_forward(<-r.recv_input) recv_forward(<-r.recv_input)
} }
case p := <-r.cache_input: case p := <-r.cache_input:
cache_forward(p) cache_forward(p)
for i := 0; len(r.cache_input) > 0 && i < r.maxForward; i++ { for i := 0; len(r.cache_input) > 0 && i < (r.maxForward-1); i++ {
cache_forward(<-r.cache_input) cache_forward(<-r.cache_input)
} }
} }