feat(processors.filter): Convert noop processor to filter processor (#14330)
This commit is contained in:
Sven Rebhan
GitHub
parent
7b5393c9e5
commit
1c2c03d778
|
|
@ -0,0 +1,5 @@
|
|||
//go:build !custom || processors || processors.filter
|
||||
|
||||
package all
|
||||
|
||||
import _ "github.com/influxdata/telegraf/plugins/processors/filter" // register plugin
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
//go:build !custom || processors || processors.noop
|
||||
|
||||
package all
|
||||
|
||||
import _ "github.com/influxdata/telegraf/plugins/processors/noop" // register plugin
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
# Filter Processor Plugin
|
||||
|
||||
The filter processor plugin allows to specify a set of rules for metrics
|
||||
with the ability to _keep_ or _drop_ those metrics. It does _not_ change the
|
||||
metric. As such a user might want to apply this processor to remove metrics
|
||||
from the processing/output stream.
|
||||
__NOTE:__ The filtering is _not_ output specific, but will apply to the metrics
|
||||
processed by this processor.
|
||||
|
||||
## Global configuration options <!-- @/docs/includes/plugin_config.md -->
|
||||
|
||||
In addition to the plugin-specific configuration settings, plugins support
|
||||
additional global and plugin configuration settings. These settings are used to
|
||||
modify metrics, tags, and field or create aliases and configure ordering, etc.
|
||||
See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
|
||||
|
||||
[CONFIGURATION.md]: ../../../docs/CONFIGURATION.md#plugins
|
||||
|
||||
## Configuration
|
||||
|
||||
```toml @sample.conf
|
||||
# Filter metrics by the given criteria
|
||||
[[processors.filter]]
|
||||
## Default action if no rule applies
|
||||
# default = "pass"
|
||||
|
||||
## Rules to apply on the incoming metrics (multiple rules are possible)
|
||||
## The rules are evaluated in order and the first matching rule is applied.
|
||||
## In case no rule matches the "default" is applied.
|
||||
## All filter criteria in a rule must apply for the rule to match the metric
|
||||
## i.e. the criteria are combined by a logical AND. If a criterion is
|
||||
## omitted it is NOT applied at all and ignored.
|
||||
[[processors.filter.rule]]
|
||||
## List of metric names to match including glob expressions
|
||||
# name = []
|
||||
|
||||
## List of tag key/values pairs to match including glob expressions
|
||||
## ALL given tags keys must exist and at least one value must match
|
||||
## for the metric to match the rule.
|
||||
# tags = {}
|
||||
|
||||
## List of field keys to match including glob expressions
|
||||
## At least one field must exist for the metric to match the rule.
|
||||
# fields = []
|
||||
|
||||
## Action to apply for this rule
|
||||
## "pass" will keep the metric and pass it on, while "drop" will remove
|
||||
## the metric
|
||||
# action = "drop"
|
||||
```
|
||||
|
||||
## Examples
|
||||
|
||||
Consider a use-case where you collected a bunch of metrics
|
||||
|
||||
```text
|
||||
machine,source="machine1",status="OK" operating_hours=37i,temperature=23.1
|
||||
machine,source="machine2",status="warning" operating_hours=1433i,temperature=48.9,message="too hot"
|
||||
machine,source="machine3",status="OK" operating_hours=811i,temperature=29.5
|
||||
machine,source="machine4",status="failure" operating_hours=1009i,temperature=67.3,message="temperature alert"
|
||||
```
|
||||
|
||||
but only want to keep the ones indicating a `status` of `failure` or `warning`:
|
||||
|
||||
```toml
|
||||
[[processors.filter]]
|
||||
namepass = ["machine"]
|
||||
default = "drop"
|
||||
|
||||
[[processors.filter.rule]]
|
||||
tags = {"status" = ["warning", "failure"]}
|
||||
action = "pass"
|
||||
```
|
||||
|
||||
Alternatively, you can "black-list" the `OK` value via
|
||||
|
||||
```toml
|
||||
[[processors.filter]]
|
||||
namepass = ["machine"]
|
||||
|
||||
[[processors.filter.rule]]
|
||||
tags = {"status" = "OK"}
|
||||
```
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
//go:generate ../../../tools/readme_config_includer/generator
|
||||
package filter
|
||||
|
||||
import (
|
||||
_ "embed"
|
||||
"fmt"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/plugins/processors"
|
||||
)
|
||||
|
||||
//go:embed sample.conf
|
||||
var sampleConfig string
|
||||
|
||||
type Filter struct {
|
||||
Rules []rule `toml:"rule"`
|
||||
DefaultAction string `toml:"default"`
|
||||
Log telegraf.Logger `toml:"-"`
|
||||
defaultPass bool
|
||||
}
|
||||
|
||||
func (*Filter) SampleConfig() string {
|
||||
return sampleConfig
|
||||
}
|
||||
|
||||
func (f *Filter) Init() error {
|
||||
// Check the default-action setting
|
||||
switch f.DefaultAction {
|
||||
case "", "pass":
|
||||
f.defaultPass = true
|
||||
case "drop":
|
||||
// Do nothing, those options are valid
|
||||
if len(f.Rules) == 0 {
|
||||
f.Log.Warn("dropping all metrics as no rule is provided")
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("invalid default action %q", f.DefaultAction)
|
||||
}
|
||||
|
||||
// Check and initialize rules
|
||||
for i := range f.Rules {
|
||||
if err := f.Rules[i].init(); err != nil {
|
||||
return fmt.Errorf("initialization of rule %d failed: %w", i+1, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *Filter) Apply(in ...telegraf.Metric) []telegraf.Metric {
|
||||
out := make([]telegraf.Metric, 0, len(in))
|
||||
for _, m := range in {
|
||||
if f.applyRules(m) {
|
||||
out = append(out, m)
|
||||
} else {
|
||||
m.Drop()
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (f *Filter) applyRules(m telegraf.Metric) bool {
|
||||
for _, r := range f.Rules {
|
||||
if pass, applies := r.apply(m); applies {
|
||||
return pass
|
||||
}
|
||||
}
|
||||
return f.defaultPass
|
||||
}
|
||||
|
||||
func init() {
|
||||
processors.Add("Filter", func() telegraf.Processor {
|
||||
return &Filter{}
|
||||
})
|
||||
}
|
||||
|
|
@ -0,0 +1,661 @@
|
|||
package filter
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/metric"
|
||||
"github.com/influxdata/telegraf/testutil"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
var testmetrics = []telegraf.Metric{
|
||||
metric.New(
|
||||
"packing",
|
||||
map[string]string{
|
||||
"source": "machine A",
|
||||
"location": "main building",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 37,
|
||||
"temperature": 23.1,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine C",
|
||||
"location": "factory X",
|
||||
"status": "failure",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1009,
|
||||
"temperature": 67.3,
|
||||
"message": "temperature alert",
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine D",
|
||||
"location": "factory Y",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 825,
|
||||
"temperature": 31.2,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
|
||||
func TestNoRules(t *testing.T) {
|
||||
logger := &testutil.CaptureLogger{}
|
||||
plugin := &Filter{
|
||||
DefaultAction: "drop",
|
||||
Log: logger,
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
warnings := logger.Warnings()
|
||||
require.Len(t, warnings, 1)
|
||||
require.Contains(t, warnings[0], "dropping all metrics")
|
||||
}
|
||||
|
||||
func TestInvalidDefaultAction(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{{Name: []string{"foo"}}},
|
||||
DefaultAction: "foo",
|
||||
}
|
||||
require.ErrorContains(t, plugin.Init(), "invalid default action")
|
||||
}
|
||||
|
||||
func TestNoMetric(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{{Name: []string{"*"}}},
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
input := []telegraf.Metric{}
|
||||
require.Empty(t, plugin.Apply(input...))
|
||||
}
|
||||
|
||||
func TestDropAll(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{{Name: []string{"*"}}},
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
require.Empty(t, plugin.Apply(testmetrics...))
|
||||
}
|
||||
|
||||
func TestDropDefault(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{{Name: []string{"foo"}, Action: "pass"}},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
require.Empty(t, plugin.Apply(testmetrics...))
|
||||
}
|
||||
|
||||
func TestPassAll(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{{Name: []string{"*"}, Action: "pass"}},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := testmetrics
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestPassDefault(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{{Name: []string{"foo"}, Action: "drop"}},
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := testmetrics
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestNamePass(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Name: []string{"welding"},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine C",
|
||||
"location": "factory X",
|
||||
"status": "failure",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1009,
|
||||
"temperature": 67.3,
|
||||
"message": "temperature alert",
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine D",
|
||||
"location": "factory Y",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 825,
|
||||
"temperature": 31.2,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestNameDrop(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Name: []string{"welding"},
|
||||
Action: "drop",
|
||||
},
|
||||
},
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"packing",
|
||||
map[string]string{
|
||||
"source": "machine A",
|
||||
"location": "main building",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 37,
|
||||
"temperature": 23.1,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestNameGlob(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Name: []string{"*ing"},
|
||||
Action: "drop",
|
||||
},
|
||||
},
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestTagPass(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Tags: map[string][]string{"status": {"OK"}},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"packing",
|
||||
map[string]string{
|
||||
"source": "machine A",
|
||||
"location": "main building",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 37,
|
||||
"temperature": 23.1,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine D",
|
||||
"location": "factory Y",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 825,
|
||||
"temperature": 31.2,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestTagDrop(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Tags: map[string][]string{"status": {"OK"}},
|
||||
Action: "drop",
|
||||
},
|
||||
},
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine C",
|
||||
"location": "factory X",
|
||||
"status": "failure",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1009,
|
||||
"temperature": 67.3,
|
||||
"message": "temperature alert",
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestTagMultiple(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Tags: map[string][]string{
|
||||
"location": {"factory X", "factory Y"},
|
||||
"status": {"OK"},
|
||||
},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine D",
|
||||
"location": "factory Y",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 825,
|
||||
"temperature": 31.2,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestTagGlob(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Tags: map[string][]string{"location": {"factory *"}},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine C",
|
||||
"location": "factory X",
|
||||
"status": "failure",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1009,
|
||||
"temperature": 67.3,
|
||||
"message": "temperature alert",
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine D",
|
||||
"location": "factory Y",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 825,
|
||||
"temperature": 31.2,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestTagDoesNotExist(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Tags: map[string][]string{
|
||||
"operator": {"peter"},
|
||||
"status": {"OK"},
|
||||
},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
require.Empty(t, plugin.Apply(testmetrics...))
|
||||
}
|
||||
|
||||
func TestFieldPass(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Fields: []string{"message", "pieces"},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine C",
|
||||
"location": "factory X",
|
||||
"status": "failure",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1009,
|
||||
"temperature": 67.3,
|
||||
"message": "temperature alert",
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestFieldDrop(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Fields: []string{"message", "pieces"},
|
||||
Action: "drop",
|
||||
},
|
||||
},
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"packing",
|
||||
map[string]string{
|
||||
"source": "machine A",
|
||||
"location": "main building",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 37,
|
||||
"temperature": 23.1,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine D",
|
||||
"location": "factory Y",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 825,
|
||||
"temperature": 31.2,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestFieldGlob(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Fields: []string{"{message,piece*}"},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"foundry",
|
||||
map[string]string{
|
||||
"source": "machine B",
|
||||
"location": "factory X",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1337,
|
||||
"temperature": 19.9,
|
||||
"pieces": 96878,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
metric.New(
|
||||
"welding",
|
||||
map[string]string{
|
||||
"source": "machine C",
|
||||
"location": "factory X",
|
||||
"status": "failure",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 1009,
|
||||
"temperature": 67.3,
|
||||
"message": "temperature alert",
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
||||
func TestRuleOrder(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Name: []string{"welding"},
|
||||
Action: "drop",
|
||||
},
|
||||
{
|
||||
Name: []string{"welding"},
|
||||
Action: "pass",
|
||||
},
|
||||
},
|
||||
DefaultAction: "drop",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
require.Empty(t, plugin.Apply(testmetrics...))
|
||||
}
|
||||
|
||||
func TestRuleMultiple(t *testing.T) {
|
||||
plugin := &Filter{
|
||||
Rules: []rule{
|
||||
{
|
||||
Name: []string{"welding"},
|
||||
Action: "drop",
|
||||
},
|
||||
{
|
||||
Name: []string{"foundry"},
|
||||
Action: "drop",
|
||||
},
|
||||
},
|
||||
DefaultAction: "pass",
|
||||
}
|
||||
require.NoError(t, plugin.Init())
|
||||
|
||||
expected := []telegraf.Metric{
|
||||
metric.New(
|
||||
"packing",
|
||||
map[string]string{
|
||||
"source": "machine A",
|
||||
"location": "main building",
|
||||
"status": "OK",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"operating_hours": 37,
|
||||
"temperature": 23.1,
|
||||
},
|
||||
time.Unix(0, 0),
|
||||
),
|
||||
}
|
||||
actual := plugin.Apply(testmetrics...)
|
||||
testutil.RequireMetricsEqual(t, expected, actual)
|
||||
}
|
||||
|
|
@ -0,0 +1,87 @@
|
|||
package filter
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/filter"
|
||||
)
|
||||
|
||||
type rule struct {
|
||||
Name []string `toml:"name"`
|
||||
Tags map[string][]string `toml:"tags"`
|
||||
Fields []string `toml:"fields"`
|
||||
Action string `toml:"action"`
|
||||
|
||||
nameFilter filter.Filter
|
||||
fieldFilter filter.Filter
|
||||
tagFilters map[string]filter.Filter
|
||||
pass bool
|
||||
}
|
||||
|
||||
func (r *rule) init() error {
|
||||
// Check the action setting
|
||||
switch r.Action {
|
||||
case "pass":
|
||||
r.pass = true
|
||||
case "", "drop":
|
||||
// Do nothing, those options are valid
|
||||
default:
|
||||
return fmt.Errorf("invalid action %q", r.Action)
|
||||
}
|
||||
|
||||
// Compile the filters
|
||||
var err error
|
||||
r.nameFilter, err = filter.Compile(r.Name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating name filter failed: %w", err)
|
||||
}
|
||||
|
||||
r.fieldFilter, err = filter.Compile(r.Fields)
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating fields filter failed: %w", err)
|
||||
}
|
||||
|
||||
r.tagFilters = make(map[string]filter.Filter, len(r.Tags))
|
||||
for k, values := range r.Tags {
|
||||
r.tagFilters[k], err = filter.Compile(values)
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating tag filter for tag %q failed: %w", k, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *rule) apply(m telegraf.Metric) (pass, applies bool) {
|
||||
// Check the metric name
|
||||
if r.nameFilter != nil {
|
||||
if !r.nameFilter.Match(m.Name()) {
|
||||
return true, false
|
||||
}
|
||||
}
|
||||
|
||||
// Check the tags if given
|
||||
tags := m.Tags()
|
||||
for k, f := range r.tagFilters {
|
||||
if value, found := tags[k]; !found || !f.Match(value) {
|
||||
return true, false
|
||||
}
|
||||
}
|
||||
|
||||
// Check the field names
|
||||
if r.fieldFilter != nil {
|
||||
var matches bool
|
||||
for _, field := range m.FieldList() {
|
||||
if r.fieldFilter.Match(field.Key) {
|
||||
matches = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !matches {
|
||||
return true, false
|
||||
}
|
||||
}
|
||||
|
||||
return r.pass, true
|
||||
}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
# Filter metrics by the given criteria
|
||||
[[processors.filter]]
|
||||
## Default action if no rule applies
|
||||
# default = "pass"
|
||||
|
||||
## Rules to apply on the incoming metrics (multiple rules are possible)
|
||||
## The rules are evaluated in order and the first matching rule is applied.
|
||||
## In case no rule matches the "default" is applied.
|
||||
## All filter criteria in a rule must apply for the rule to match the metric
|
||||
## i.e. the criteria are combined by a logical AND. If a criterion is
|
||||
## omitted it is NOT applied at all and ignored.
|
||||
[[processors.filter.rule]]
|
||||
## List of metric names to match including glob expressions
|
||||
# name = []
|
||||
|
||||
## List of tag key/values pairs to match including glob expressions
|
||||
## ALL given tags keys must exist and at least one value must match
|
||||
## for the metric to match the rule.
|
||||
# tags = {}
|
||||
|
||||
## List of field keys to match including glob expressions
|
||||
## At least one field must exist for the metric to match the rule.
|
||||
# fields = []
|
||||
|
||||
## Action to apply for this rule
|
||||
## "pass" will keep the metric and pass it on, while "drop" will remove
|
||||
## the metric
|
||||
# action = "drop"
|
||||
|
|
@ -1,60 +0,0 @@
|
|||
# Noop Processor Plugin
|
||||
|
||||
The noop processor plugin does nothing to metrics. Instead it can be used to
|
||||
apply the global configuration options after other processing. Global config
|
||||
options like tagpass, fieldpass, and others are applied before a processor,
|
||||
aggregator, or output. As such a user might want to apply these after doing
|
||||
processing, but before an output or another processor.
|
||||
|
||||
## Global configuration options <!-- @/docs/includes/plugin_config.md -->
|
||||
|
||||
In addition to the plugin-specific configuration settings, plugins support
|
||||
additional global and plugin configuration settings. These settings are used to
|
||||
modify metrics, tags, and field or create aliases and configure ordering, etc.
|
||||
See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
|
||||
|
||||
[CONFIGURATION.md]: ../../../docs/CONFIGURATION.md#plugins
|
||||
|
||||
## Configuration
|
||||
|
||||
```toml @sample.conf
|
||||
# Do nothing processor
|
||||
[[processors.noop]]
|
||||
|
||||
## Metric Filtering
|
||||
## The following options provide mechanisms to include or exclude entire
|
||||
## metrics. For specific details and examples see the metric filtering docs:
|
||||
## https://github.com/influxdata/telegraf/blob/master/docs/CONFIGURATION.md#metric-filtering
|
||||
|
||||
## Metric Selectors - These will drop entire metrics
|
||||
## Filter on metric name or tag key + value
|
||||
# namepass = []
|
||||
# namedrop = []
|
||||
# tagpass = {}
|
||||
# tagdrop = {}
|
||||
|
||||
## Filter on Common Expression Language (CEL) expression
|
||||
# metricpass = ""
|
||||
|
||||
## Metric Modifiers - These will drop tags and fields from metrics
|
||||
## Filter on tag key or field key
|
||||
# taginclude = []
|
||||
# tagexclude = []
|
||||
# fieldpass = []
|
||||
# fielddrop = []
|
||||
```
|
||||
|
||||
## Examples
|
||||
|
||||
Consider a use-case where you have processed a metric based on a tag, but no
|
||||
longer need that tag for additional processing:
|
||||
|
||||
```toml
|
||||
[[processors.ifname]]
|
||||
order = 1
|
||||
...
|
||||
|
||||
[[processors.noop]]
|
||||
order = 2
|
||||
tagexclude = ["useless_tag"]
|
||||
```
|
||||
|
|
@ -1,28 +0,0 @@
|
|||
//go:generate ../../../tools/readme_config_includer/generator
|
||||
package noop
|
||||
|
||||
import (
|
||||
_ "embed"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/plugins/processors"
|
||||
)
|
||||
|
||||
//go:embed sample.conf
|
||||
var sampleConfig string
|
||||
|
||||
type Noop struct{}
|
||||
|
||||
func (*Noop) SampleConfig() string {
|
||||
return sampleConfig
|
||||
}
|
||||
|
||||
func (p *Noop) Apply(in ...telegraf.Metric) []telegraf.Metric {
|
||||
return in
|
||||
}
|
||||
|
||||
func init() {
|
||||
processors.Add("noop", func() telegraf.Processor {
|
||||
return &Noop{}
|
||||
})
|
||||
}
|
||||
|
|
@ -1,72 +0,0 @@
|
|||
package noop
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/influxdata/telegraf"
|
||||
"github.com/influxdata/telegraf/testutil"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestNoopNoMetric(t *testing.T) {
|
||||
processor := Noop{}
|
||||
|
||||
m := []telegraf.Metric{}
|
||||
actual := processor.Apply(m...)
|
||||
require.Empty(t, actual)
|
||||
testutil.RequireMetricsEqual(t, m, actual)
|
||||
}
|
||||
|
||||
func TestNoopSingleMetric(t *testing.T) {
|
||||
processor := Noop{}
|
||||
|
||||
m := []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"test",
|
||||
map[string]string{
|
||||
"tag": "tag_value",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"value": 42,
|
||||
},
|
||||
time.Now(),
|
||||
telegraf.Gauge,
|
||||
),
|
||||
}
|
||||
actual := processor.Apply(m...)
|
||||
require.Len(t, actual, 1)
|
||||
testutil.RequireMetricsEqual(t, m, actual)
|
||||
}
|
||||
|
||||
func TestNoopMultipleMetrics(t *testing.T) {
|
||||
processor := Noop{}
|
||||
|
||||
m := []telegraf.Metric{
|
||||
testutil.MustMetric(
|
||||
"test",
|
||||
map[string]string{
|
||||
"tag": "tag_value",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"value": 42,
|
||||
},
|
||||
time.Now(),
|
||||
telegraf.Gauge,
|
||||
),
|
||||
testutil.MustMetric(
|
||||
"test",
|
||||
map[string]string{
|
||||
"tag": "tag_value",
|
||||
},
|
||||
map[string]interface{}{
|
||||
"value": 42,
|
||||
},
|
||||
time.Now(),
|
||||
telegraf.Gauge,
|
||||
),
|
||||
}
|
||||
actual := processor.Apply(m...)
|
||||
require.Len(t, actual, 2)
|
||||
testutil.RequireMetricsEqual(t, m, actual)
|
||||
}
|
||||
|
|
@ -1,24 +0,0 @@
|
|||
# Do nothing processor
|
||||
[[processors.noop]]
|
||||
|
||||
## Metric Filtering
|
||||
## The following options provide mechanisms to include or exclude entire
|
||||
## metrics. For specific details and examples see the metric filtering docs:
|
||||
## https://github.com/influxdata/telegraf/blob/master/docs/CONFIGURATION.md#metric-filtering
|
||||
|
||||
## Metric Selectors - These will drop entire metrics
|
||||
## Filter on metric name or tag key + value
|
||||
# namepass = []
|
||||
# namedrop = []
|
||||
# tagpass = {}
|
||||
# tagdrop = {}
|
||||
|
||||
## Filter on Common Expression Language (CEL) expression
|
||||
# metricpass = ""
|
||||
|
||||
## Metric Modifiers - These will drop tags and fields from metrics
|
||||
## Filter on tag key or field key
|
||||
# taginclude = []
|
||||
# tagexclude = []
|
||||
# fieldpass = []
|
||||
# fielddrop = []
|
||||
Loading…
Reference in New Issue