configfilesdiscovery: add heartbeats (#53719)

### What does this PR do?

Implements configuration files discovery heartbeats.

When an autodiscovery config is first scheduled, the Agent collects and sends its configuration files. It then retains the collection target and periodically runs the collector again.

Each heartbeat:
  - Recollects the configuration files from the target runtime.
  - Sends every file returned by the collector.

Repeated autodiscovery `Schedule` calls do not cause collection between heartbeats. Unscheduling removes the corresponding watch and stops future collections.

Heartbeat scheduling uses configurable jitter to distribute intake traffic across Agents. Failed collections and sends are retried using a shorter jittered retry interval.

The following settings are added:
- `config_files_discovery.heartbeat_interval`
- `config_files_discovery.heartbeat_jitter`

### Motivation

DSCVR-463

### Describe how you validated your changes

Unit tests

### Additional Notes

The default configuration is:
```
  config_files_discovery:
    heartbeat_interval: 1h
    heartbeat_jitter: 10m
```

The configured jitter is applied in both directions around the heartbeat interval. Its effective value is clamped to the smaller of:
- 1 hour.
- Half of the configured heartbeat interval.

This prevents jitter from producing excessively short or negative heartbeat delays.

Co-authored-by: guillaume.pagnoux <guillaume.pagnoux@datadoghq.com>

Author: Yumasi

comp/core/configfilesdiscovery/fx/fx.go

@@ -57,6 +57,7 @@ func newOptionalComponent(reqs Requires) Provides {
 
 	provides := configfilesdiscoveryimpl.NewComponent(configfilesdiscoveryimpl.Requires{
 		Lifecycle:     reqs.Lifecycle,
+		Config:        reqs.Config,
 		Autodiscovery: reqs.Autodiscovery,
 		Hostname:      reqs.Hostname,
 		WorkloadMeta:  reqs.WorkloadMeta,

comp/core/configfilesdiscovery/impl/BUILD.bazel

@@ -20,6 +20,7 @@ go_library(
         "//comp/core/autodiscovery/def",
         "//comp/core/autodiscovery/integration",
         "//comp/core/autodiscovery/scheduler",
+        "//comp/core/config",
         "//comp/core/configfilesdiscovery/def",
         "//comp/core/hostname",
         "//comp/core/workloadmeta/def",
@@ -30,6 +31,7 @@ go_library(
         "//pkg/util/containers/cri",
         "//pkg/util/docker",
         "//pkg/util/log",
+        "@com_github_benbjohnson_clock//:clock",
         "@com_github_containerd_containerd_v2//pkg/oci",
         "@com_github_datadog_agent_payload_v5//agentdiscovery",
         "@org_golang_google_protobuf//proto",
@@ -73,6 +75,7 @@ dd_agent_go_test(
         "//comp/def",
         "//comp/forwarder/eventplatform/def",
         "//pkg/logs/message",
+        "@com_github_benbjohnson_clock//:clock",
         "@com_github_containerd_containerd_v2//pkg/oci",
         "@com_github_datadog_agent_payload_v5//agentdiscovery",
         "@com_github_opencontainers_runtime_spec//specs-go",

comp/core/configfilesdiscovery/impl/configfilesdiscovery.go

@@ -8,21 +8,31 @@ package configfilesdiscoveryimpl
 
 import (
 	"context"
+	"time"
 
 	autodiscovery "github.com/DataDog/datadog-agent/comp/core/autodiscovery/def"
 	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/scheduler"
+	"github.com/DataDog/datadog-agent/comp/core/config"
 	configfilesdiscovery "github.com/DataDog/datadog-agent/comp/core/configfilesdiscovery/def"
 	"github.com/DataDog/datadog-agent/comp/core/hostname"
 	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
 	compdef "github.com/DataDog/datadog-agent/comp/def"
 	eventplatform "github.com/DataDog/datadog-agent/comp/forwarder/eventplatform/def"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
+)
+
+const (
+	heartbeatIntervalConfigKey = "config_files_discovery.heartbeat_interval"
+	heartbeatJitterConfigKey   = "config_files_discovery.heartbeat_jitter"
+	startupJitterConfigKey     = "config_files_discovery.startup_jitter"
 )
 
 // Requires defines the dependencies for the config files discovery component.
 type Requires struct {
 	compdef.In
 
 	Lifecycle     compdef.Lifecycle
+	Config        config.Component
 	Autodiscovery autodiscovery.Component
 	Hostname      hostname.Component
 	WorkloadMeta  workloadmeta.Component
@@ -47,6 +57,16 @@ func newComponent(
 	resolver targetResolver,
 	sender collectedConfigSender,
 	configCollectors map[string]ConfigCollector,
+) *component {
+	return newComponentWithSchedulerConfig(ad, resolver, sender, configCollectors, defaultADSchedulerConfig())
+}
+
+func newComponentWithSchedulerConfig(
+	ad autodiscovery.Component,
+	resolver targetResolver,
+	sender collectedConfigSender,
+	configCollectors map[string]ConfigCollector,
+	schedulerCfg adSchedulerConfig,
 ) *component {
 	readers := map[RuntimeType]configReaderFactory{
 		RuntimeDocker:     newDockerConfigReader,
@@ -57,22 +77,71 @@ func newComponent(
 	}
 	return &component{
 		ad:        ad,
-		scheduler: newADScheduler(resolver, readers, configCollectors, sender),
+		scheduler: newADSchedulerWithConfig(resolver, readers, configCollectors, sender, schedulerCfg),
 	}
 }
 
 // NewComponent creates the config files discovery component.
 func NewComponent(reqs Requires) Provides {
-	c := newComponent(
+	schedulerCfg := defaultADSchedulerConfig()
+	if reqs.Config != nil {
+		schedulerCfg = adSchedulerConfigFromAgentConfig(reqs.Config)
+	}
+
+	c := newComponentWithSchedulerConfig(
 		reqs.Autodiscovery,
 		targetResolver{store: reqs.WorkloadMeta},
 		newEventPlatformCollectedConfigSender(reqs.EventPlatform, reqs.Hostname.GetSafe(context.Background())),
 		reqs.Collectors,
+		schedulerCfg,
 	)
 	reqs.Lifecycle.Append(compdef.Hook{OnStart: c.start, OnStop: c.stop})
 	return Provides{Comp: c}
 }
 
+func adSchedulerConfigFromAgentConfig(agentConfig config.Component) adSchedulerConfig {
+	cfg := defaultADSchedulerConfig()
+	if agentConfig == nil {
+		return cfg
+	}
+
+	heartbeatInterval := agentConfig.GetDuration(heartbeatIntervalConfigKey)
+	if heartbeatInterval <= 0 {
+		log.Warnf("configured %s must be positive, using default %s", heartbeatIntervalConfigKey, defaultHeartbeatInterval)
+	} else {
+		cfg.heartbeatInterval = heartbeatInterval
+	}
+
+	heartbeatJitter := agentConfig.GetDuration(heartbeatJitterConfigKey)
+	jitterLimit := heartbeatJitterLimit(cfg.heartbeatInterval)
+	switch {
+	case heartbeatJitter < 0:
+		log.Warnf("configured %s must be non-negative, using 0", heartbeatJitterConfigKey)
+		cfg.heartbeatJitter = 0
+	case heartbeatJitter > jitterLimit:
+		log.Warnf("configured %s exceeds maximum %s for heartbeat interval %s, clamping", heartbeatJitterConfigKey, jitterLimit, cfg.heartbeatInterval)
+		cfg.heartbeatJitter = jitterLimit
+	default:
+		cfg.heartbeatJitter = heartbeatJitter
+	}
+
+	startupJitter := agentConfig.GetDuration(startupJitterConfigKey)
+	if startupJitter < 0 {
+		log.Warnf("configured %s must be non-negative, using 0", startupJitterConfigKey)
+		cfg.startupJitter = 0
+	} else {
+		cfg.startupJitter = startupJitter
+	}
+
+	if cfg.heartbeatCheckInterval > cfg.heartbeatInterval/10 {
+		cfg.heartbeatCheckInterval = cfg.heartbeatInterval / 10
+	}
+	if cfg.heartbeatCheckInterval <= 0 {
+		cfg.heartbeatCheckInterval = time.Second
+	}
+	return cfg
+}
+
 func (c *component) start(context.Context) error {
 	c.ad.AddScheduler(schedulerName, c.scheduler, true)
 	return nil