feat：扩展 metricsfilter processor 的 drop action 过滤功能，支持额外条件过滤

Author: AIvanzqx

pkg/collector/processor/metricsfilter/config.go

@@ -59,7 +59,29 @@ func (c *Config) Validate() error {
 }
 
 type DropAction struct {
-	Metrics []string `config:"metrics" mapstructure:"metrics"`
+	Metrics []string    `config:"metrics" mapstructure:"metrics"`
+	Op      Op          `config:"op" mapstructure:"op"`
+	Rules   []*DropRule `config:"extra_rules" mapstructure:"extra_rules"`
+}
+
+func (d *DropAction) MetricMatch(name string) bool {
+	switch d.Op {
+	case OpNotIn:
+		return !slices.Contains(d.Metrics, name)
+
+	default:
+		return slices.Contains(d.Metrics, name)
+	}
+}
+
+type DropRule struct {
+	PredicateKey string      `config:"predicate_key" mapstructure:"predicate_key"`
+	MatchConfig  MatchConfig `config:"match" mapstructure:"match"`
+}
+
+type MatchConfig struct {
+	Op    string `config:"op" mapstructure:"op"`
+	Value string `config:"value" mapstructure:"value"`
 }
 
 type ReplaceAction struct {

pkg/collector/processor/metricsfilter/factory.go

@@ -17,8 +17,10 @@ import (
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/confengine"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/define"
+	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/internal/fields"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/internal/foreach"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/internal/mapstructure"
+	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/internal/opmatch"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/internal/promlabels"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/processor"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/utils/logger"
@@ -140,18 +142,34 @@ func (p *metricsFilter) Process(record *define.Record) (*define.Record, error) {
 func (p *metricsFilter) dropAction(record *define.Record, config Config) {
 	switch record.RecordType {
 	case define.RecordMetrics:
-		for _, name := range config.Drop.Metrics {
-			pdMetrics := record.Data.(pmetric.Metrics)
-			pdMetrics.ResourceMetrics().RemoveIf(func(resourceMetrics pmetric.ResourceMetrics) bool {
-				resourceMetrics.ScopeMetrics().RemoveIf(func(scopeMetrics pmetric.ScopeMetrics) bool {
-					scopeMetrics.Metrics().RemoveIf(func(metric pmetric.Metric) bool {
-						return metric.Name() == name
-					})
-					return scopeMetrics.Metrics().Len() == 0
+		action := config.Drop
+		pdMetrics := record.Data.(pmetric.Metrics)
+		pdMetrics.ResourceMetrics().RemoveIf(func(resourceMetrics pmetric.ResourceMetrics) bool {
+			rs := resourceMetrics.Resource().Attributes()
+			resourceMetrics.ScopeMetrics().RemoveIf(func(scopeMetrics pmetric.ScopeMetrics) bool {
+				scopeMetrics.Metrics().RemoveIf(func(metric pmetric.Metric) bool {
+					ruleMatch := true
+					if len(action.Rules) > 0 {
+						for _, rule := range action.Rules {
+							ff, pk := fields.DecodeFieldFrom(rule.PredicateKey)
+							switch ff {
+							// TODO(aivan): 目前 predicateKey 暂时只支持 resource 后续可能会扩展
+							case fields.FieldFromResource:
+								rv, ok := rs.Get(pk)
+								if !ok || !opmatch.Match(rv.AsString(), rule.MatchConfig.Value, rule.MatchConfig.Op) {
+									ruleMatch = false
+								}
+							default:
+								logger.Errorf("unsupported field %s", rule.PredicateKey)
+							}
+						}
+					}
+					return action.MetricMatch(metric.Name()) && ruleMatch
 				})
-				return resourceMetrics.ScopeMetrics().Len() == 0
+				return scopeMetrics.Metrics().Len() == 0
 			})
-		}
+			return resourceMetrics.ScopeMetrics().Len() == 0
+		})
 	}
 }
 

pkg/collector/processor/metricsfilter/factory_test.go

@@ -138,24 +138,78 @@ func TestMetricsNoAction(t *testing.T) {
 }
 
 func TestMetricsDropAction(t *testing.T) {
-	content := `
+	t.Run("simple_in", func(t *testing.T) {
+		content := `
 processor:
    - name: "metrics_filter/drop"
      config:
        drop:
          metrics:
            - "my_metrics"
 `
-	factory := processor.MustCreateFactory(content, NewFactory)
+		factory := processor.MustCreateFactory(content, NewFactory)
 
-	record := define.Record{
-		RecordType: define.RecordMetrics,
-		Data:       makeMetricsRecord(1),
-	}
+		record := define.Record{
+			RecordType: define.RecordMetrics,
+			Data:       makeMetricsRecord(1),
+		}
 
-	testkits.MustProcess(t, factory, record)
-	metrics := record.Data.(pmetric.Metrics).ResourceMetrics()
-	assert.Equal(t, 0, metrics.Len())
+		testkits.MustProcess(t, factory, record)
+		metrics := record.Data.(pmetric.Metrics).ResourceMetrics()
+		assert.Equal(t, 0, metrics.Len())
+	})
+
+	t.Run("simple_notin", func(t *testing.T) {
+		content := `
+processor:
+   - name: "metrics_filter/drop"
+     config:
+       drop:
+         metrics:
+           - "my_metrics1"
+         op: "notin"
+`
+		factory := processor.MustCreateFactory(content, NewFactory)
+
+		record := define.Record{
+			RecordType: define.RecordMetrics,
+			Data:       makeMetricsRecord(1),
+		}
+
+		testkits.MustProcess(t, factory, record)
+		metrics := record.Data.(pmetric.Metrics).ResourceMetrics()
+		assert.Equal(t, 0, metrics.Len())
+	})
+
+	t.Run("complex_notin", func(t *testing.T) {
+		content := `
+processor:
+   - name: "metrics_filter/drop"
+     config:
+       drop:
+         metrics:
+           - "my_metrics111"
+         op: "notin"
+         extra_rules:
+           - predicate_key: "resource.telemetry.distro.name"
+             match:
+               op: "eq"
+               value: "opentelemetry-java-instrumentation"
+`
+		factory := processor.MustCreateFactory(content, NewFactory)
+
+		record := define.Record{
+			RecordType: define.RecordMetrics,
+			Data: makeMetricsRecordWith(
+				"my_metrics", 1, map[string]string{
+					"telemetry.distro.name": "opentelemetry-java-instrumentation",
+				}, map[string]string{}),
+		}
+
+		testkits.MustProcess(t, factory, record)
+		metrics := record.Data.(pmetric.Metrics).ResourceMetrics()
+		assert.Equal(t, 0, metrics.Len())
+	})
 }
 
 func TestMetricsReplaceAction(t *testing.T) {

pkg/collector/support-files/templates/linux/x86_64/etc/bk-collector-application.conf.tpl

@@ -86,6 +86,7 @@ default:
 {% if metrics_filter_config is defined %}
       - name: "{{ metrics_filter_config.name }}"
         config:
+          {%- if metrics_filter_config.code_relabel is defined %}
           code_relabel:
             {%- for item in metrics_filter_config.code_relabel %}
             - metrics: {{ item.metrics | tojson }}
@@ -103,6 +104,21 @@ default:
                 {%- endfor %}
               {%- endfor %}
             {%- endfor %}
+          {%- endif %}
+          {%- if metrics_filter_config.drop is defined %}
+          drop:
+            metrics: {{ metrics_filter_config.drop.metrics | tojson }}
+            op: {{metrics_filter_config.drop.get("op", "in")}}
+            {%- if metrics_filter_config.drop.extra_rules is defined %}
+            extra_rules:
+              {%- for item in metrics_filter_config.drop.get("extra_rules", []) %}
+              - predicate_key: '{{ item.predicate_key }}'
+                match:
+                  op: '{{ item.match.op }}'
+                  value: '{{ item.match.value }}'
+              {%- endfor %}
+            {%- endif %}
+          {%- endif %}
 {%- endif %}
 
 {% if db_slow_command_config is defined %}