Signalfx exporter: add translation rule to drop redundant metrics

exporter/signalfxexporter/factory_test.go

@@ -172,7 +172,7 @@ func TestCreateMetricsExporterWithDefaultTranslaitonRules(t *testing.T) {
 
 	// Validate that default translation rules are loaded
 	// Expected values has to be updated once default config changed
-	assert.Equal(t, 37, len(config.TranslationRules))
+	assert.Equal(t, 38, len(config.TranslationRules))
 	assert.Equal(t, translation.ActionRenameDimensionKeys, config.TranslationRules[0].Action)
 	assert.Equal(t, 32, len(config.TranslationRules[0].Mapping))
 }
@@ -642,18 +642,14 @@ func TestDefaultDiskTranslations(t *testing.T) {
 		m[pt.Metric] = l
 	}
 
-	dtPts := m["disk.total"]
-	require.Equal(t, 4, len(dtPts))
-	require.Equal(t, 4, len(dtPts[0].Dimensions))
+	_, ok := m["disk.total"]
+	require.False(t, ok)
 
-	dstPts := m["disk.summary_total"]
-	require.Equal(t, 1, len(dstPts))
-	require.Equal(t, 3, len(dstPts[0].Dimensions))
+	_, ok = m["disk.summary_total"]
+	require.False(t, ok)
 
-	utPts, ok := m["df_complex.used_total"]
-	require.True(t, ok)
-	require.Equal(t, 1, len(utPts))
-	require.Equal(t, 3, len(utPts[0].Dimensions))
+	_, ok = m["df_complex.used_total"]
+	require.False(t, ok)
 
 	du, ok := m["disk.utilization"]
 	require.True(t, ok)

exporter/signalfxexporter/translation/constants.go

@@ -360,5 +360,12 @@ translation_rules:
 - action: multiply_float
   scale_factors_float:
     memory.utilization: 100
+
+# remove redundant metrics
+- action: drop_metrics
+  metric_names:
+    df_complex.used_total: true
+    disk.summary_total: true
+    disk.total: true
 `
 )

exporter/signalfxexporter/translation/translator.go

@@ -29,7 +29,7 @@ type Action string
 
 const (
 	// ActionRenameDimensionKeys renames dimension keys using Rule.Mapping.
-	// The rule can be applied only to a particular metric if MetricName is provided,
+	// The rule can be applied only to particular metrics if MetricNames are provided,
 	// otherwise applied to all metrics.
 	ActionRenameDimensionKeys Action = "rename_dimension_keys"
 
@@ -109,6 +109,9 @@ const (
 	// new metric will also get any attributes of the 'memory.used' metric except for its value and metric name.
 	// Currently only integer inputs are handled and only division is supported.
 	ActionCalculateNewMetric Action = "calculate_new_metric"
+
+	// ActionDropMetrics drops datapoints with metric name defined in "metric_names".
+	ActionDropMetrics Action = "drop_metrics"
 )
 
 type MetricOperator string
@@ -177,8 +180,7 @@ type Rule struct {
 	// excluded by aggregation.
 	WithoutDimensions []string `mapstructure:"without_dimensions"`
 
-	// MetricName is used by "split_metric" translation rule to specify a name
-	// of a metric that will be split.
+	// MetricNames is used by "rename_dimension_keys" and "drop_metrics" translation rules.
 	MetricNames map[string]bool `mapstructure:"metric_names"`
 
 	Operand1Metric string         `mapstructure:"operand1_metric"`
@@ -281,6 +283,10 @@ func validateTranslationRules(rules []Rule) error {
 			if tr.Operator != MetricOperatorDivision {
 				return fmt.Errorf("invalid operator %q for %q translation rule", tr.Operator, tr.Action)
 			}
+		case ActionDropMetrics:
+			if len(tr.MetricNames) == 0 {
+				return fmt.Errorf(`field "metric_names" is required for %q translation rule`, tr.Action)
+			}
 
 		default:
 			return fmt.Errorf("unknown \"action\" value: %q", tr.Action)
@@ -405,6 +411,18 @@ func (mp *MetricTranslator) TranslateDataPoints(logger *zap.Logger, sfxDataPoint
 			}
 			aggregatedDps := aggregateDatapoints(dpsToAggregate, tr.WithoutDimensions, tr.AggregationMethod)
 			processedDataPoints = append(otherDps, aggregatedDps...)
+
+		case ActionDropMetrics:
+			resultSliceLen := 0
+			for i, dp := range processedDataPoints {
+				if match := tr.MetricNames[dp.Metric]; !match {
+					if resultSliceLen < i {
+						processedDataPoints[resultSliceLen] = dp
+					}
+					resultSliceLen++
+				}
+			}
+			processedDataPoints = processedDataPoints[:resultSliceLen]
 		}
 	}
 

exporter/signalfxexporter/translation/translator_test.go

@@ -439,6 +439,27 @@ func TestNewMetricTranslator(t *testing.T) {
 			wantError: `fields "metric_name", "operand1_metric", "operand2_metric", and "operator" ` +
 				`are required for "calculate_new_metric" translation rule`,
 		},
+		{
+			name: "drop_metrics_valid",
+			trs: []Rule{
+				{
+					Action:      ActionDropMetrics,
+					MetricNames: map[string]bool{"metric": true},
+				},
+			},
+			wantDimensionsMap: nil,
+			wantError:         "",
+		},
+		{
+			name: "drop_metrics_invalid",
+			trs: []Rule{
+				{
+					Action: ActionDropMetrics,
+				},
+			},
+			wantDimensionsMap: nil,
+			wantError:         `field "metric_names" is required for "drop_metrics" translation rule`,
+		},
 	}
 
 	for _, tt := range tests {
@@ -1474,6 +1495,54 @@ func TestTranslateDataPoints(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "drop_metrics",
+			trs: []Rule{
+				{
+					Action: ActionDropMetrics,
+					MetricNames: map[string]bool{
+						"metric1": true,
+						"metric2": true,
+					},
+				},
+			},
+			dps: []*sfxpb.DataPoint{
+				{
+					Metric:     "metric1",
+					Timestamp:  msec,
+					MetricType: &gaugeType,
+					Value: sfxpb.Datum{
+						DoubleValue: generateFloatPtr(1.2),
+					},
+				},
+				{
+					Metric:     "metric2",
+					Timestamp:  msec,
+					MetricType: &gaugeType,
+					Value: sfxpb.Datum{
+						DoubleValue: generateFloatPtr(2.2),
+					},
+				},
+				{
+					Metric:     "metric3",
+					Timestamp:  msec,
+					MetricType: &gaugeType,
+					Value: sfxpb.Datum{
+						DoubleValue: generateFloatPtr(2.2),
+					},
+				},
+			},
+			want: []*sfxpb.DataPoint{
+				{
+					Metric:     "metric3",
+					Timestamp:  msec,
+					MetricType: &gaugeType,
+					Value: sfxpb.Datum{
+						DoubleValue: generateFloatPtr(2.2),
+					},
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {