add add_tags option to SinkConfig

config.go

@@ -98,4 +98,5 @@ type SinkConfig struct {
 	MaxTagLength  int                  `yaml:"max_tag_length"`
 	MaxTags       int                  `yaml:"max_tags"`
 	StripTags     []matcher.TagMatcher `yaml:"strip_tags"`
+	AddTags       map[string]string    `yaml:"extra_tags"`
 }

flusher.go

@@ -154,6 +154,7 @@ func (s *Server) flushSink(
 				maxNameLengthCount += 1
 				continue metricLoop
 			}
+
 			filteredTags := []string{}
 			if len(sink.stripTags) == 0 && sink.maxTagLength == 0 {
 				filteredTags = metric.Tags
@@ -175,6 +176,16 @@ func (s *Server) flushSink(
 					filteredTags = append(filteredTags, tag)
 				}
 			}
+
+			for k, v := range sink.addTags {
+				tag := fmt.Sprintf("%s:%s", k, v)
+				if sink.maxTagLength != 0 && len(tag) > sink.maxTagLength {
+					maxTagLengthCount += 1
+					continue metricLoop
+				}
+				filteredTags = append(filteredTags, tag)
+			}
+
 			if sink.maxTags != 0 && len(filteredTags) > sink.maxTags {
 				s.Statsd.Count("dropped_metrics", 1, []string{
 					sinkNameTag, sinkKindTag, "metric_name:" + metric.Name, "reason:max_tags", "veneurglobalonly:true",

flusher_test.go

@@ -378,7 +378,8 @@ func TestFlush(t *testing.T) {
 					matcher.CreateTagMatcher(&matcher.TagMatcherConfig{
 						Kind:  "prefix",
 						Value: "foo",
-					})},
+					}),
+				},
 			}},
 			StatsAddress: "localhost:8125",
 		},
@@ -763,6 +764,214 @@ func TestFlush(t *testing.T) {
 	})
 }
 
+func TestFlushWithAddTags(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	channel := make(chan []samplers.InterMetric)
+	mockStatsd := scopedstatsd.NewMockClient(ctrl)
+	server, err := NewFromConfig(ServerConfig{
+		Logger: logrus.New(),
+		Config: Config{
+			Debug: true,
+			Features: Features{
+				EnableMetricSinkRouting: true,
+			},
+			Hostname: "localhost",
+			Interval: DefaultFlushInterval,
+			MetricSinkRouting: []SinkRoutingConfig{{
+				Name: "default",
+				Match: []matcher.Matcher{{
+					Name: matcher.CreateNameMatcher(&matcher.NameMatcherConfig{
+						Kind: "any",
+					}),
+					Tags: []matcher.TagMatcher{},
+				}},
+				Sinks: SinkRoutingSinks{
+					Matched: []string{"channel"},
+				},
+			}},
+			MetricSinks: []SinkConfig{{
+				Kind:    "channel",
+				Name:    "channel",
+				AddTags: map[string]string{"foo": "bar"},
+				StripTags: []matcher.TagMatcher{
+					matcher.CreateTagMatcher(&matcher.TagMatcherConfig{
+						Kind:  "prefix",
+						Value: "foo",
+					}),
+				},
+			}},
+			StatsAddress: "localhost:8125",
+		},
+		MetricSinkTypes: MetricSinkTypes{
+			"channel": {
+				Create: func(
+					server *Server, name string, logger *logrus.Entry, config Config,
+					sinkConfig MetricSinkConfig,
+				) (sinks.MetricSink, error) {
+					sink, err := NewChannelMetricSink(channel)
+					if err != nil {
+						return nil, err
+					}
+					return sink, nil
+				},
+				ParseConfig: func(
+					name string, config interface{},
+				) (MetricSinkConfig, error) {
+					return nil, nil
+				},
+			},
+		},
+	})
+	assert.NoError(t, err)
+	server.Statsd = mockStatsd
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		server.Start()
+		wg.Done()
+	}()
+	defer func() {
+		server.Shutdown()
+		wg.Wait()
+	}()
+
+	mockStatsd.EXPECT().
+		Count(
+			gomock.Not("flushed_metrics"), gomock.All(), gomock.All(), gomock.All()).
+		AnyTimes()
+	mockStatsd.EXPECT().
+		Gauge(
+			gomock.Not("flushed_metrics"), gomock.All(), gomock.All(), gomock.All()).
+		AnyTimes()
+	mockStatsd.EXPECT().
+		Timing(
+			gomock.Not("flushed_metrics"), gomock.All(), gomock.All(), gomock.All()).
+		AnyTimes()
+
+	t.Run("WithAddTags", func(t *testing.T) {
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:skipped",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:max_name_length",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:max_tags",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:max_tag_length",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(1), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:flushed",
+			"veneurglobalonly:true",
+		}, 1.0)
+
+		server.Workers[0].PacketChan <- samplers.UDPMetric{
+			MetricKey: samplers.MetricKey{
+				Name: "test.metric",
+				Type: "counter",
+			},
+			Digest:     0,
+			Scope:      samplers.LocalOnly,
+			Tags:       []string{},
+			Value:      1.0,
+			SampleRate: 1.0,
+		}
+
+		result := <-channel
+		if assert.Len(t, result, 1) {
+			assert.Equal(t, "test.metric", result[0].Name)
+			if assert.Len(t, result[0].Tags, 1) {
+				assert.Equal(t, "foo:bar", result[0].Tags[0])
+			}
+		}
+
+		server.Workers[0].PacketChan <- samplers.UDPMetric{
+			MetricKey: samplers.MetricKey{
+				Name:       "test.metric",
+				Type:       "counter",
+				JoinedTags: "foo:not_bar",
+			},
+			Digest:     0,
+			Scope:      samplers.LocalOnly,
+			Tags:       []string{"foo:not_bar"},
+			Value:      1.0,
+			SampleRate: 1.0,
+		}
+	})
+
+	t.Run("WithAddTagsAndStripTags", func(t *testing.T) {
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:skipped",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:max_name_length",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:max_tags",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(0), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:max_tag_length",
+			"veneurglobalonly:true",
+		}, 1.0)
+		mockStatsd.EXPECT().Count("flushed_metrics", int64(1), []string{
+			"sink_name:channel",
+			"sink_kind:channel",
+			"status:flushed",
+			"veneurglobalonly:true",
+		}, 1.0)
+
+		server.Workers[0].PacketChan <- samplers.UDPMetric{
+			MetricKey: samplers.MetricKey{
+				Name:       "test.metric",
+				Type:       "counter",
+				JoinedTags: "foo:not_bar",
+			},
+			Digest:     0,
+			Scope:      samplers.LocalOnly,
+			Tags:       []string{"foo:not_bar"},
+			Value:      1.0,
+			SampleRate: 1.0,
+		}
+
+		result := <-channel
+		if assert.Len(t, result, 1) {
+			assert.Equal(t, "test.metric", result[0].Name)
+			if assert.Len(t, result[0].Tags, 1) {
+				assert.Equal(t, "foo:bar", result[0].Tags[0])
+			}
+		}
+	})
+}
+
 type anyOfMatcher struct {
 	s []string
 }

server.go

@@ -185,6 +185,7 @@ type internalMetricSink struct {
 	maxTagLength  int
 	maxTags       int
 	stripTags     []matcher.TagMatcher
+	addTags       map[string]string
 }
 
 type GlobalListeningPerProtocolMetrics struct {
@@ -443,6 +444,7 @@ func (server *Server) createMetricSinks(
 			maxTagLength:  sinkConfig.MaxTagLength,
 			maxTags:       sinkConfig.MaxTags,
 			stripTags:     sinkConfig.StripTags,
+			addTags:       sinkConfig.AddTags,
 		})
 	}
 	return sinks, nil