diff --git a/CHANGELOG.md b/CHANGELOG.md index 21227d78c43b..02c8318c57e8 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/docs/configuration/traces-config.md b/docs/configuration/traces-config.md index 625583f2574d..3bff07a23000 100644 --- a/docs/configuration/traces-config.md +++ b/docs/configuration/traces-config.md @@ -132,6 +132,11 @@ receivers: # match is found then relabeling rules are applied. scrape_configs: - [] +# 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: | default = "upsert" ] # spanmetrics supports aggregating Request, Error and Duration (R.E.D) metrics # from span data. diff --git a/docs/upgrade-guide/_index.md b/docs/upgrade-guide/_index.md index e586ae0ccc86..5d9605995877 100644 --- a/docs/upgrade-guide/_index.md +++ b/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 diff --git a/pkg/traces/config.go b/pkg/traces/config.go index 4ebbad659c11..9fd82e1d4080 100644 --- a/pkg/traces/config.go +++ b/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, } } diff --git a/pkg/traces/config_test.go b/pkg/traces/config_test.go index 59f813b02c19..738307639737 100644 --- a/pkg/traces/config_test.go +++ b/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"] `, }, } diff --git a/pkg/traces/promsdprocessor/factory.go b/pkg/traces/promsdprocessor/factory.go index fa52497580b3..74ab46c9f1ca 100644 --- a/pkg/traces/promsdprocessor/factory.go +++ b/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) } diff --git a/pkg/traces/promsdprocessor/prom_sd_processor.go b/pkg/traces/promsdprocessor/prom_sd_processor.go index 79ebb7dfdc74..cb480d96c354 100644 --- a/pkg/traces/promsdprocessor/prom_sd_processor.go +++ b/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)) + } } } diff --git a/pkg/traces/promsdprocessor/prom_sd_processor_test.go b/pkg/traces/promsdprocessor/prom_sd_processor_test.go index 994c0975b49b..f3b5932461ab 100644 --- a/pkg/traces/promsdprocessor/prom_sd_processor_test.go +++ b/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()) + }) + } +}