Making configurable what method is used when adding k/v to spans

CHANGELOG.md

@@ -31,6 +31,9 @@ for specific instructions.
 
 - [ENHANCEMENT] Update jsonnet-libs to 1.21 for Kubernetes 1.21+ compatability. (@MurzNN)
 
+- [ENHANCEMENT] Make method used to add k/v to spans in prom_sd processor
+  configurable. (@mapno)
+
 - [BUGFIX] Regex capture groups like `${1}` will now be kept intact when
   using `-config.expand-env`. (@rfratto)
 

docs/configuration/traces-config.md

@@ -132,6 +132,11 @@ receivers: <receivers>
 # match is found then relabeling rules are applied.
 scrape_configs:
   - [<scrape_config>]
+# Defines what method is used when adding k/v to spans.
+# Options are `update`, `insert` and `upsert`.
+# `update` only modifies an existing k/v and `insert` only appends if the k/v
+# is not present. `upsert` does both.
+[ prom_sd_operation_type: <string> | default = "upsert" ]
 
 # spanmetrics supports aggregating Request, Error and Duration (R.E.D) metrics
 # from span data.

docs/upgrade-guide/_index.md

@@ -12,8 +12,6 @@ releases and how to migrate to newer versions.
 
 These changes will come in a future version.
 
-
-
 ### Traces: Deprecation of "tempo" in config and metrics. (Deprecation)
 
 The term `tempo` in the config has been deprecated of favor of `traces`. This

pkg/traces/config.go

@@ -114,8 +114,9 @@ type InstanceConfig struct {
 	// Attributes: https://github.com/open-telemetry/opentelemetry-collector/blob/7d7ae2eb34b5d387627875c498d7f43619f37ee3/processor/attributesprocessor/config.go#L30
 	Attributes map[string]interface{} `yaml:"attributes,omitempty"`
 
-	// prom service discovery
+	// prom service discovery config
 	ScrapeConfigs []interface{} `yaml:"scrape_configs,omitempty"`
+	OperationType string        `yaml:"prom_sd_operation_type,omitempty"`
 
 	// SpanMetricsProcessor: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/spanmetricsprocessor/README.md
 	SpanMetrics *SpanMetricsConfig `yaml:"spanmetrics,omitempty"`
@@ -448,9 +449,14 @@ func (c *InstanceConfig) otelConfig() (*config.Config, error) {
 	processors := map[string]interface{}{}
 	processorNames := []string{}
 	if c.ScrapeConfigs != nil {
+		opType := promsdprocessor.OperationTypeUpsert
+		if c.OperationType != "" {
+			opType = c.OperationType
+		}
 		processorNames = append(processorNames, promsdprocessor.TypeStr)
 		processors[promsdprocessor.TypeStr] = map[string]interface{}{
 			"scrape_configs": c.ScrapeConfigs,
+			"operation_type": opType,
 		}
 	}
 

pkg/traces/config_test.go

@@ -774,6 +774,44 @@ service:
       exporters: ["otlphttp/0", "otlp/1"]
       processors: []
       receivers: ["jaeger"]
+`,
+		},
+		{
+			name: "prom SD config",
+			cfg: `
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+remote_write:
+  - endpoint: example.com:12345
+    protocol: grpc
+scrape_configs:
+  - im_a_scrape_config
+prom_sd_operation_type: update
+`,
+			expectedConfig: `
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+exporters:
+  otlp/0:
+    endpoint: example.com:12345
+    compression: gzip
+    retry_on_failure:
+      max_elapsed_time: 60s
+processors:
+  prom_sd_processor:
+    scrape_configs:
+      - im_a_scrape_config
+    operation_type: update
+service:
+  pipelines:
+    traces:
+      exporters: ["otlp/0"]
+      processors: ["prom_sd_processor"]
+      receivers: ["jaeger"]
 `,
 		},
 	}

pkg/traces/promsdprocessor/factory.go

@@ -15,10 +15,20 @@ import (
 // TypeStr is the unique identifier for the Prometheus SD processor.
 const TypeStr = "prom_sd_processor"
 
+const (
+	// OperationTypeInsert inserts a new k/v if it isn't already present
+	OperationTypeInsert = "insert"
+	// OperationTypeUpdate only modifies an existing k/v
+	OperationTypeUpdate = "update"
+	// OperationTypeUpsert does both of above
+	OperationTypeUpsert = "upsert"
+)
+
 // Config holds the configuration for the Prometheus SD processor.
 type Config struct {
 	config.ProcessorSettings `mapstructure:",squash"`
 	ScrapeConfigs            []interface{} `mapstructure:"scrape_configs"`
+	OperationType            string        `mapstructure:"operation_type"`
 }
 
 // NewFactory returns a new factory for the Attributes processor.
@@ -55,5 +65,5 @@ func createTraceProcessor(
 		return nil, fmt.Errorf("unable to unmarshal bytes to []*config.ScrapeConfig: %w", err)
 	}
 
-	return newTraceProcessor(nextConsumer, scrapeConfigs)
+	return newTraceProcessor(nextConsumer, oCfg.OperationType, scrapeConfigs)
 }

pkg/traces/promsdprocessor/prom_sd_processor.go

@@ -2,6 +2,7 @@ package promsdprocessor
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"strings"
 	"sync"
@@ -19,6 +20,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenterror"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/model/pdata"
+	"go.opentelemetry.io/collector/translator/conventions"
 )
 
 type promServiceDiscoProcessor struct {
@@ -31,27 +33,38 @@ type promServiceDiscoProcessor struct {
 	hostLabels     map[string]model.LabelSet
 	mtx            sync.Mutex
 
+	operationType string
+
 	logger log.Logger
 }
 
-func newTraceProcessor(nextConsumer consumer.Traces, scrapeConfigs []*config.ScrapeConfig) (component.TracesProcessor, error) {
+func newTraceProcessor(nextConsumer consumer.Traces, operationType string, scrapeConfigs []*config.ScrapeConfig) (component.TracesProcessor, error) {
 	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	logger := log.With(util.Logger, "component", "traces service disco")
 	mgr := discovery.NewManager(ctx, logger, discovery.Name("traces service disco"))
 
 	relabelConfigs := map[string][]*relabel.Config{}
-	cfg := map[string]discovery.Configs{}
+	managerConfig := map[string]discovery.Configs{}
 	for _, v := range scrapeConfigs {
-		cfg[v.JobName] = v.ServiceDiscoveryConfigs
+		managerConfig[v.JobName] = v.ServiceDiscoveryConfigs
 		relabelConfigs[v.JobName] = v.RelabelConfigs
 	}
 
-	err := mgr.ApplyConfig(cfg)
+	err := mgr.ApplyConfig(managerConfig)
 	if err != nil {
-		cancel()
 		return nil, err
 	}
 
+	switch operationType {
+	case OperationTypeUpsert, OperationTypeInsert, OperationTypeUpdate:
+	case "": // Use Upsert by default
+		operationType = OperationTypeUpsert
+	default:
+		return nil, fmt.Errorf("unknown operation type %s", operationType)
+	}
+
 	if nextConsumer == nil {
 		cancel()
 		return nil, componenterror.ErrNilNextConsumer
@@ -64,6 +77,7 @@ func newTraceProcessor(nextConsumer consumer.Traces, scrapeConfigs []*config.Scr
 		relabelConfigs:   relabelConfigs,
 		hostLabels:       make(map[string]model.LabelSet),
 		logger:           logger,
+		operationType:    operationType,
 	}, nil
 }
 
@@ -81,8 +95,8 @@ func (p *promServiceDiscoProcessor) ConsumeTraces(ctx context.Context, td pdata.
 func (p *promServiceDiscoProcessor) processAttributes(attrs pdata.AttributeMap) {
 	// find the ip
 	ipTagNames := []string{
-		"ip",          // jaeger/opentracing? default
-		"net.host.ip", // otel semantics for host ip
+		"ip",                           // jaeger/opentracing? default
+		conventions.AttributeNetHostIP, // otel semantics for host ip
 	}
 
 	var ip string
@@ -112,7 +126,14 @@ func (p *promServiceDiscoProcessor) processAttributes(attrs pdata.AttributeMap)
 	}
 
 	for k, v := range labels {
-		attrs.UpsertString(string(k), string(v))
+		switch p.operationType {
+		case OperationTypeUpsert:
+			attrs.UpsertString(string(k), string(v))
+		case OperationTypeInsert:
+			attrs.InsertString(string(k), string(v))
+		case OperationTypeUpdate:
+			attrs.UpdateString(string(k), string(v))
+		}
 	}
 }
 

pkg/traces/promsdprocessor/prom_sd_processor_test.go

@@ -8,6 +8,10 @@ import (
 	"github.com/prometheus/prometheus/discovery/targetgroup"
 	"github.com/prometheus/prometheus/pkg/relabel"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.opentelemetry.io/collector/translator/conventions"
 )
 
 func TestSyncGroups(t *testing.T) {
@@ -114,3 +118,88 @@ func TestSyncGroups(t *testing.T) {
 		})
 	}
 }
+
+func TestOperationType(t *testing.T) {
+	const (
+		attrKey = "key"
+		attrIP  = "1.1.1.1"
+	)
+	testCases := []struct {
+		name            string
+		operationType   string
+		attributeExists bool
+		newValue        string
+		expectedValue   string
+	}{
+		{
+			name:            "Upsert updates the attribute already exists",
+			operationType:   OperationTypeUpsert,
+			attributeExists: true,
+			newValue:        "new-value",
+			expectedValue:   "new-value",
+		},
+		{
+			name:            "Update updates the attribute already exists",
+			operationType:   OperationTypeUpdate,
+			attributeExists: true,
+			newValue:        "new-value",
+			expectedValue:   "new-value",
+		},
+		{
+			name:            "Insert does not update the attribute if it's already present",
+			operationType:   OperationTypeInsert,
+			attributeExists: true,
+			newValue:        "new-value",
+			expectedValue:   "old-value",
+		},
+		{
+			name:            "Upsert updates the attribute if it isn't present",
+			operationType:   OperationTypeUpsert,
+			attributeExists: false,
+			newValue:        "new-value",
+			expectedValue:   "new-value",
+		},
+		{
+			name:            "Update updates the attribute already exists",
+			operationType:   OperationTypeUpdate,
+			attributeExists: false,
+			newValue:        "new-value",
+			expectedValue:   "",
+		},
+		{
+			name:            "Insert updates the attribute if it isn't present",
+			operationType:   OperationTypeInsert,
+			attributeExists: false,
+			newValue:        "new-value",
+			expectedValue:   "new-value",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockProcessor := new(consumertest.TracesSink)
+			p, err := newTraceProcessor(mockProcessor, tc.operationType, nil)
+			require.NoError(t, err)
+
+			attrValue := pdata.NewAttributeValueString("old-value")
+			ipAttrValue := pdata.NewAttributeValueString(attrIP)
+
+			attrMap := pdata.NewAttributeMap()
+			if tc.attributeExists {
+				attrMap.Insert(attrKey, attrValue)
+			}
+			attrMap.Insert(conventions.AttributeNetHostIP, ipAttrValue)
+
+			hostLabels := map[string]model.LabelSet{
+				attrIP: {
+					attrKey: model.LabelValue(tc.newValue),
+				},
+			}
+			p.(*promServiceDiscoProcessor).hostLabels = hostLabels
+			p.(*promServiceDiscoProcessor).processAttributes(attrMap)
+
+			actualAttrValue, _ := attrMap.Get(attrKey)
+			assert.Equal(t, tc.expectedValue, actualAttrValue.StringVal())
+		})
+	}
+}