Add override processor (#3773)
This plugin can perform the standard metric modifications using override semantics.
This commit is contained in:
parent
64c1a7a860
commit
82d224d9e2
|
@ -1,5 +1,6 @@
|
|||
package all
|
||||
|
||||
import (
|
||||
_ "github.com/influxdata/telegraf/plugins/processors/override"
|
||||
_ "github.com/influxdata/telegraf/plugins/processors/printer"
|
||||
)
|
||||
|
|
|
@ -0,0 +1,30 @@
|
|||
# Override Processor Plugin
|
||||
|
||||
The override processor plugin allows overriding all modifications that are
|
||||
supported by input plugins and aggregators:
|
||||
|
||||
* name_override
|
||||
* name_prefix
|
||||
* name_suffix
|
||||
* tags
|
||||
|
||||
All metrics passing through this processor will be modified accordingly. Values
|
||||
of *name_override*, *name_prefix*, *name_suffix* and already present *tags* with
|
||||
conflicting keys will be overwritten. Absent *tags* will be created.
|
||||
|
||||
Use-case of this plugin encompass ensuring certain tags or naming conventions
|
||||
are adhered to irrespective of input plugin configurations, e.g. by
|
||||
`taginclude`.
|
||||
|
||||
### Configuration:
|
||||
|
||||
```toml
|
||||
# Add a global tag to all metrics
|
||||
[[processors.override]]
|
||||
name_override = "new name_override"
|
||||
name_prefix = "new name_prefix"
|
||||
name_suffix = ":new name_suffix"
|
||||
[processors.tags.add]
|
||||
additional_tag = "tag_value"
|
||||
existing_tag = "new tag_value"
|
||||
```
|
|
@ -0,0 +1,60 @@
|
|||
package override
|
||||
|
||||
import (
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/plugins/processors"
|
||||
)
|
||||
|
||||
var sampleConfig = `
|
||||
## NOTE This processor will override names, name prefixes, name suffixes and
|
||||
## values of tags, that are already present in the metric passed through this
|
||||
## filter.
|
||||
|
||||
## All modifications on inputs and aggregators can be overridden:
|
||||
# name_override = "new name"
|
||||
# name_prefix = "new name_prefix"
|
||||
# name_suffix = "new name_suffix"
|
||||
|
||||
## Tags to be added (all values must be strings)
|
||||
# [processors.overide.tags]
|
||||
# additional_tag = "tag_value"
|
||||
`
|
||||
|
||||
type Override struct {
|
||||
NameOverride string
|
||||
NamePrefix string
|
||||
NameSuffix string
|
||||
Tags map[string]string
|
||||
}
|
||||
|
||||
func (p *Override) SampleConfig() string {
|
||||
return sampleConfig
|
||||
}
|
||||
|
||||
func (p *Override) Description() string {
|
||||
return "Add all configured tags to all metrics that pass through this filter."
|
||||
}
|
||||
|
||||
func (p *Override) Apply(in ...telegraf.Metric) []telegraf.Metric {
|
||||
for _, metric := range in {
|
||||
if len(p.NameOverride) > 0 {
|
||||
metric.SetName(p.NameOverride)
|
||||
}
|
||||
if len(p.NamePrefix) > 0 {
|
||||
metric.SetPrefix(p.NamePrefix)
|
||||
}
|
||||
if len(p.NameSuffix) > 0 {
|
||||
metric.SetSuffix(p.NameSuffix)
|
||||
}
|
||||
for key, value := range p.Tags {
|
||||
metric.AddTag(key, value)
|
||||
}
|
||||
}
|
||||
return in
|
||||
}
|
||||
|
||||
func init() {
|
||||
processors.Add("override", func() telegraf.Processor {
|
||||
return &Override{}
|
||||
})
|
||||
}
|
|
@ -0,0 +1,80 @@
|
|||
package override
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/metric"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func createTestMetric() telegraf.Metric {
|
||||
metric, _ := metric.New("m1",
|
||||
map[string]string{"metric_tag": "from_metric"},
|
||||
map[string]interface{}{"value": int64(1)},
|
||||
time.Now(),
|
||||
)
|
||||
return metric
|
||||
}
|
||||
|
||||
func calculateProcessedTags(processor Override, metric telegraf.Metric) map[string]string {
|
||||
processed := processor.Apply(metric)
|
||||
return processed[0].Tags()
|
||||
}
|
||||
|
||||
func TestRetainsTags(t *testing.T) {
|
||||
processor := Override{}
|
||||
|
||||
tags := calculateProcessedTags(processor, createTestMetric())
|
||||
|
||||
value, present := tags["metric_tag"]
|
||||
assert.True(t, present, "Tag of metric was not present")
|
||||
assert.Equal(t, "from_metric", value, "Value of Tag was changed")
|
||||
}
|
||||
|
||||
func TestAddTags(t *testing.T) {
|
||||
processor := Override{Tags: map[string]string{"added_tag": "from_config", "another_tag": ""}}
|
||||
|
||||
tags := calculateProcessedTags(processor, createTestMetric())
|
||||
|
||||
value, present := tags["added_tag"]
|
||||
assert.True(t, present, "Additional Tag of metric was not present")
|
||||
assert.Equal(t, "from_config", value, "Value of Tag was changed")
|
||||
assert.Equal(t, 3, len(tags), "Should have one previous and two added tags.")
|
||||
}
|
||||
|
||||
func TestOverwritesPresentTagValues(t *testing.T) {
|
||||
processor := Override{Tags: map[string]string{"metric_tag": "from_config"}}
|
||||
|
||||
tags := calculateProcessedTags(processor, createTestMetric())
|
||||
|
||||
value, present := tags["metric_tag"]
|
||||
assert.True(t, present, "Tag of metric was not present")
|
||||
assert.Equal(t, 1, len(tags), "Should only have one tag.")
|
||||
assert.Equal(t, "from_config", value, "Value of Tag was not changed")
|
||||
}
|
||||
|
||||
func TestOverridesName(t *testing.T) {
|
||||
processor := Override{NameOverride: "overridden"}
|
||||
|
||||
processed := processor.Apply(createTestMetric())
|
||||
|
||||
assert.Equal(t, "overridden", processed[0].Name(), "Name was not overridden")
|
||||
}
|
||||
|
||||
func TestNamePrefix(t *testing.T) {
|
||||
processor := Override{NamePrefix: "Pre-"}
|
||||
|
||||
processed := processor.Apply(createTestMetric())
|
||||
|
||||
assert.Equal(t, "Pre-m1", processed[0].Name(), "Prefix was not applied")
|
||||
}
|
||||
|
||||
func TestNameSuffix(t *testing.T) {
|
||||
processor := Override{NameSuffix: "-suff"}
|
||||
|
||||
processed := processor.Apply(createTestMetric())
|
||||
|
||||
assert.Equal(t, "m1-suff", processed[0].Name(), "Suffix was not applied")
|
||||
}
|
Loading…
Reference in New Issue