chore(processors.filter): Convert noop processor to filter processor

plugins/processors/all/filter.go

@@ -0,0 +1,5 @@
+//go:build !custom || processors || processors.filter
+
+package all
+
+import _ "github.com/influxdata/telegraf/plugins/processors/filter" // register plugin

plugins/processors/filter/README.md

@@ -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"}
+```

plugins/processors/filter/filter.go

@@ -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{}
+	})
+}