Implement metrics for infraattributesprocessor (DataDog#25711)

* Initial version

* Rename tagenrichmentprocessor to infraattributesprocessor

* Fix merge

* Rename config_logs_strict

* Add tagger component

* Fix tests and lint

* tidy

* Refactor to tag multiple entities

* Rename

* Temp update mapping go attributes

* tidy

* Update attributes mapping

* Fix resource metrics loop, add more tests

* Update attributes to v0.16.1

* Update attributes to v0.16.1

* Add container_image_metadata

* generate licenses

* Rename

* Fix merge conflicts

* generate deps

LICENSE-3rdparty.csv

@@ -3085,6 +3085,7 @@ core,go.opentelemetry.io/collector/semconv/v1.13.0,Apache-2.0,Copyright The Open
 core,go.opentelemetry.io/collector/semconv/v1.16.0,Apache-2.0,Copyright The OpenTelemetry Authors
 core,go.opentelemetry.io/collector/semconv/v1.17.0,Apache-2.0,Copyright The OpenTelemetry Authors
 core,go.opentelemetry.io/collector/semconv/v1.18.0,Apache-2.0,Copyright The OpenTelemetry Authors
+core,go.opentelemetry.io/collector/semconv/v1.21.0,Apache-2.0,Copyright The OpenTelemetry Authors
 core,go.opentelemetry.io/collector/semconv/v1.22.0,Apache-2.0,Copyright The OpenTelemetry Authors
 core,go.opentelemetry.io/collector/semconv/v1.6.1,Apache-2.0,Copyright The OpenTelemetry Authors
 core,go.opentelemetry.io/collector/semconv/v1.8.0,Apache-2.0,Copyright The OpenTelemetry Authors

cmd/otel-agent/subcommands/run/command.go

@@ -20,6 +20,8 @@ import (
 	corelogimpl "github.com/DataDog/datadog-agent/comp/core/log/logimpl"
 	"github.com/DataDog/datadog-agent/comp/core/secrets"
 	"github.com/DataDog/datadog-agent/comp/core/sysprobeconfig"
+	"github.com/DataDog/datadog-agent/comp/core/tagger"
+	"github.com/DataDog/datadog-agent/comp/core/tagger/taggerimpl"
 	"github.com/DataDog/datadog-agent/comp/core/workloadmeta"
 	"github.com/DataDog/datadog-agent/comp/forwarder"
 	"github.com/DataDog/datadog-agent/comp/forwarder/defaultforwarder"
@@ -81,6 +83,8 @@ func runOTelAgentCommand(_ context.Context, params *subcommands.GlobalParams, op
 		inventoryagentimpl.Module(),
 		workloadmeta.Module(),
 		hostnameimpl.Module(),
+		fx.Provide(tagger.NewTaggerParams),
+		taggerimpl.Module(),
 		sysprobeconfig.NoneModule(),
 		fetchonlyimpl.Module(),
 		collectorfx.Module(),

cmd/serverless/dependencies_linux_amd64.txt

@@ -611,6 +611,7 @@ go.opentelemetry.io/collector/receiver/otlpreceiver/internal/trace
 go.opentelemetry.io/collector/receiver/receiverhelper
 go.opentelemetry.io/collector/semconv/v1.17.0
 go.opentelemetry.io/collector/semconv/v1.18.0
+go.opentelemetry.io/collector/semconv/v1.21.0
 go.opentelemetry.io/collector/semconv/v1.6.1
 go.opentelemetry.io/collector/service
 go.opentelemetry.io/collector/service/extensions

cmd/serverless/dependencies_linux_arm64.txt

@@ -610,6 +610,7 @@ go.opentelemetry.io/collector/receiver/otlpreceiver/internal/trace
 go.opentelemetry.io/collector/receiver/receiverhelper
 go.opentelemetry.io/collector/semconv/v1.17.0
 go.opentelemetry.io/collector/semconv/v1.18.0
+go.opentelemetry.io/collector/semconv/v1.21.0
 go.opentelemetry.io/collector/semconv/v1.6.1
 go.opentelemetry.io/collector/service
 go.opentelemetry.io/collector/service/extensions

comp/otelcol/collector/impl-pipeline/pipeline.go

@@ -14,6 +14,7 @@ import (
 	"github.com/DataDog/datadog-agent/comp/core/config"
 	corelog "github.com/DataDog/datadog-agent/comp/core/log"
 	"github.com/DataDog/datadog-agent/comp/core/status"
+	"github.com/DataDog/datadog-agent/comp/core/tagger"
 	compdef "github.com/DataDog/datadog-agent/comp/def"
 	"github.com/DataDog/datadog-agent/comp/metadata/inventoryagent"
 	collector "github.com/DataDog/datadog-agent/comp/otelcol/collector/def"
@@ -51,6 +52,8 @@ type Requires struct {
 
 	// InventoryAgent require the inventory metadata payload, allowing otelcol to add data to it.
 	InventoryAgent inventoryagent.Component
+
+	Tagger tagger.Component
 }
 
 // Provides specifics the types returned by the constructor
@@ -68,6 +71,7 @@ type collectorImpl struct {
 	serializer     serializer.MetricSerializer
 	logsAgent      optional.Option[logsagentpipeline.Component]
 	inventoryAgent inventoryagent.Component
+	tagger         tagger.Component
 }
 
 func (c *collectorImpl) start(context.Context) error {
@@ -83,7 +87,7 @@ func (c *collectorImpl) start(context.Context) error {
 		}
 	}
 	var err error
-	col, err := otlp.NewPipelineFromAgentConfig(c.config, c.serializer, logch)
+	col, err := otlp.NewPipelineFromAgentConfig(c.config, c.serializer, logch, c.tagger)
 	if err != nil {
 		// failure to start the OTLP component shouldn't fail startup
 		c.log.Errorf("Error creating the OTLP ingest pipeline: %v", err)
@@ -121,6 +125,7 @@ func NewComponent(reqs Requires) (Provides, error) {
 		serializer:     reqs.Serializer,
 		logsAgent:      reqs.LogsAgent,
 		inventoryAgent: reqs.InventoryAgent,
+		tagger:         reqs.Tagger,
 	}
 
 	reqs.Lc.Append(compdef.Hook{

comp/otelcol/collector/impl/collector.go

@@ -17,6 +17,7 @@ import (
 	flarebuilder "github.com/DataDog/datadog-agent/comp/core/flare/builder"
 	"github.com/DataDog/datadog-agent/comp/core/hostname"
 	corelog "github.com/DataDog/datadog-agent/comp/core/log"
+	"github.com/DataDog/datadog-agent/comp/core/tagger"
 	compdef "github.com/DataDog/datadog-agent/comp/def"
 	collectorcontrib "github.com/DataDog/datadog-agent/comp/otelcol/collector-contrib/def"
 	collector "github.com/DataDog/datadog-agent/comp/otelcol/collector/def"
@@ -48,6 +49,7 @@ type Requires struct {
 	Serializer       serializer.MetricSerializer
 	LogsAgent        optional.Option[logsagentpipeline.Component]
 	HostName         hostname.Component
+	Tagger           tagger.Component
 }
 
 // Provides declares the output types from the constructor
@@ -76,7 +78,7 @@ func NewComponent(reqs Requires) (Provides, error) {
 			} else {
 				factories.Exporters[datadogexporter.Type] = datadogexporter.NewFactory(reqs.Serializer, nil, reqs.HostName)
 			}
-			factories.Processors[infraattributesprocessor.Type] = infraattributesprocessor.NewFactory()
+			factories.Processors[infraattributesprocessor.Type] = infraattributesprocessor.NewFactory(reqs.Tagger)
 			return factories, nil
 		},
 		ConfigProvider: reqs.Provider,

comp/otelcol/otlp/collector.go

@@ -82,7 +82,7 @@ func (t *tagEnricher) Enrich(_ context.Context, extraTags []string, dimensions *
 	return enrichedTags
 }
 
-func getComponents(s serializer.MetricSerializer, logsAgentChannel chan *message.Message) (
+func getComponents(s serializer.MetricSerializer, logsAgentChannel chan *message.Message, tagger tagger.Component) (
 	otelcol.Factories,
 	error,
 ) {
@@ -115,10 +115,11 @@ func getComponents(s serializer.MetricSerializer, logsAgentChannel chan *message
 		errs = append(errs, err)
 	}
 
-	processors, err := processor.MakeFactoryMap(
-		batchprocessor.NewFactory(),
-		infraattributesprocessor.NewFactory(),
-	)
+	processorFactories := []processor.Factory{batchprocessor.NewFactory()}
+	if tagger != nil {
+		processorFactories = append(processorFactories, infraattributesprocessor.NewFactory(tagger))
+	}
+	processors, err := processor.MakeFactoryMap(processorFactories...)
 	if err != nil {
 		errs = append(errs, err)
 	}
@@ -194,7 +195,7 @@ type Pipeline struct {
 type CollectorStatus = datatype.CollectorStatus
 
 // NewPipeline defines a new OTLP pipeline.
-func NewPipeline(cfg PipelineConfig, s serializer.MetricSerializer, logsAgentChannel chan *message.Message) (*Pipeline, error) {
+func NewPipeline(cfg PipelineConfig, s serializer.MetricSerializer, logsAgentChannel chan *message.Message, tagger tagger.Component) (*Pipeline, error) {
 	buildInfo, err := getBuildInfo()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get build info: %w", err)
@@ -214,7 +215,7 @@ func NewPipeline(cfg PipelineConfig, s serializer.MetricSerializer, logsAgentCha
 
 	col, err := otelcol.NewCollector(otelcol.CollectorSettings{
 		Factories: func() (otelcol.Factories, error) {
-			return getComponents(s, logsAgentChannel)
+			return getComponents(s, logsAgentChannel, tagger)
 		},
 		BuildInfo:               buildInfo,
 		DisableGracefulShutdown: true,
@@ -251,7 +252,7 @@ func (p *Pipeline) Stop() {
 
 // NewPipelineFromAgentConfig creates a new pipeline from the given agent configuration, metric serializer and logs channel. It returns
 // any potential failure.
-func NewPipelineFromAgentConfig(cfg config.Component, s serializer.MetricSerializer, logsAgentChannel chan *message.Message) (*Pipeline, error) {
+func NewPipelineFromAgentConfig(cfg config.Component, s serializer.MetricSerializer, logsAgentChannel chan *message.Message, tagger tagger.Component) (*Pipeline, error) {
 	pcfg, err := FromAgentConfig(cfg)
 	if err != nil {
 		pipelineError.Store(fmt.Errorf("config error: %w", err))
@@ -260,7 +261,7 @@ func NewPipelineFromAgentConfig(cfg config.Component, s serializer.MetricSeriali
 	if err := checkAndUpdateCfg(cfg, pcfg, logsAgentChannel); err != nil {
 		return nil, err
 	}
-	p, err := NewPipeline(pcfg, s, logsAgentChannel)
+	p, err := NewPipeline(pcfg, s, logsAgentChannel, tagger)
 	if err != nil {
 		pipelineError.Store(fmt.Errorf("failed to build pipeline: %w", err))
 		return nil, pipelineError.Load()

comp/otelcol/otlp/collector_test.go

@@ -13,6 +13,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/DataDog/datadog-agent/comp/core/tagger/taggerimpl"
 	"github.com/DataDog/datadog-agent/comp/otelcol/otlp/testutil"
 	"github.com/DataDog/datadog-agent/pkg/config"
 	"github.com/DataDog/datadog-agent/pkg/logs/message"
@@ -23,13 +24,17 @@ import (
 )
 
 func TestGetComponents(t *testing.T) {
-	_, err := getComponents(&serializer.MockSerializer{}, make(chan *message.Message))
+	fakeTagger := taggerimpl.SetupFakeTagger(t)
+	defer fakeTagger.ResetTagger()
+	_, err := getComponents(&serializer.MockSerializer{}, make(chan *message.Message), fakeTagger)
 	// No duplicate component
 	require.NoError(t, err)
 }
 
 func AssertSucessfulRun(t *testing.T, pcfg PipelineConfig) {
-	p, err := NewPipeline(pcfg, &serializer.MockSerializer{}, make(chan *message.Message))
+	fakeTagger := taggerimpl.SetupFakeTagger(t)
+	defer fakeTagger.ResetTagger()
+	p, err := NewPipeline(pcfg, &serializer.MockSerializer{}, make(chan *message.Message), fakeTagger)
 	require.NoError(t, err)
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
@@ -54,7 +59,9 @@ func AssertSucessfulRun(t *testing.T, pcfg PipelineConfig) {
 }
 
 func AssertFailedRun(t *testing.T, pcfg PipelineConfig, expected string) {
-	p, err := NewPipeline(pcfg, &serializer.MockSerializer{}, make(chan *message.Message))
+	fakeTagger := taggerimpl.SetupFakeTagger(t)
+	defer fakeTagger.ResetTagger()
+	p, err := NewPipeline(pcfg, &serializer.MockSerializer{}, make(chan *message.Message), fakeTagger)
 	require.NoError(t, err)
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()

comp/otelcol/otlp/components/exporter/datadogexporter/go.mod

@@ -85,7 +85,7 @@ require (
 	github.com/DataDog/datadog-agent/pkg/logs/message v0.54.0-rc.2
 	github.com/DataDog/datadog-agent/pkg/serializer v0.54.0-rc.2
 	github.com/DataDog/datadog-agent/pkg/util/hostname/validate v0.54.0-rc.2
-	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.16.0
+	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.16.1
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/metrics v0.16.0
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/component v0.100.0

comp/otelcol/otlp/components/exporter/logsagentexporter/go.mod

@@ -42,7 +42,7 @@ require (
 	github.com/DataDog/datadog-agent/pkg/logs/message v0.54.0-rc.2
 	github.com/DataDog/datadog-agent/pkg/logs/sources v0.54.0-rc.2
 	github.com/DataDog/datadog-agent/pkg/util/scrubber v0.54.0-rc.2
-	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.14.0
+	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.16.1
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/logs v0.14.0
 	github.com/stormcat24/protodep v0.1.8
 	github.com/stretchr/testify v1.9.0

comp/otelcol/otlp/components/exporter/serializerexporter/go.mod

@@ -62,7 +62,7 @@ require (
 	github.com/DataDog/datadog-agent/pkg/serializer v0.54.0-rc.2
 	github.com/DataDog/datadog-agent/pkg/tagset v0.54.0-rc.2
 	github.com/DataDog/datadog-agent/pkg/util/log v0.54.0-rc.2
-	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.14.0
+	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.16.1
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/metrics v0.14.0
 	github.com/DataDog/opentelemetry-mapping-go/pkg/quantile v0.14.0
 	github.com/mitchellh/mapstructure v1.5.1-0.20231216201459-8508981c8b6c // indirect

comp/otelcol/otlp/components/pipeline/provider/go.mod

@@ -82,7 +82,7 @@ require (
 	github.com/DataDog/go-tuf v1.1.0-0.5.2 // indirect
 	github.com/DataDog/gohai v0.0.0-20230524154621-4316413895ee // indirect
 	github.com/DataDog/opentelemetry-mapping-go/pkg/inframetadata v0.16.0 // indirect
-	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.16.0 // indirect
+	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.16.1 // indirect
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/logs v0.16.0 // indirect
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/metrics v0.16.0 // indirect
 	github.com/DataDog/opentelemetry-mapping-go/pkg/quantile v0.16.0 // indirect

comp/otelcol/otlp/components/processor/infraattributesprocessor/config.go

@@ -7,13 +7,17 @@ package infraattributesprocessor
 
 import (
 	"go.opentelemetry.io/collector/component"
+
+	"github.com/DataDog/datadog-agent/comp/core/tagger/types"
 )
 
 // Config defines configuration for processor.
 type Config struct {
 	Metrics MetricInfraAttributes `mapstructure:"metrics"`
 	Logs    LogInfraAttributes    `mapstructure:"logs"`
 	Traces  TraceInfraAttributes  `mapstructure:"traces"`
+
+	Cardinality types.TagCardinality `mapstructure:"cardinality"`
 }
 
 // MetricInfraAttributes - configuration for metrics.

comp/otelcol/otlp/components/processor/infraattributesprocessor/config_test.go

@@ -13,6 +13,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
+
+	"github.com/DataDog/datadog-agent/comp/core/tagger/taggerimpl"
 )
 
 // TestLoadingConfigStrictLogs tests loading testdata/logs_strict.yaml
@@ -34,8 +36,10 @@ func TestLoadingConfigStrictLogs(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.id.String(), func(t *testing.T) {
-			factory := NewFactory()
-			cfg := factory.CreateDefaultConfig()
+			fakeTagger := taggerimpl.SetupFakeTagger(t)
+			defer fakeTagger.ResetTagger()
+			f := NewFactory(fakeTagger)
+			cfg := f.CreateDefaultConfig()
 
 			sub, err := cm.Sub(tt.id.String())
 			require.NoError(t, err)