smart agent: add extraDimension support (open-telemetry#69)

Author: rmfitzpatrick

internal/receiver/smartagentreceiver/output.go

@@ -23,6 +23,7 @@ import (
 	"github.com/signalfx/golib/v3/trace"
 	"github.com/signalfx/signalfx-agent/pkg/core/dpfilters"
 	"github.com/signalfx/signalfx-agent/pkg/monitors/types"
+	"github.com/signalfx/signalfx-agent/pkg/utils"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/obsreport"
 	"go.uber.org/zap"
@@ -34,20 +35,22 @@ const internalTransport = "internal"
 // It is what provides metrics to the next MetricsConsumer (to be implemented later).  At this stage it is only
 // a logging instance.
 type Output struct {
-	receiverName string
-	nextConsumer consumer.MetricsConsumer
-	logger       *zap.Logger
-	converter    Converter
+	receiverName    string
+	nextConsumer    consumer.MetricsConsumer
+	logger          *zap.Logger
+	converter       Converter
+	extraDimensions map[string]string
 }
 
 var _ types.FilteringOutput = (*Output)(nil)
 
 func NewOutput(config Config, nextConsumer consumer.MetricsConsumer, logger *zap.Logger) *Output {
 	return &Output{
-		receiverName: config.Name(),
-		nextConsumer: nextConsumer,
-		logger:       logger,
-		converter:    Converter{logger: logger},
+		receiverName:    config.Name(),
+		nextConsumer:    nextConsumer,
+		logger:          logger,
+		converter:       Converter{logger: logger},
+		extraDimensions: map[string]string{},
 	}
 }
 
@@ -70,15 +73,23 @@ func (output *Output) HasAnyExtraMetrics() bool {
 	return false
 }
 
+// Some monitors will clone their Output to provide to child monitors with their own extraDimensions
 func (output *Output) Copy() types.Output {
-	output.logger.Debug("Copy has been called.")
-	return output
+	output.logger.Debug("Copying Output", zap.Any("output", output))
+	cp := *output
+	cp.extraDimensions = utils.CloneStringMap(output.extraDimensions)
+	return &cp
 }
 
 func (output *Output) SendDatapoints(datapoints ...*datapoint.Datapoint) {
 	ctx := obsreport.ReceiverContext(context.Background(), output.receiverName, internalTransport)
 	ctx = obsreport.StartMetricsReceiveOp(ctx, typeStr, internalTransport)
 
+	for _, dp := range datapoints {
+		// Output's extraDimensions take priority over datapoint's
+		dp.Dimensions = utils.MergeStringMaps(dp.Dimensions, output.extraDimensions)
+	}
+
 	metrics, numDropped := output.converter.toMetrics(datapoints, time.Now())
 	if numDropped > 0 {
 		output.logger.Debug("SendDatapoints has dropped points", zap.Int("numDropped", numDropped))
@@ -98,15 +109,17 @@ func (output *Output) SendSpans(spans ...*trace.Span) {
 }
 
 func (output *Output) SendDimensionUpdate(dimension *types.Dimension) {
-	output.logger.Debug("SendDimensionUpdate has been called.", zap.Any("dimension", dimension))
+	output.logger.Debug("SendDimensionUpdate has been called.", zap.Any("dimension", dimension.String()))
 }
 
 func (output *Output) AddExtraDimension(key, value string) {
-	output.logger.Debug("AddExtraDimension has been called.", zap.String("key", key), zap.String("value", value))
+	output.logger.Debug("Adding extra dimension", zap.String("key", key), zap.String("value", value))
+	output.extraDimensions[key] = value
 }
 
 func (output *Output) RemoveExtraDimension(key string) {
-	output.logger.Debug("RemoveExtraDimension has been called.", zap.String("key", key))
+	output.logger.Debug("Removing extra dimension", zap.String("key", key))
+	delete(output.extraDimensions, key)
 }
 
 func (output *Output) AddExtraSpanTag(key, value string) {

internal/receiver/smartagentreceiver/output_test.go

@@ -21,21 +21,18 @@ import (
 	"github.com/signalfx/signalfx-agent/pkg/core/dpfilters"
 	"github.com/signalfx/signalfx-agent/pkg/monitors/types"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.uber.org/zap"
 )
 
 func TestOutput(t *testing.T) {
-	output := Output{
-		nextConsumer: consumertest.NewMetricsNop(),
-		logger:       zap.NewNop(),
-	}
-
+	output := NewOutput(Config{}, consumertest.NewMetricsNop(), zap.NewNop())
 	output.AddDatapointExclusionFilter(dpfilters.DatapointFilter(nil))
 	assert.Empty(t, output.EnabledMetrics())
 	assert.False(t, output.HasEnabledMetricInGroup(""))
 	assert.False(t, output.HasAnyExtraMetrics())
-	assert.Same(t, &output, output.Copy())
+	assert.NotSame(t, &output, output.Copy())
 	output.SendDatapoints()
 	output.SendEvent(new(event.Event))
 	output.SendSpans()
@@ -47,3 +44,25 @@ func TestOutput(t *testing.T) {
 	output.AddDefaultSpanTag("", "")
 	output.RemoveDefaultSpanTag("")
 }
+
+func TestExtraDimensions(t *testing.T) {
+	output := NewOutput(Config{}, consumertest.NewMetricsNop(), zap.NewNop())
+	assert.Empty(t, output.extraDimensions)
+
+	output.RemoveExtraDimension("not_a_known_dimension_name")
+
+	output.AddExtraDimension("a_dimension_name", "a_value")
+	assert.Equal(t, "a_value", output.extraDimensions["a_dimension_name"])
+
+	cp, ok := output.Copy().(*Output)
+	require.True(t, ok)
+	assert.Equal(t, "a_value", cp.extraDimensions["a_dimension_name"])
+
+	cp.RemoveExtraDimension("a_dimension_name")
+	assert.Empty(t, cp.extraDimensions["a_dimension_name"])
+	assert.Equal(t, "a_value", output.extraDimensions["a_dimension_name"])
+
+	cp.AddExtraDimension("another_dimension_name", "another_dimension_value")
+	assert.Equal(t, "another_dimension_value", cp.extraDimensions["another_dimension_name"])
+	assert.Empty(t, output.extraDimensions["another_dimension_name"])
+}

internal/receiver/smartagentreceiver/receiver.go

@@ -114,12 +114,17 @@ func (r *Receiver) createMonitor(monitorType string) (interface{}, error) {
 		}
 	}
 
+	var output *Output
 	if monitorOutputValue.IsValid() {
-		output := NewOutput(*r.config, r.nextConsumer, r.logger)
+		output = NewOutput(*r.config, r.nextConsumer, r.logger)
 		monitorOutputValue.Set(reflect.ValueOf(output))
 	} else {
 		return nil, fmt.Errorf("invalid monitor instance: %#v", monitor)
 	}
 
+	for k, v := range r.config.monitorConfig.MonitorConfigCore().ExtraDimensions {
+		output.AddExtraDimension(k, v)
+	}
+
 	return monitor, nil
 }

internal/receiver/smartagentreceiver/receiver_test.go

@@ -57,6 +57,9 @@ func newConfig(nameVal, monitorType string, intervalSeconds int) Config {
 			MonitorConfig: config.MonitorConfig{
 				Type:            monitorType,
 				IntervalSeconds: intervalSeconds,
+				ExtraDimensions: map[string]string{
+					"required_dimension": "required_value",
+				},
 			},
 			ReportPerCPU: true,
 		},
@@ -67,12 +70,15 @@ func TestSmartAgentReceiver(t *testing.T) {
 	cfg := newConfig("valid", "cpu", 10)
 	consumer := new(consumertest.MetricsSink)
 	receiver := NewReceiver(zap.NewNop(), cfg, consumer)
+
 	err := receiver.Start(context.Background(), componenttest.NewNopHost())
 	require.NoError(t, err)
 
 	assert.EqualValues(t, "smartagentvalid", cfg.monitorConfig.MonitorConfigCore().MonitorID)
+	monitor, isMonitor := receiver.monitor.(*cpu.Monitor)
+	require.True(t, isMonitor)
 
-	monitorOutput := receiver.monitor.(*cpu.Monitor).Output
+	monitorOutput := monitor.Output
 	_, isOutput := monitorOutput.(*Output)
 	assert.True(t, isOutput)
 
@@ -129,10 +135,16 @@ func TestSmartAgentReceiver(t *testing.T) {
 						default:
 							t.Errorf("unexpected type %#v for metric %s", metric.DataType(), name)
 						}
-						if labels.Len() == 0 {
-							assert.False(t, seenTotalMetric[name], "unexpected repeated total metric for %v", name)
-							seenTotalMetric[name] = true
-						}
+
+						labelVal, ok := labels.Get("required_dimension")
+						require.True(t, ok)
+						assert.Equal(t, "required_value", labelVal)
+
+						// mark metric as having been seen
+						cpuNum, _ := labels.Get("cpu")
+						seenName := fmt.Sprintf("%s%s", name, cpuNum)
+						assert.False(t, seenTotalMetric[seenName], "unexpectedly repeated metric: %v", seenName)
+						seenTotalMetric[seenName] = true
 					}
 				}
 			}