[Backport 7.82.x] Add metric lookback DogStatsD retention substrate

.github/CODEOWNERS

@@ -439,6 +439,7 @@
 /comp/haagent @DataDog/network-device-monitoring-core
 /comp/languagedetection/client @DataDog/container-platform
 /comp/logonduration @DataDog/windows-products
+/comp/metriclookback @DataDog/q-branch
 /comp/networkconfigmanagement @DataDog/ndm-integrations
 /comp/notableevents @DataDog/windows-products
 /comp/offlinereporter @DataDog/agent-metric-pipelines
@@ -486,6 +487,8 @@
 
 # pkg
 /pkg/                                   @DataDog/agent-runtimes
+/pkg/metriclookback/                    @DataDog/q-branch
+/pkg/metriclookback/dogstatsd/          @DataDog/q-branch @DataDog/agent-metric-pipelines
 /pkg/api/                               @DataDog/agent-runtimes
 /pkg/api/security/cert/cert_getter_dca.go   @DataDog/agent-runtimes @DataDog/container-platform
 /pkg/aggregator/                        @DataDog/agent-metric-pipelines

cmd/agent/subcommands/run/command.go

@@ -151,6 +151,7 @@ import (
 	runner "github.com/DataDog/datadog-agent/comp/metadata/runner/def"
 	securityagentmetadata "github.com/DataDog/datadog-agent/comp/metadata/securityagent/def"
 	systemprobemetadata "github.com/DataDog/datadog-agent/comp/metadata/systemprobe/def"
+	metriclookbackdef "github.com/DataDog/datadog-agent/comp/metriclookback/def"
 	"github.com/DataDog/datadog-agent/comp/ndmtmp"
 	"github.com/DataDog/datadog-agent/comp/netflow"
 	netflowServer "github.com/DataDog/datadog-agent/comp/netflow/server/def"
@@ -285,6 +286,7 @@ func run(log log.Component,
 	rcclient rcclient.Component,
 	_ runner.Component,
 	demultiplexer demultiplexer.Component,
+	metricLookback metriclookbackdef.Component,
 	_ serializer.MetricSerializer,
 	_ option.Option[logsAgent.Component],
 	_ statsd.Component,
@@ -371,6 +373,7 @@ func run(log log.Component,
 		ac,
 		rcclient,
 		demultiplexer,
+		metricLookback,
 		invChecks,
 		logReceiver,
 		collector,
@@ -469,6 +472,7 @@ func getSharedFxOption() fx.Option {
 		grpcAgentfx.Module(),
 		commonendpoints.Module(),
 		filterlist.Module(),
+		metriclookbackModule(),
 		demultiplexerimpl.Module(demultiplexerimpl.NewDefaultParams(demultiplexerimpl.WithDogstatsdNoAggregationPipelineConfig())),
 		demultiplexerendpointfx.Module(),
 		dogstatsd.Bundle(dogstatsdServer.Params{Serverless: false}),
@@ -639,6 +643,7 @@ func startAgent(
 	ac autodiscovery.Component,
 	rcclient rcclient.Component,
 	demultiplexer demultiplexer.Component,
+	metricLookback metriclookbackdef.Component,
 	invChecks inventorychecks.Component,
 	logReceiver option.Option[integrations.Component],
 	collectorComponent collector.Component,
@@ -742,7 +747,9 @@ func startAgent(
 
 	// Set up check collector
 	commonchecks.RegisterChecks(wmeta, filterStore, tagger, cfg, tlm, rcclient, flare, snmpScanManager, traceroute, ncmComp)
-	ac.AddScheduler("check", pkgcollector.InitCheckScheduler(option.New(collectorComponent), demultiplexer, logReceiver, tagger, filterStore), true)
+	checkScheduler := pkgcollector.InitCheckScheduler(option.New(collectorComponent), demultiplexer, logReceiver, tagger, filterStore)
+	checkScheduler.SetMetricLookbackShadowSenderManager(metricLookback.NewSenderManager(ctx, hostnameDetected))
+	ac.AddScheduler("check", checkScheduler, true)
 
 	demultiplexer.AddAgentStartupTelemetry(version.AgentVersion)
 

cmd/agent/subcommands/run/command_windows.go

@@ -76,6 +76,7 @@ import (
 	inventoryhost "github.com/DataDog/datadog-agent/comp/metadata/inventoryhost/def"
 	packagesigning "github.com/DataDog/datadog-agent/comp/metadata/packagesigning/def"
 	runner "github.com/DataDog/datadog-agent/comp/metadata/runner/def"
+	metriclookbackdef "github.com/DataDog/datadog-agent/comp/metriclookback/def"
 	netflowServer "github.com/DataDog/datadog-agent/comp/netflow/server/def"
 	otelcollector "github.com/DataDog/datadog-agent/comp/otelcol/collector/def"
 	processAgent "github.com/DataDog/datadog-agent/comp/process/agent/def"
@@ -124,6 +125,7 @@ func StartAgentWithDefaults(ctxChan <-chan context.Context) (<-chan error, error
 			_ serializer.MetricSerializer,
 			_ otelcollector.Component,
 			demultiplexer demultiplexer.Component,
+			metricLookback metriclookbackdef.Component,
 			_ host.Component,
 			_ inventoryagent.Component,
 			_ inventoryhost.Component,
@@ -165,6 +167,7 @@ func StartAgentWithDefaults(ctxChan <-chan context.Context) (<-chan error, error
 				ac,
 				rcclient,
 				demultiplexer,
+				metricLookback,
 				invChecks,
 				logsReceiver,
 				collector,

cmd/agent/subcommands/run/metriclookback.go

@@ -0,0 +1,18 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build python || test
+
+package run
+
+import (
+	"go.uber.org/fx"
+
+	metriclookbackfx "github.com/DataDog/datadog-agent/comp/metriclookback/fx"
+)
+
+func metriclookbackModule() fx.Option {
+	return metriclookbackfx.Module()
+}

cmd/agent/subcommands/run/metriclookback_nopython.go

@@ -0,0 +1,31 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+//go:build !python && !test
+
+package run
+
+import (
+	"context"
+
+	"go.uber.org/fx"
+
+	metriclookbackdef "github.com/DataDog/datadog-agent/comp/metriclookback/def"
+	"github.com/DataDog/datadog-agent/pkg/aggregator/sender"
+)
+
+type unavailableMetricLookback struct{}
+
+func (unavailableMetricLookback) NewSenderManager(context.Context, string) sender.SenderManager {
+	return nil
+}
+
+func metriclookbackModule() fx.Option {
+	// Keep concrete metric lookback retention out of non-Python Agent binaries,
+	// while satisfying shared Agent startup wiring with an unavailable component.
+	return fx.Provide(func() metriclookbackdef.Component {
+		return unavailableMetricLookback{}
+	})
+}

comp/README.md

@@ -891,6 +891,12 @@ Package client implements a component to send process metadata to the Cluster-Ag
 
 Package logonduration provides a component that monitors the duration of a user logon after boot and forwards them to the Datadog Event Management v2 API.
 
+### [comp/metriclookback](https://pkg.go.dev/github.com/DataDog/datadog-agent/comp/metriclookback)
+
+*Datadog Team*: q-branch
+
+Package metriclookback defines the metric lookback component.
+
 ### [comp/networkconfigmanagement](https://pkg.go.dev/github.com/DataDog/datadog-agent/comp/networkconfigmanagement)
 
 *Datadog Team*: ndm-integrations

comp/aggregator/demultiplexer/impl/BUILD.bazel

@@ -90,7 +90,9 @@ dd_agent_go_test(
         "//comp/serializer/logscompression/fx-mock",
         "//comp/serializer/metricscompression/def",
         "//comp/serializer/metricscompression/fx-mock",
+        "//pkg/aggregator",
         "//pkg/aggregator/mocksender",
+        "//pkg/serializer",
         "//pkg/util/fxutil",
         "@com_github_stretchr_testify//require",
         "@org_uber_go_fx//:fx",

comp/aggregator/demultiplexer/impl/demultiplexer.go

@@ -40,18 +40,19 @@ func Module(params Params) fxutil.Module {
 // Dependencies defines the dependencies required by the demultiplexer component.
 type Dependencies struct {
 	compdef.In
-	Lc                     compdef.Lifecycle
-	Config                 config.Component
-	Log                    log.Component
-	SharedForwarder        defaultforwarder.Component
-	OrchestratorForwarder  orchestratorforwarder.Component
-	EventPlatformForwarder eventplatform.Component
-	HaAgent                haagent.Component
-	Compressor             compression.Component
-	Tagger                 tagger.Component
-	Hostname               hostnameinterface.Component
-	FilterList             filterlist.Component
-	Observer               observer.Component `optional:"true"`
+	Lc                       compdef.Lifecycle
+	Config                   config.Component
+	Log                      log.Component
+	SharedForwarder          defaultforwarder.Component
+	OrchestratorForwarder    orchestratorforwarder.Component
+	EventPlatformForwarder   eventplatform.Component
+	HaAgent                  haagent.Component
+	Compressor               compression.Component
+	Tagger                   tagger.Component
+	Hostname                 hostnameinterface.Component
+	FilterList               filterlist.Component
+	Observer                 observer.Component                  `optional:"true"`
+	DogStatsDLookbackFactory aggregator.DogStatsDLookbackFactory `optional:"true"`
 
 	Params Params
 }
@@ -81,7 +82,7 @@ func NewComponent(deps Dependencies) (Provides, error) {
 			return Provides{}, deps.Log.Errorf("Error while getting hostname, exiting: %v", err)
 		}
 	}
-	options := createAgentDemultiplexerOptions(deps.Config, deps.Params)
+	options := createAgentDemultiplexerOptions(deps.Config, deps.Params, deps.DogStatsDLookbackFactory)
 	agentDemultiplexer := aggregator.InitAndStartAgentDemultiplexer(
 		deps.Log,
 		deps.SharedForwarder,
@@ -113,7 +114,7 @@ func NewComponent(deps Dependencies) (Provides, error) {
 	}, nil
 }
 
-func createAgentDemultiplexerOptions(config config.Component, params Params) aggregator.AgentDemultiplexerOptions {
+func createAgentDemultiplexerOptions(config config.Component, params Params, dogStatsDLookbackFactory aggregator.DogStatsDLookbackFactory) aggregator.AgentDemultiplexerOptions {
 	options := aggregator.DefaultAgentDemultiplexerOptions()
 	if params.useDogstatsdNoAggregationPipelineConfig {
 		if config.GetBool("dogstatsd_no_aggregation_pipeline") {
@@ -124,6 +125,8 @@ func createAgentDemultiplexerOptions(config config.Component, params Params) agg
 		}
 	}
 
+	options.DogStatsDLookbackFactory = dogStatsDLookbackFactory
+
 	// Override FlushInterval only if flushInterval is set by the user
 	if v, ok := params.flushInterval.Get(); ok {
 		options.FlushInterval = v

comp/aggregator/demultiplexer/impl/demultiplexer_options_test.go

@@ -12,6 +12,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	configmock "github.com/DataDog/datadog-agent/comp/core/config"
+	"github.com/DataDog/datadog-agent/pkg/aggregator"
+	"github.com/DataDog/datadog-agent/pkg/serializer"
 )
 
 func TestCreateAgentDemultiplexerOptionsNoAggWorkerCountNotReadWithoutConfigOption(t *testing.T) {
@@ -20,7 +22,7 @@ func TestCreateAgentDemultiplexerOptionsNoAggWorkerCountNotReadWithoutConfigOpti
 		"dogstatsd_no_aggregation_pipeline_workers_count": 4,
 	})
 
-	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams())
+	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(), nil)
 
 	require.Equal(t, 0, options.NoAggregationPipelineWorkersCount)
 }
@@ -31,7 +33,7 @@ func TestCreateAgentDemultiplexerOptionsNoAggWorkerCountFromConfig(t *testing.T)
 		"dogstatsd_no_aggregation_pipeline_workers_count": 4,
 	})
 
-	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()))
+	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()), nil)
 
 	require.Equal(t, 4, options.NoAggregationPipelineWorkersCount)
 }
@@ -41,7 +43,7 @@ func TestCreateAgentDemultiplexerOptionsNoAggWorkerCountDefaultsToOneWhenEnabled
 		"dogstatsd_no_aggregation_pipeline": true,
 	})
 
-	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()))
+	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()), nil)
 
 	require.Equal(t, 1, options.NoAggregationPipelineWorkersCount)
 }
@@ -52,7 +54,7 @@ func TestCreateAgentDemultiplexerOptionsNoAggWorkerCountDisabled(t *testing.T) {
 		"dogstatsd_no_aggregation_pipeline_workers_count": 4,
 	})
 
-	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()))
+	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()), nil)
 
 	require.Equal(t, 0, options.NoAggregationPipelineWorkersCount)
 }
@@ -65,9 +67,20 @@ func TestCreateAgentDemultiplexerOptionsNoAggWorkerCountFallsBackToOne(t *testin
 				"dogstatsd_no_aggregation_pipeline_workers_count": configured,
 			})
 
-			options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()))
+			options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(WithDogstatsdNoAggregationPipelineConfig()), nil)
 
 			require.Equal(t, 1, options.NoAggregationPipelineWorkersCount)
 		})
 	}
 }
+
+func TestCreateAgentDemultiplexerOptionsStoresLookbackFactory(t *testing.T) {
+	cfg := configmock.NewMock(t)
+	factory := aggregator.DogStatsDLookbackFactory(func(serializer.MetricSerializer) aggregator.DogStatsDLookback {
+		return nil
+	})
+
+	options := createAgentDemultiplexerOptions(cfg, NewDefaultParams(), factory)
+
+	require.NotNil(t, options.DogStatsDLookbackFactory)
+}

comp/metriclookback/BUILD.bazel

@@ -0,0 +1 @@
+# gazelle:ignore

comp/metriclookback/def/BUILD.bazel

@@ -0,0 +1,9 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "def",
+    srcs = ["component.go"],
+    importpath = "github.com/DataDog/datadog-agent/comp/metriclookback/def",
+    visibility = ["//visibility:public"],
+    deps = ["//pkg/aggregator/sender"],
+)

comp/metriclookback/def/component.go

@@ -0,0 +1,23 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+// Package metriclookback defines the metric lookback component.
+package metriclookback
+
+import (
+	"context"
+
+	"github.com/DataDog/datadog-agent/pkg/aggregator/sender"
+)
+
+// team: q-branch
+
+// Component is the metric lookback component.
+type Component interface {
+	// NewSenderManager returns the sender manager used exclusively by metric
+	// lookback shadow checks. It returns nil when lookback is unavailable in the
+	// current Agent build.
+	NewSenderManager(context.Context, string) sender.SenderManager
+}

comp/metriclookback/fx/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "fx",
+    srcs = ["fx.go"],
+    importpath = "github.com/DataDog/datadog-agent/comp/metriclookback/fx",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//comp/metriclookback/impl",
+        "//pkg/util/fxutil",
+    ],
+)

comp/metriclookback/fx/fx.go

@@ -0,0 +1,19 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+// Package fx provides the fx module for the metric lookback component.
+package fx
+
+import (
+	metriclookbackimpl "github.com/DataDog/datadog-agent/comp/metriclookback/impl"
+	"github.com/DataDog/datadog-agent/pkg/util/fxutil"
+)
+
+// Module defines the fx options for this component.
+func Module() fxutil.Module {
+	return fxutil.Component(
+		fxutil.ProvideComponentConstructor(metriclookbackimpl.NewComponent),
+	)
+}