Override common dimensions with signalfx_vary_key_by (#780)

* fix bug where varyBy no longer worked with commonDimensions

* remove unnecessary TODO comment

* add a warning for when no SignalFx client matching tag value exists

* add TestSignalFxVaryByOverride

Co-authored-by: Randall Ma <randall@randallma.com>

sinks/signalfx/signalfx.go

@@ -65,6 +65,7 @@ func (c *collection) addPoint(key string, point *datapoint.Datapoint) {
 			c.pointsByKey[key] = append(c.pointsByKey[key], point)
 			return
 		}
+		c.sink.log.Warn(fmt.Sprintf("No SignalFx client exists for tag value '%s', using default client", key))
 	}
 	c.points = append(c.points, point)
 }
@@ -431,25 +432,28 @@ METRICLOOP: // Convenience label so that inner nested loops and `continue` easil
 			}
 		}
 
-		// metric-specified API key, if present, should override the common dimension
 		metricKey := ""
-		metricVaryByOverride := false
 
+		// Metric-specified API key, if present, should override the common dimension
+		metricOverrodeVaryBy := false
 		if sfx.varyBy != "" {
-			if val, ok := dims[sfx.varyBy]; ok {
-				metricKey = val
-				metricVaryByOverride = true
+			if _, ok := dims[sfx.varyBy]; ok {
+				metricOverrodeVaryBy = true
 			}
 		}
 
-		// Copy common dimensions
+		// Copy common dimensions, except for sfx.varyBy
 		for k, v := range sfx.commonDimensions {
+			if metricOverrodeVaryBy && k == sfx.varyBy {
+				continue
+			}
 			dims[k] = v
 		}
 
-		// re-copy metric-specified API key, if present
-		if metricVaryByOverride {
-			dims[sfx.varyBy] = metricKey
+		if sfx.varyBy != "" {
+			if val, ok := dims[sfx.varyBy]; ok {
+				metricKey = val
+			}
 		}
 
 		for k := range sfx.excludedTags {
@@ -462,7 +466,6 @@ METRICLOOP: // Convenience label so that inner nested loops and `continue` easil
 		case samplers.GaugeMetric:
 			point = sfxclient.GaugeF(metric.Name, dims, metric.Value)
 		case samplers.CounterMetric:
-			// TODO I am not certain if this should be a Counter or a Cumulative
 			point = sfxclient.Counter(metric.Name, dims, int64(metric.Value))
 		case samplers.StatusMetric:
 			countStatusMetrics++

sinks/signalfx/signalfx_test.go

@@ -779,3 +779,43 @@ LOOP:
 	assert.Subset(t, actualPerTagClients, expectedPerTagClients, "The expected values should be a subset of the actual values")
 
 }
+
+func TestSignalFxVaryByOverride(t *testing.T) {
+	derived := newDerivedProcessor()
+
+	varyByTagKey := "vary_by"
+	commonDimensions := map[string]string{"vary_by": "bar"}
+	defaultFakeSink := NewFakeSink()
+	customFakeSinkFoo := NewFakeSink()
+	customFakeSinkBar := NewFakeSink()
+	perTagClients := make(map[string]DPClient)
+	perTagClients["foo"] = customFakeSinkFoo
+	perTagClients["bar"] = customFakeSinkBar
+
+	sink, err := NewSignalFxSink("host", "glooblestoots", commonDimensions, logrus.New(), defaultFakeSink, varyByTagKey, perTagClients, nil, nil, derived, 0, "", false, time.Second, "", "", nil)
+
+	assert.NoError(t, err)
+
+	interMetrics := []samplers.InterMetric{samplers.InterMetric{
+		Name:      "a.b.c",
+		Timestamp: 1476119058,
+		Value:     float64(100),
+		Tags: []string{
+			"vary_by:foo",
+		},
+		Type: samplers.GaugeMetric,
+	},
+		samplers.InterMetric{
+			Name:      "a.b.d",
+			Timestamp: 1476119059,
+			Value:     float64(100),
+			Tags:      []string{},
+			Type:      samplers.GaugeMetric,
+		}}
+
+	sink.Flush(context.TODO(), interMetrics)
+
+	assert.Equal(t, 0, len(defaultFakeSink.points))
+	assert.Equal(t, 1, len(customFakeSinkFoo.points))
+	assert.Equal(t, 1, len(customFakeSinkBar.points))
+}