Signalfx exporter: add filtering option to copy_metrics translation rule

exporter/signalfxexporter/go.sum

@@ -1367,6 +1367,7 @@ google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
+google.golang.org/protobuf v1.24.0 h1:UhZDfRO8JRQru4/+LlLE0BRKGF8L+PICnvYZmx/fEGA=
 google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGjtUeSXeh4=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

exporter/signalfxexporter/translation/translator.go

@@ -48,7 +48,10 @@ const (
 	// Rule.TypesMapping key/values as metric_name/new_type.
 	ActionConvertValues Action = "convert_values"
 
-	// ActionCopyMetrics copies metrics using Rule.Mapping
+	// ActionCopyMetrics copies metrics using Rule.Mapping.
+	// Rule.DimensionKey and Rule.DimensionValues can be used to filter datapoints that must be copied,
+	// if these fields are set, only metics having a dimension with key == Rule.DimensionKey and
+	// value in Rule.DimensionValues will be copied.
 	ActionCopyMetrics Action = "copy_metrics"
 
 	// ActionSplitMetric splits a metric with Rule.MetricName into multiple metrics
@@ -129,8 +132,13 @@ type Rule struct {
 	// DimensionKey is used by "split_metric" translation rule action to specify dimension key
 	// that will be used to translate the metric datapoints. Datapoints that don't have
 	// the specified dimension key will not be translated.
+	// DimensionKey is also used by "copy_metrics" for filterring.
 	DimensionKey string `mapstructure:"dimension_key"`
 
+	// DimensionValues is used by "copy_metrics" to filter out datapoints with dimensions values
+	// not matching values set in this field
+	DimensionValues map[string]bool `mapstructure:"dimension_values"`
+
 	// TypesMapping is represents metric_name/metric_type key/value pairs,
 	// used by ActionConvertValues.
 	TypesMapping map[string]MetricValueType `mapstructure:"types_mapping"`
@@ -200,6 +208,11 @@ func validateTranslationRules(rules []Rule) error {
 			if tr.Mapping == nil {
 				return fmt.Errorf("field \"mapping\" is required for %q translation rule", tr.Action)
 			}
+			if tr.DimensionKey != "" && len(tr.DimensionValues) == 0 {
+				return fmt.Errorf(
+					"\"dimension_values_filer\" has to be provided if \"dimension_key\" is set for %q translation rule",
+					tr.Action)
+			}
 		case ActionSplitMetric:
 			if tr.MetricName == "" || tr.DimensionKey == "" || tr.Mapping == nil {
 				return fmt.Errorf(
@@ -293,15 +306,14 @@ func (mp *MetricTranslator) TranslateDataPoints(logger *zap.Logger, sfxDataPoint
 				}
 			}
 		case ActionCopyMetrics:
-			newDataPoints := []*sfxpb.DataPoint{}
 			for _, dp := range processedDataPoints {
 				if newMetric, ok := tr.Mapping[dp.Metric]; ok {
-					newDataPoint := proto.Clone(dp).(*sfxpb.DataPoint)
-					newDataPoint.Metric = newMetric
-					newDataPoints = append(newDataPoints, newDataPoint)
+					newDataPoint := copyMetric(tr, dp, newMetric)
+					if newDataPoint != nil {
+						processedDataPoints = append(processedDataPoints, newDataPoint)
+					}
 				}
 			}
-			processedDataPoints = append(processedDataPoints, newDataPoints...)
 		case ActionSplitMetric:
 			for _, dp := range processedDataPoints {
 				if tr.MetricName == dp.Metric {
@@ -490,3 +502,22 @@ func convertMetricValue(logger *zap.Logger, dp *sfxpb.DataPoint, newType MetricV
 		dp.Value = sfxpb.Datum{DoubleValue: &floatVal}
 	}
 }
+
+func copyMetric(tr Rule, dp *sfxpb.DataPoint, newMetricName string) *sfxpb.DataPoint {
+	if tr.DimensionKey != "" {
+		var match bool
+		for _, d := range dp.Dimensions {
+			if d.Key == tr.DimensionKey {
+				match = tr.DimensionValues[d.Value]
+				break
+			}
+		}
+		if !match {
+			return nil
+		}
+	}
+
+	newDataPoint := proto.Clone(dp).(*sfxpb.DataPoint)
+	newDataPoint.Metric = newMetricName
+	return newDataPoint
+}

exporter/signalfxexporter/translation/translator_test.go

@@ -241,7 +241,7 @@ func TestNewMetricTranslator(t *testing.T) {
 			wantError:         "",
 		},
 		{
-			name: "copy_metric_invalid",
+			name: "copy_metric_invalid_no_mapping",
 			trs: []Rule{
 				{
 					Action: ActionCopyMetrics,
@@ -250,6 +250,21 @@ func TestNewMetricTranslator(t *testing.T) {
 			wantDimensionsMap: nil,
 			wantError:         "field \"mapping\" is required for \"copy_metrics\" translation rule",
 		},
+		{
+			name: "copy_metric_invalid_no_dimensions_filter",
+			trs: []Rule{
+				{
+					Action: ActionCopyMetrics,
+					Mapping: map[string]string{
+						"metric1": "metric2",
+					},
+					DimensionKey: "dim1",
+				},
+			},
+			wantDimensionsMap: nil,
+			wantError: "\"dimension_values_filer\" has to be provided if \"dimension_key\" is set " +
+				"for \"copy_metrics\" translation rule",
+		},
 		{
 			name: "split_metric_valid",
 			trs: []Rule{
@@ -616,6 +631,133 @@ func TestTranslateDataPoints(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "copy_with_dimension_filter",
+			trs: []Rule{
+				{
+					Action: ActionCopyMetrics,
+					Mapping: map[string]string{
+						"metric1": "metric2",
+					},
+					DimensionKey: "dim1",
+					DimensionValues: map[string]bool{
+						"val1": true,
+						"val2": true,
+					},
+				},
+			},
+			dps: []*sfxpb.DataPoint{
+				{
+					Metric:    "metric1",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val1",
+						},
+					},
+				},
+				{
+					Metric:    "metric1",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val2",
+						},
+					},
+				},
+				// must not be copied
+				{
+					Metric:    "metric1",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					MetricType: &gaugeType,
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val3",
+						},
+					},
+				},
+			},
+			want: []*sfxpb.DataPoint{
+				{
+					Metric:    "metric1",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val1",
+						},
+					},
+				},
+				{
+					Metric:    "metric1",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val2",
+						},
+					},
+				},
+				{
+					Metric:    "metric1",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					MetricType: &gaugeType,
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val3",
+						},
+					},
+				},
+				{
+					Metric:    "metric2",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val1",
+						},
+					},
+				},
+				{
+					Metric:    "metric2",
+					Timestamp: msec,
+					Value: sfxpb.Datum{
+						IntValue: generateIntPtr(9),
+					},
+					Dimensions: []*sfxpb.Dimension{
+						{
+							Key:   "dim1",
+							Value: "val2",
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "copy_and_rename",
 			trs: []Rule{