diff --git a/cmd/agent/subcommands/run/BUILD.bazel b/cmd/agent/subcommands/run/BUILD.bazel index 10e887047385..e33c487ba9dc 100644 --- a/cmd/agent/subcommands/run/BUILD.bazel +++ b/cmd/agent/subcommands/run/BUILD.bazel @@ -44,6 +44,8 @@ go_library( "//comp/collector/collector/impl", "//comp/connectivitychecker/fx", "//comp/core", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/agenttelemetry/def", "//comp/core/agenttelemetry/fx", "//comp/core/autodiscovery/def", @@ -286,6 +288,7 @@ go_library( "//comp/trace/etwtracer/def", "//comp/trace/etwtracer/fx", "//pkg/config/model", + "//pkg/config/setup", "//pkg/util/winutil", ], "//conditions:default": [], @@ -305,5 +308,12 @@ dd_agent_go_test( "//comp/core/pid/impl", "//pkg/util/fxutil", "@com_github_stretchr_testify//require", - ], + ] + select({ + "@rules_go//go/platform:windows": [ + "//comp/core/config", + "//pkg/config/setup", + "@org_uber_go_fx//:fx", + ], + "//conditions:default": [], + }), ) diff --git a/cmd/agent/subcommands/run/command.go b/cmd/agent/subcommands/run/command.go index 3416aa989e2c..b1a92faaefa6 100644 --- a/cmd/agent/subcommands/run/command.go +++ b/cmd/agent/subcommands/run/command.go @@ -74,6 +74,8 @@ import ( collectorimpl "github.com/DataDog/datadog-agent/comp/collector/collector/impl" connectivitycheckerfx "github.com/DataDog/datadog-agent/comp/connectivitychecker/fx" "github.com/DataDog/datadog-agent/comp/core" + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + agentlifecyclefx "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/fx" autodiscovery "github.com/DataDog/datadog-agent/comp/core/autodiscovery/def" adfx "github.com/DataDog/datadog-agent/comp/core/autodiscovery/fx" "github.com/DataDog/datadog-agent/comp/core/autodiscovery/providers" @@ -235,7 +237,7 @@ func Commands(globalParams *command.GlobalParams) []*cobra.Command { config.WithExtraConfFiles(cliParams.ExtraConfFilePath), config.WithFleetPoliciesDirPath(cliParams.FleetPoliciesDirPath), } - return fxutil.OneShot(run, + return fxutil.OneShotWithStartupGate[agentlifecycle.Component](run, fx.Invoke(func(_ log.Component) { ddruntime.SetMaxProcs() }), @@ -245,6 +247,8 @@ func Commands(globalParams *command.GlobalParams) []*cobra.Command { LogParams: log.ForDaemon(command.LoggerName, "log_file", defaultpaths.GetDefaultLogFile()), }), fx.Supply(pidimpl.NewParams(cliParams.pidfilePath)), + fx.Supply(agentlifecycle.Params{ComponentName: "agent"}), + agentlifecyclefx.Module(), logging.EnableFxLoggingOnDebug[log.Component](), fxinstrumentation.Module(), getSharedFxOption(), @@ -392,7 +396,6 @@ func run(log log.Component, ); err != nil { return err } - agentStarted := tlm.NewCounter( "runtime", "started", diff --git a/cmd/agent/subcommands/run/command_windows.go b/cmd/agent/subcommands/run/command_windows.go index 0cf67e08ba4f..395177394be0 100644 --- a/cmd/agent/subcommands/run/command_windows.go +++ b/cmd/agent/subcommands/run/command_windows.go @@ -10,6 +10,7 @@ package run import ( "context" + "errors" _ "expvar" // Blank import used because this isn't directly used in this file _ "net/http/pprof" // Blank import used because this isn't directly used in this file @@ -84,6 +85,7 @@ import ( rcclient "github.com/DataDog/datadog-agent/comp/remote-config/rcclient/def" snmpscanmanager "github.com/DataDog/datadog-agent/comp/snmpscanmanager/def" softwareinventoryfx "github.com/DataDog/datadog-agent/comp/softwareinventory/fx" + configsetup "github.com/DataDog/datadog-agent/pkg/config/setup" "github.com/DataDog/datadog-agent/pkg/serializer" "github.com/DataDog/datadog-agent/pkg/util/defaultpaths" "github.com/DataDog/datadog-agent/pkg/util/fxutil" @@ -210,6 +212,7 @@ func StartAgentWithDefaults(ctxChan <-chan context.Context) (<-chan error, error SysprobeConfigParams: sysprobeconfigimpl.NewParams(), LogParams: log.ForDaemon(command.LoggerName, "log_file", defaultpaths.GetDefaultLogFile()), }), + fx.Invoke(validateWindowsPreparedRollout), getSharedFxOption(), getPlatformModules(), ) @@ -228,6 +231,13 @@ func StartAgentWithDefaults(ctxChan <-chan context.Context) (<-chan error, error return errChan, nil } +func validateWindowsPreparedRollout(config config.Component) error { + if config.GetBool(configsetup.ExperimentalNodeAgentRolloutEnabled) { + return errors.New("experimental node Agent rollout is not supported by the Windows service") + } + return nil +} + // Re-register the ctrl handler because Go uses SetConsoleCtrlHandler and Python uses posix signal calls. // This is only needed when using the embedded Python module. // When Python imports the signal module, it overrides the ctrl handler set by Go and installs the Windows posix signal handler. diff --git a/cmd/agent/subcommands/run/command_windows_test.go b/cmd/agent/subcommands/run/command_windows_test.go index a3937ab9fb2a..44d0ac11ce62 100644 --- a/cmd/agent/subcommands/run/command_windows_test.go +++ b/cmd/agent/subcommands/run/command_windows_test.go @@ -11,8 +11,11 @@ import ( "context" "testing" + coreconfig "github.com/DataDog/datadog-agent/comp/core/config" + configsetup "github.com/DataDog/datadog-agent/pkg/config/setup" "github.com/DataDog/datadog-agent/pkg/util/fxutil" "github.com/stretchr/testify/require" + "go.uber.org/fx" ) func TestStartAgentWithDefaults(t *testing.T) { @@ -23,3 +26,22 @@ func TestStartAgentWithDefaults(t *testing.T) { require.NoError(t, err) }) } + +func TestWindowsServiceRejectsPreparedRolloutBeforeStart(t *testing.T) { + cfg := coreconfig.NewMockWithOverrides(t, map[string]interface{}{ + configsetup.ExperimentalNodeAgentRolloutEnabled: true, + }) + started := false + app := fx.New( + fx.Provide(func() coreconfig.Component { return cfg }), + fx.Invoke(validateWindowsPreparedRollout), + fx.Invoke(func(lifecycle fx.Lifecycle) { + lifecycle.Append(fx.Hook{OnStart: func(context.Context) error { + started = true + return nil + }}) + }), + ) + require.ErrorContains(t, app.Err(), "not supported") + require.False(t, started) +} diff --git a/cmd/host-profiler/subcommands/run/BUILD.bazel b/cmd/host-profiler/subcommands/run/BUILD.bazel index 00c002275c99..168fc5d8feac 100644 --- a/cmd/host-profiler/subcommands/run/BUILD.bazel +++ b/cmd/host-profiler/subcommands/run/BUILD.bazel @@ -11,6 +11,8 @@ go_library( "//cmd/agent/command", "//cmd/host-profiler/globalparams", "//comp/core", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/config", "//comp/core/configsync/def", "//comp/core/configsync/fx", @@ -46,6 +48,8 @@ go_library( "//cmd/agent/command", "//cmd/host-profiler/globalparams", "//comp/core", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/config", "//comp/core/configsync/def", "//comp/core/configsync/fx", diff --git a/cmd/host-profiler/subcommands/run/command.go b/cmd/host-profiler/subcommands/run/command.go index 57136a5a1e0f..f64a05f6c4e5 100644 --- a/cmd/host-profiler/subcommands/run/command.go +++ b/cmd/host-profiler/subcommands/run/command.go @@ -22,6 +22,8 @@ import ( "github.com/DataDog/datadog-agent/cmd/agent/command" "github.com/DataDog/datadog-agent/cmd/host-profiler/globalparams" "github.com/DataDog/datadog-agent/comp/core" + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + agentlifecyclefx "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/fx" "github.com/DataDog/datadog-agent/comp/core/config" configsync "github.com/DataDog/datadog-agent/comp/core/configsync/def" configsyncfx "github.com/DataDog/datadog-agent/comp/core/configsync/fx" @@ -90,6 +92,8 @@ func runHostProfilerCommand(ctx context.Context, cliParams *cliParams) error { } var opts = []fx.Option{ + fx.Supply(agentlifecycle.Params{ComponentName: "host-profiler"}), + agentlifecyclefx.Module(), hostprofiler.Bundle(collectorimpl.NewParams(cliParams.GlobalParams.ConfigURI(), cliParams.GoRuntimeMetrics)), logging.DefaultFxLoggingOption(), } @@ -120,7 +124,7 @@ func runHostProfilerCommand(ctx context.Context, cliParams *cliParams) error { ) } - return fxutil.OneShot(run, opts...) + return fxutil.OneShotWithStartupGate[agentlifecycle.Component](run, opts...) } func run(collector collector.Component) error { diff --git a/cmd/otel-agent/subcommands/run/BUILD.bazel b/cmd/otel-agent/subcommands/run/BUILD.bazel index f25e2b0c1ff8..6a04b6978a04 100644 --- a/cmd/otel-agent/subcommands/run/BUILD.bazel +++ b/cmd/otel-agent/subcommands/run/BUILD.bazel @@ -15,6 +15,8 @@ go_library( deps = [ "//cmd/otel-agent/config", "//cmd/otel-agent/subcommands", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/agenttelemetry/fx", "//comp/core/config", "//comp/core/configsync/def", diff --git a/cmd/otel-agent/subcommands/run/command.go b/cmd/otel-agent/subcommands/run/command.go index d2a1441d05ac..5c13f90b434b 100644 --- a/cmd/otel-agent/subcommands/run/command.go +++ b/cmd/otel-agent/subcommands/run/command.go @@ -20,6 +20,8 @@ import ( agentConfig "github.com/DataDog/datadog-agent/cmd/otel-agent/config" "github.com/DataDog/datadog-agent/cmd/otel-agent/subcommands" + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + agentlifecyclefx "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/fx" agenttelemetryfx "github.com/DataDog/datadog-agent/comp/core/agenttelemetry/fx" coreconfig "github.com/DataDog/datadog-agent/comp/core/config" configsync "github.com/DataDog/datadog-agent/comp/core/configsync/def" @@ -132,11 +134,16 @@ func runOTelAgentCommand(ctx context.Context, params *cliParams, opts ...fx.Opti fmt.Println("*** OpenTelemetry Collector is not enabled, exiting application ***. Set the config option `otelcollector.enabled` or the environment variable `DD_OTELCOLLECTOR_ENABLED` at true to enable it.") return nil } + gateOptions := []fx.Option{ + fx.Supply(agentlifecycle.Params{ComponentName: "otel-agent"}), + agentlifecyclefx.Module(), + } uris := buildConfigURIs(params) if err == agentConfig.ErrNoDDExporter { - return fxutil.Run( + return fxutil.RunWithStartupGate[agentlifecycle.Component]( + fx.Options(gateOptions...), fx.Supply(uris), fx.Provide(func() coreconfig.Component { return acfg @@ -170,12 +177,14 @@ func runOTelAgentCommand(ctx context.Context, params *cliParams, opts ...fx.Opti } if acfg.GetBool("otel_standalone") { - return fxutil.Run( + return fxutil.RunWithStartupGate[agentlifecycle.Component]( + fx.Options(gateOptions...), commonAgentFxOptions(ctx, params, acfg, uris, opts...), standaloneAgentFxOptions(params), ) } - return fxutil.Run( + return fxutil.RunWithStartupGate[agentlifecycle.Component]( + fx.Options(gateOptions...), commonAgentFxOptions(ctx, params, acfg, uris, opts...), connectedAgentFxOptions(params), ) diff --git a/cmd/privateactionrunner/subcommands/run/BUILD.bazel b/cmd/privateactionrunner/subcommands/run/BUILD.bazel index 6aa84c5c8f0c..084345047c2b 100644 --- a/cmd/privateactionrunner/subcommands/run/BUILD.bazel +++ b/cmd/privateactionrunner/subcommands/run/BUILD.bazel @@ -14,6 +14,8 @@ go_library( deps = [ "//cmd/privateactionrunner/command", "//comp/core", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/config", "//comp/core/hostname/remotehostnameimpl", "//comp/core/ipc/fx", diff --git a/cmd/privateactionrunner/subcommands/run/command.go b/cmd/privateactionrunner/subcommands/run/command.go index fccbfaf4ccc8..c72d7a35fee9 100644 --- a/cmd/privateactionrunner/subcommands/run/command.go +++ b/cmd/privateactionrunner/subcommands/run/command.go @@ -21,6 +21,8 @@ import ( "github.com/DataDog/datadog-agent/cmd/privateactionrunner/command" "github.com/DataDog/datadog-agent/comp/core" + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + agentlifecyclefx "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/fx" "github.com/DataDog/datadog-agent/comp/core/config" ipcfx "github.com/DataDog/datadog-agent/comp/core/ipc/fx" log "github.com/DataDog/datadog-agent/comp/core/log/def" @@ -103,6 +105,8 @@ func runPrivateActionRunner(ctx context.Context, confPath string, extraConfFiles ConfigParams: config.NewAgentParams(confPath, config.WithExtraConfFiles(extraConfFiles)), LogParams: log.ForDaemon(command.LoggerName, pkgconfigsetup.PARLogFile, defaultpaths.GetDefaultPrivateActionRunnerLogFile())}), core.Bundle(core.WithSecrets()), + fx.Supply(agentlifecycle.Params{ComponentName: "private-action-runner"}), + agentlifecyclefx.Module(), fx.Provide(func(c config.Component) settings.Params { return settings.Params{ Settings: map[string]settings.RuntimeSetting{ @@ -127,7 +131,7 @@ func runPrivateActionRunner(ctx context.Context, confPath string, extraConfFiles privateactionrunnerfx.Module(), } - err := fxutil.Run(fxOptions...) + err := fxutil.RunWithStartupGate[agentlifecycle.Component](fxOptions...) if errors.Is(err, privateactionrunner.ErrNotEnabled) { return nil } diff --git a/cmd/process-agent/command/BUILD.bazel b/cmd/process-agent/command/BUILD.bazel index f54e82900374..9641fedf42b8 100644 --- a/cmd/process-agent/command/BUILD.bazel +++ b/cmd/process-agent/command/BUILD.bazel @@ -17,6 +17,8 @@ go_library( "//comp/agent/autoexit/def", "//comp/agent/autoexit/fx", "//comp/core", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/config", "//comp/core/configstreamconsumer/def", "//comp/core/configstreamconsumer/fx", diff --git a/cmd/process-agent/command/main_common.go b/cmd/process-agent/command/main_common.go index 07491589512c..3b262ce185a8 100644 --- a/cmd/process-agent/command/main_common.go +++ b/cmd/process-agent/command/main_common.go @@ -18,6 +18,8 @@ import ( autoexit "github.com/DataDog/datadog-agent/comp/agent/autoexit/def" autoexitfx "github.com/DataDog/datadog-agent/comp/agent/autoexit/fx" "github.com/DataDog/datadog-agent/comp/core" + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + agentlifecyclefx "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/fx" "github.com/DataDog/datadog-agent/comp/core/config" configstreamconsumer "github.com/DataDog/datadog-agent/comp/core/configstreamconsumer/def" configstreamconsumerfx "github.com/DataDog/datadog-agent/comp/core/configstreamconsumer/fx" @@ -108,7 +110,9 @@ func runApp(ctx context.Context, globalParams *GlobalParams) error { RCClient rcclient.Component WorkloadMeta workloadmeta.Component } + var startupGate fxutil.StartupGate app := fx.New( + fxutil.StartupGateOption[agentlifecycle.Component](ctx, &startupGate), fx.Supply( core.BundleParams{ SysprobeConfigParams: sysprobeconfigimpl.NewParams( @@ -119,6 +123,8 @@ func runApp(ctx context.Context, globalParams *GlobalParams) error { LogParams: DaemonLogParams, }, ), + fx.Supply(agentlifecycle.Params{ComponentName: "process-agent"}), + agentlifecyclefx.Module(), fx.Supply( status.Params{ PythonVersionGetFunc: python.GetPythonVersion, @@ -233,6 +239,15 @@ func runApp(ctx context.Context, globalParams *GlobalParams) error { return nil }), ) + if err := app.Err(); err != nil && !errors.Is(err, errAgentDisabled) { + if startupGate != nil { + return errors.Join(fxutil.UnwrapIfErrArgumentsFailed(err), startupGate.Close()) + } + return fxutil.UnwrapIfErrArgumentsFailed(err) + } + if startupGate != nil { + defer startupGate.Close() + } err := app.Start(ctx) if err != nil { @@ -252,7 +267,6 @@ func runApp(ctx context.Context, globalParams *GlobalParams) error { return err } } - // Wait for exit signal select { case <-exitSignal: diff --git a/cmd/system-probe/subcommands/run/BUILD.bazel b/cmd/system-probe/subcommands/run/BUILD.bazel index d3bb1fca9d8c..51f4806c139e 100644 --- a/cmd/system-probe/subcommands/run/BUILD.bazel +++ b/cmd/system-probe/subcommands/run/BUILD.bazel @@ -17,6 +17,8 @@ go_library( "//cmd/system-probe/common", "//comp/agent/autoexit/def", "//comp/agent/autoexit/fx", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/config", "//comp/core/configstreamconsumer/def", "//comp/core/configstreamconsumer/fx", diff --git a/cmd/system-probe/subcommands/run/command.go b/cmd/system-probe/subcommands/run/command.go index 57591d2c7839..ff0b502d48cc 100644 --- a/cmd/system-probe/subcommands/run/command.go +++ b/cmd/system-probe/subcommands/run/command.go @@ -28,6 +28,8 @@ import ( "github.com/DataDog/datadog-agent/cmd/system-probe/common" autoexit "github.com/DataDog/datadog-agent/comp/agent/autoexit/def" autoexitfx "github.com/DataDog/datadog-agent/comp/agent/autoexit/fx" + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + agentlifecyclefx "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/fx" "github.com/DataDog/datadog-agent/comp/core/config" configstreamconsumer "github.com/DataDog/datadog-agent/comp/core/configstreamconsumer/def" configstreamconsumerfx "github.com/DataDog/datadog-agent/comp/core/configstreamconsumer/fx" @@ -122,6 +124,8 @@ func Commands(globalParams *command.GlobalParams) []*cobra.Command { Long: `Runs the system-probe in the foreground`, RunE: func(_ *cobra.Command, _ []string) error { opts := []fx.Option{ + fx.Supply(agentlifecycle.Params{ComponentName: "system-probe"}), + agentlifecyclefx.Module(), fx.Invoke(func(_ log.Component) { ddruntime.SetMaxProcs() }), @@ -135,7 +139,7 @@ func Commands(globalParams *command.GlobalParams) []*cobra.Command { configstreamconsumerfx.Module(), getSharedFxOption(), } - return fxutil.OneShot(run, opts...) + return fxutil.OneShotWithStartupGate[agentlifecycle.Component](run, opts...) }, } runCmd.Flags().StringVarP(&cliParams.pidfilePath, "pid", "p", "", "path to the pidfile") diff --git a/cmd/trace-agent/subcommands/run/BUILD.bazel b/cmd/trace-agent/subcommands/run/BUILD.bazel index e097ff24924e..d0a9a90c68b1 100644 --- a/cmd/trace-agent/subcommands/run/BUILD.bazel +++ b/cmd/trace-agent/subcommands/run/BUILD.bazel @@ -14,6 +14,8 @@ go_library( "//cmd/trace-agent/subcommands", "//comp/agent/autoexit/def", "//comp/agent/autoexit/fx", + "//comp/core/agentlifecycle/def", + "//comp/core/agentlifecycle/fx", "//comp/core/agenttelemetry/def", "//comp/core/agenttelemetry/fx", "//comp/core/config", diff --git a/cmd/trace-agent/subcommands/run/command.go b/cmd/trace-agent/subcommands/run/command.go index c0cb625ce004..a1528fb7a224 100644 --- a/cmd/trace-agent/subcommands/run/command.go +++ b/cmd/trace-agent/subcommands/run/command.go @@ -16,6 +16,8 @@ import ( "github.com/DataDog/datadog-agent/cmd/trace-agent/subcommands" autoexit "github.com/DataDog/datadog-agent/comp/agent/autoexit/def" autoexitfx "github.com/DataDog/datadog-agent/comp/agent/autoexit/fx" + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + agentlifecyclefx "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/fx" agenttelemetry "github.com/DataDog/datadog-agent/comp/core/agenttelemetry/def" agenttelemetryfx "github.com/DataDog/datadog-agent/comp/core/agenttelemetry/fx" coreconfig "github.com/DataDog/datadog-agent/comp/core/config" @@ -81,11 +83,13 @@ func runTraceAgentProcess(ctx context.Context, cliParams *Params, defaultConfPat if cliParams.ConfPath == "" { cliParams.ConfPath = defaultConfPath } - err := fxutil.Run( + err := fxutil.RunWithStartupGate[agentlifecycle.Component]( // ctx is required to be supplied from here, as Windows needs to inject its own context // to allow the agent to work as a service. fx.Provide(func() context.Context { return ctx }), // fx.Supply(ctx) fails with a missing type error. fx.Supply(coreconfig.NewAgentParams(cliParams.ConfPath, coreconfig.WithFleetPoliciesDirPath(cliParams.FleetPoliciesDirPath))), + fx.Supply(agentlifecycle.Params{ComponentName: "trace-agent"}), + agentlifecyclefx.Module(), secretsfx.Module(), delegatedauthfx.Module(), telemetryfx.Module(), diff --git a/comp/README.md b/comp/README.md index ef04613ecaba..491b0dc13614 100644 --- a/comp/README.md +++ b/comp/README.md @@ -99,6 +99,10 @@ Package collector defines the collector component. Package core implements the "core" bundle, providing services common to all agent flavors and binaries. +### [comp/core/agentlifecycle](https://pkg.go.dev/github.com/DataDog/datadog-agent/comp/core/agentlifecycle) + +Package agentlifecycle defines the experimental Agent startup gate. + ### [comp/core/agenttelemetry](https://pkg.go.dev/github.com/DataDog/datadog-agent/comp/core/agenttelemetry) Package agenttelemetry implements a component to generate Agent telemetry diff --git a/comp/core/agentlifecycle/def/BUILD.bazel b/comp/core/agentlifecycle/def/BUILD.bazel new file mode 100644 index 000000000000..df4b1ab12f4a --- /dev/null +++ b/comp/core/agentlifecycle/def/BUILD.bazel @@ -0,0 +1,8 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "def", + srcs = ["component.go"], + importpath = "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def", + visibility = ["//visibility:public"], +) diff --git a/comp/core/agentlifecycle/def/component.go b/comp/core/agentlifecycle/def/component.go new file mode 100644 index 000000000000..98ce22e3cd04 --- /dev/null +++ b/comp/core/agentlifecycle/def/component.go @@ -0,0 +1,24 @@ +// 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 2026-present Datadog, Inc. + +// Package agentlifecycle defines the experimental Agent startup gate. +package agentlifecycle + +import "context" + +// team: agent-runtimes + +// Params identifies the Agent process using the lifecycle gate. +type Params struct { + ComponentName string +} + +// Component gates Agent construction until the older same-name container has stopped. +type Component interface { + // Wait blocks until no older same-name container owned by the same DaemonSet remains on the node. + Wait(context.Context) error + // Close releases gate resources. + Close() error +} diff --git a/comp/core/agentlifecycle/fx/BUILD.bazel b/comp/core/agentlifecycle/fx/BUILD.bazel new file mode 100644 index 000000000000..fcbbb058b9c8 --- /dev/null +++ b/comp/core/agentlifecycle/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/core/agentlifecycle/fx", + visibility = ["//visibility:public"], + deps = [ + "//comp/core/agentlifecycle/impl", + "//pkg/util/fxutil", + ], +) diff --git a/comp/core/agentlifecycle/fx/fx.go b/comp/core/agentlifecycle/fx/fx.go new file mode 100644 index 000000000000..7adcdfd60126 --- /dev/null +++ b/comp/core/agentlifecycle/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 2026-present Datadog, Inc. + +// Package fx provides the experimental Agent lifecycle component. +package fx + +import ( + agentlifecycleimpl "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/impl" + "github.com/DataDog/datadog-agent/pkg/util/fxutil" +) + +// Module defines the fx options for the experimental Agent lifecycle component. +func Module() fxutil.Module { + return fxutil.Component( + fxutil.ProvideComponentConstructor(agentlifecycleimpl.NewComponent), + ) +} diff --git a/comp/core/agentlifecycle/impl/BUILD.bazel b/comp/core/agentlifecycle/impl/BUILD.bazel new file mode 100644 index 000000000000..83a4da49a6a1 --- /dev/null +++ b/comp/core/agentlifecycle/impl/BUILD.bazel @@ -0,0 +1,32 @@ +load("@rules_go//go:def.bzl", "go_library") +load("//bazel/rules/go:dd_agent_go_test.bzl", "dd_agent_go_test") + +go_library( + name = "impl", + srcs = [ + "agent_lifecycle.go", + "local_pods_kubelet.go", + "local_pods_nokubelet.go", + ], + importpath = "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/impl", + visibility = ["//visibility:public"], + deps = [ + "//comp/core/agentlifecycle/def", + "//comp/core/config", + "//comp/core/log/def", + "//comp/def", + "//pkg/util/kubernetes/kubelet", + ], +) + +dd_agent_go_test( + name = "impl_test", + srcs = ["agent_lifecycle_test.go"], + embed = [":impl"], + deps = [ + "//comp/core/agentlifecycle/def", + "//comp/core/config", + "//comp/core/log/mock", + "@com_github_stretchr_testify//require", + ], +) diff --git a/comp/core/agentlifecycle/impl/agent_lifecycle.go b/comp/core/agentlifecycle/impl/agent_lifecycle.go new file mode 100644 index 000000000000..db72f4ab6ed2 --- /dev/null +++ b/comp/core/agentlifecycle/impl/agent_lifecycle.go @@ -0,0 +1,299 @@ +// 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 2026-present Datadog, Inc. + +// Package agentlifecycleimpl implements the experimental Agent startup gate. +package agentlifecycleimpl + +import ( + "context" + "errors" + "fmt" + "runtime" + "strings" + "time" + + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + "github.com/DataDog/datadog-agent/comp/core/config" + log "github.com/DataDog/datadog-agent/comp/core/log/def" + compdef "github.com/DataDog/datadog-agent/comp/def" +) + +const ( + siblingPollInterval = time.Second + rolloutEnabledKey = "experimental.node_agent_rollout.enabled" + rolloutPodUIDKey = "experimental.node_agent_rollout.pod_uid" + + freshInstallObservations = 2 + missingOlderObservations = 2 +) + +type podOwner struct { + kind string + uid string + controller bool +} + +type localContainer struct { + name string + terminated bool + ready bool +} + +type localPod struct { + uid string + name string + namespace string + createdAt time.Time + deletionTimestamp *time.Time + owners []podOwner + declaredContainers []string + containers []localContainer +} + +type localPodSource interface { + ListLocalPods(context.Context) ([]localPod, error) +} + +type dependencies struct { + compdef.In + + Config config.Component + Log log.Component + Params agentlifecycle.Params +} + +type component struct { + enabled bool + componentName string + podUID string + log log.Component + pods localPodSource + pollInterval time.Duration +} + +var _ agentlifecycle.Component = (*component)(nil) + +// NewComponent creates the experimental Agent startup gate. +func NewComponent(deps dependencies) (agentlifecycle.Component, error) { + return newComponent(deps, newLocalPodSource(), runtime.GOOS) +} + +func newComponent(deps dependencies, pods localPodSource, goos string) (agentlifecycle.Component, error) { + if !deps.Config.GetBool(rolloutEnabledKey) { + return &component{}, nil + } + if goos != "linux" { + return nil, fmt.Errorf("experimental node Agent rollout is Linux-only (running on %s)", goos) + } + if deps.Params.ComponentName == "" || strings.ContainsAny(deps.Params.ComponentName, `/\\`) { + return nil, errors.New("experimental node Agent rollout requires a path-safe component name") + } + podUID := strings.TrimSpace(deps.Config.GetString(rolloutPodUIDKey)) + if podUID == "" { + return nil, fmt.Errorf("%s must identify this Pod", rolloutPodUIDKey) + } + c := &component{ + enabled: true, + componentName: deps.Params.ComponentName, + podUID: podUID, + log: deps.Log, + pods: pods, + pollInterval: siblingPollInterval, + } + return c, nil +} + +// Wait leaves the executable asleep before later Fx construction until the +// older instance of this same container has stopped. Kubelet errors fail closed. +func (c *component) Wait(ctx context.Context) (err error) { + if !c.enabled { + return nil + } + ticker := time.NewTicker(c.pollInterval) + defer ticker.Stop() + // The value records that kubelet has shown deletion beginning for the Pod. + // A later omission is only a safe handoff signal after that observation. + knownOlder := map[string]bool{} + missingOlder := map[string]int{} + olderObservedLogged := false + emptyObservations := 0 + pollErrorLogged := false + + for { + pods, err := c.pods.ListLocalPods(ctx) + if err == nil { + var older []localPod + older, err = olderSiblingPods(pods, c.podUID) + if err == nil { + for i := range older { + knownOlder[older[i].uid] = knownOlder[older[i].uid] || older[i].deletionTimestamp != nil + } + + if len(knownOlder) == 0 { + emptyObservations++ + if emptyObservations >= freshInstallObservations && (c.componentName == "agent" || replacementCoreReady(pods, c.podUID)) { + c.log.Infof("%s found no older container on the node and is starting", c.componentName) + return nil + } + } else { + emptyObservations = 0 + if !olderObservedLogged { + olderObservedLogged = true + c.log.Infof("%s is waiting for its older container to stop", c.componentName) + } + if err == nil && olderContainersStopped(older, knownOlder, missingOlder, c.componentName) { + if c.componentName == "agent" { + c.log.Info("the older core Agent container stopped; starting") + return nil + } + if replacementCoreReady(pods, c.podUID) { + c.log.Infof("the older %s container stopped and the replacement core Agent is ready; starting", c.componentName) + return nil + } + } + } + } + } + if err != nil { + if !pollErrorLogged { + c.log.Warnf("%s cannot verify node-local containers; remaining asleep: %v", c.componentName, err) + pollErrorLogged = true + } + } else { + pollErrorLogged = false + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +func replacementCoreReady(pods []localPod, selfUID string) bool { + for i := range pods { + if pods[i].uid != selfUID { + continue + } + for j := range pods[i].containers { + container := pods[i].containers[j] + if container.name == "agent" { + return container.ready && !container.terminated + } + } + return false + } + return false +} + +func olderSiblingPods(pods []localPod, selfUID string) ([]localPod, error) { + var self *localPod + for i := range pods { + if pods[i].uid == selfUID { + if self != nil { + return nil, fmt.Errorf("kubelet returned duplicate entries for self Pod UID %q", selfUID) + } + self = &pods[i] + } + } + if self == nil { + return nil, fmt.Errorf("self Pod UID %q is absent from the kubelet Pod list", selfUID) + } + ownerUID, err := daemonSetOwnerUID(*self) + if err != nil { + return nil, err + } + if self.createdAt.IsZero() { + return nil, fmt.Errorf("self Pod %s/%s has no creation timestamp", self.namespace, self.name) + } + + var older []localPod + for i := range pods { + candidate := pods[i] + if candidate.uid == selfUID || candidate.namespace != self.namespace { + continue + } + candidateOwnerUID, ownerErr := daemonSetOwnerUID(candidate) + if ownerErr != nil || candidateOwnerUID != ownerUID { + continue + } + if candidate.createdAt.IsZero() || candidate.createdAt.Before(self.createdAt) { + older = append(older, candidate) + continue + } + if candidate.createdAt.Equal(self.createdAt) { + return nil, fmt.Errorf("cannot order same-timestamp Pods %s/%s and %s/%s", candidate.namespace, candidate.name, self.namespace, self.name) + } + } + return older, nil +} + +func olderContainersStopped(current []localPod, known map[string]bool, missing map[string]int, componentName string) bool { + byUID := make(map[string]localPod, len(current)) + for i := range current { + byUID[current[i].uid] = current[i] + } + for uid, deletionObserved := range known { + pod, present := byUID[uid] + if !present { + missing[uid]++ + if !deletionObserved && missing[uid] < missingOlderObservations { + return false + } + continue + } + missing[uid] = 0 + if olderContainerStillExists(pod, componentName) { + return false + } + } + return len(known) > 0 +} + +func olderContainerStillExists(pod localPod, componentName string) bool { + for i := range pod.containers { + container := pod.containers[i] + if container.name != componentName { + continue + } + // A crashed container may be Terminated briefly before kubelet restarts + // it. Only use Terminated as a handoff signal once Pod deletion proves + // that Kubernetes intends to stop the old generation. + return pod.deletionTimestamp == nil || !container.terminated + } + declared := false + for _, name := range pod.declaredContainers { + declared = declared || name == componentName + } + if !declared { + return false + } + // A missing status is not positive evidence that the old runtime process + // exited. Wait for either a Terminated state or Pod removal after deletion + // was observed. + return true +} + +func daemonSetOwnerUID(pod localPod) (string, error) { + var ownerUID string + for _, owner := range pod.owners { + if owner.kind != "DaemonSet" || !owner.controller { + continue + } + if owner.uid == "" || ownerUID != "" { + return "", fmt.Errorf("Pod %s/%s has an invalid DaemonSet controller", pod.namespace, pod.name) + } + ownerUID = owner.uid + } + if ownerUID == "" { + return "", fmt.Errorf("Pod %s/%s is not controlled by a DaemonSet", pod.namespace, pod.name) + } + return ownerUID, nil +} + +func (c *component) Close() error { + return nil +} diff --git a/comp/core/agentlifecycle/impl/agent_lifecycle_test.go b/comp/core/agentlifecycle/impl/agent_lifecycle_test.go new file mode 100644 index 000000000000..d41c7e197d7b --- /dev/null +++ b/comp/core/agentlifecycle/impl/agent_lifecycle_test.go @@ -0,0 +1,290 @@ +// 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 2026-present Datadog, Inc. + +//go:build test + +package agentlifecycleimpl + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + agentlifecycle "github.com/DataDog/datadog-agent/comp/core/agentlifecycle/def" + "github.com/DataDog/datadog-agent/comp/core/config" + logmock "github.com/DataDog/datadog-agent/comp/core/log/mock" +) + +const ( + selfPodUID = "new-pod-uid" + daemonUID = "daemonset-uid" +) + +type scriptedPodSource struct { + mu sync.Mutex + responses []podResponse + calls chan struct{} +} + +type podResponse struct { + pods []localPod + err error +} + +func (s *scriptedPodSource) ListLocalPods(context.Context) ([]localPod, error) { + s.mu.Lock() + defer s.mu.Unlock() + select { + case s.calls <- struct{}{}: + default: + } + if len(s.responses) == 0 { + return nil, errors.New("no scripted kubelet response") + } + response := s.responses[0] + if len(s.responses) > 1 { + s.responses = s.responses[1:] + } + return response.pods, response.err +} + +func (s *scriptedPodSource) setResponses(responses ...podResponse) { + s.mu.Lock() + defer s.mu.Unlock() + s.responses = responses +} + +func TestDisabledLifecycleIsNoop(t *testing.T) { + deps := dependencies{Config: config.NewMock(t), Log: logmock.New(t)} + comp, err := newComponent(deps, nil, "linux") + require.NoError(t, err) + require.NoError(t, comp.Wait(context.Background())) + require.NoError(t, comp.Close()) +} + +func TestFreshPodStartsAfterTwoSuccessfulObservations(t *testing.T) { + comp, source := newEnabledComponent(t, podResponse{pods: []localPod{selfPod()}}) + require.NoError(t, comp.Wait(context.Background())) + require.GreaterOrEqual(t, len(source.calls), freshInstallObservations) +} + +func TestReplacementWaitsForSameContainerToTerminate(t *testing.T) { + old := siblingPod("old-pod-uid", "old-agent") + old.containers = []localContainer{{name: "test-agent"}} + comp, source := newEnabledComponent(t, podResponse{pods: []localPod{selfPod(), old}}) + + result := make(chan error, 1) + go func() { result <- comp.Wait(context.Background()) }() + <-source.calls + <-source.calls + select { + case err := <-result: + t.Fatalf("replacement started while old container was running: %v", err) + default: + } + + now := time.Now() + terminatedOld := siblingPod("old-pod-uid", "old-agent") + terminatedOld.deletionTimestamp = &now + terminatedOld.containers = []localContainer{{name: "test-agent", terminated: true}} + source.setResponses(podResponse{pods: []localPod{selfPod(), terminatedOld}}) + require.NoError(t, <-result) +} + +func TestCrashedOlderContainerDoesNotReleaseGate(t *testing.T) { + old := siblingPod("old-pod-uid", "old-agent") + old.containers = []localContainer{{name: "test-agent", terminated: true}} + known := map[string]bool{old.uid: false} + missing := map[string]int{} + require.False(t, olderContainersStopped([]localPod{old}, known, missing, "test-agent")) + + now := time.Now() + old.deletionTimestamp = &now + known[old.uid] = true + require.True(t, olderContainersStopped([]localPod{old}, known, missing, "test-agent")) +} + +func TestContainersHandOffIndependently(t *testing.T) { + old := siblingPod("old-pod-uid", "old-agent") + old.containers = []localContainer{ + {name: "agent", terminated: true}, + {name: "system-probe"}, + } + now := time.Now() + old.deletionTimestamp = &now + known := map[string]bool{old.uid: true} + missing := map[string]int{} + require.True(t, olderContainersStopped([]localPod{old}, known, missing, "agent")) + require.False(t, olderContainersStopped([]localPod{old}, known, missing, "system-probe")) +} + +func TestNonCoreWaitsForReplacementCoreReadiness(t *testing.T) { + old := siblingPod("old-pod-uid", "old-agent") + old.containers = []localContainer{{name: "test-agent", terminated: true}} + now := time.Now() + old.deletionTimestamp = &now + self := selfPod() + self.containers[0].ready = false + comp, source := newEnabledComponent(t, podResponse{pods: []localPod{self, old}}) + + result := make(chan error, 1) + go func() { result <- comp.Wait(context.Background()) }() + <-source.calls + <-source.calls + select { + case err := <-result: + t.Fatalf("non-core component started before replacement core was ready: %v", err) + default: + } + readySelf := selfPod() + readySelf.containers[0].ready = true + source.setResponses(podResponse{pods: []localPod{readySelf, old}}) + require.NoError(t, <-result) +} + +func TestDisappearedOlderPodReleasesGateAfterDeletionWasObserved(t *testing.T) { + old := siblingPod("old-pod-uid", "old-agent") + old.containers = []localContainer{{name: "test-agent"}} + known := map[string]bool{old.uid: false} + missing := map[string]int{} + require.False(t, olderContainersStopped([]localPod{old}, known, missing, "test-agent")) + + now := time.Now() + old.deletionTimestamp = &now + known[old.uid] = true + require.False(t, olderContainersStopped([]localPod{old}, known, missing, "test-agent")) + require.True(t, olderContainersStopped(nil, known, missing, "test-agent")) +} + +func TestMissingOlderPodWithoutObservedDeletionRequiresTwoSnapshots(t *testing.T) { + known := map[string]bool{"old-pod-uid": false} + missing := map[string]int{} + require.False(t, olderContainersStopped(nil, known, missing, "test-agent")) + require.True(t, olderContainersStopped(nil, known, missing, "test-agent")) +} + +func TestKubeletErrorsFailClosed(t *testing.T) { + comp, source := newEnabledComponent(t, podResponse{err: errors.New("kubelet unavailable")}) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { result <- comp.Wait(ctx) }() + <-source.calls + <-source.calls + select { + case err := <-result: + t.Fatalf("kubelet error opened the gate: %v", err) + default: + } + cancel() + require.ErrorIs(t, <-result, context.Canceled) +} + +func TestMissingContainerStatusRequiresPodDeletion(t *testing.T) { + old := siblingPod("old-pod-uid", "old-agent") + old.declaredContainers = []string{"test-agent"} + known := map[string]bool{old.uid: false} + missing := map[string]int{} + require.False(t, olderContainersStopped([]localPod{old}, known, missing, "test-agent")) + now := time.Now() + old.deletionTimestamp = &now + known[old.uid] = true + require.False(t, olderContainersStopped([]localPod{old}, known, missing, "test-agent")) +} + +func TestComponentAddedByNewRevisionDoesNotWait(t *testing.T) { + old := siblingPod("old-pod-uid", "old-agent") + old.declaredContainers = []string{"agent"} + known := map[string]bool{old.uid: false} + require.True(t, olderContainersStopped([]localPod{old}, known, map[string]int{}, "system-probe")) +} + +func TestDifferentDaemonSetAndNewerPodDoNotBlock(t *testing.T) { + otherDaemon := siblingPod("other-daemon-pod", "other-daemon") + otherDaemon.owners[0].uid = "other-daemonset-uid" + newer := siblingPod("newer-pod", "newer-agent") + newer.createdAt = time.Unix(300, 0) + + older, err := olderSiblingPods([]localPod{selfPod(), otherDaemon, newer}, selfPodUID) + require.NoError(t, err) + require.Empty(t, older) +} + +func TestSameTimestampFailsClosed(t *testing.T) { + other := siblingPod("other-pod", "other-agent") + other.createdAt = selfPod().createdAt + _, err := olderSiblingPods([]localPod{selfPod(), other}, selfPodUID) + require.ErrorContains(t, err, "same-timestamp") +} + +func TestLifecycleRequiresValidConfiguration(t *testing.T) { + tests := map[string]struct { + component string + podUID string + platform string + }{ + "missing component": {podUID: selfPodUID, platform: "linux"}, + "unsafe component": {component: "../agent", podUID: selfPodUID, platform: "linux"}, + "missing Pod UID": {component: "test-agent", platform: "linux"}, + "non Linux": {component: "test-agent", podUID: selfPodUID, platform: "windows"}, + } + for name, test := range tests { + t.Run(name, func(t *testing.T) { + deps := dependencies{ + Config: config.NewMockWithOverrides(t, map[string]interface{}{ + rolloutEnabledKey: true, + rolloutPodUIDKey: test.podUID, + }), + Log: logmock.New(t), + Params: agentlifecycle.Params{ComponentName: test.component}, + } + _, err := newComponent(deps, &scriptedPodSource{}, test.platform) + require.Error(t, err) + }) + } +} + +func newEnabledComponent(t *testing.T, responses ...podResponse) (*component, *scriptedPodSource) { + t.Helper() + source := &scriptedPodSource{responses: responses, calls: make(chan struct{}, 20)} + comp, err := newComponent(enabledDependencies(t), source, "linux") + require.NoError(t, err) + result := comp.(*component) + result.pollInterval = time.Millisecond + return result, source +} + +func enabledDependencies(t *testing.T) dependencies { + return dependencies{ + Config: config.NewMockWithOverrides(t, map[string]interface{}{ + rolloutEnabledKey: true, + rolloutPodUIDKey: selfPodUID, + }), + Log: logmock.New(t), + Params: agentlifecycle.Params{ComponentName: "test-agent"}, + } +} + +func selfPod() localPod { + return localPod{ + uid: selfPodUID, + name: "new-agent", + namespace: "datadog-agent", + createdAt: time.Unix(200, 0), + owners: []podOwner{{kind: "DaemonSet", uid: daemonUID, controller: true}}, + containers: []localContainer{{name: "agent", ready: true}}, + } +} + +func siblingPod(uid, name string) localPod { + pod := selfPod() + pod.uid = uid + pod.name = name + pod.createdAt = time.Unix(100, 0) + return pod +} diff --git a/comp/core/agentlifecycle/impl/local_pods_kubelet.go b/comp/core/agentlifecycle/impl/local_pods_kubelet.go new file mode 100644 index 000000000000..7d67270d99d5 --- /dev/null +++ b/comp/core/agentlifecycle/impl/local_pods_kubelet.go @@ -0,0 +1,69 @@ +// 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 2026-present Datadog, Inc. + +//go:build kubelet + +package agentlifecycleimpl + +import ( + "context" + + "github.com/DataDog/datadog-agent/pkg/util/kubernetes/kubelet" +) + +type kubeletLocalPodSource struct{} + +func newLocalPodSource() localPodSource { + return kubeletLocalPodSource{} +} + +func (kubeletLocalPodSource) ListLocalPods(ctx context.Context) ([]localPod, error) { + kubeUtil, err := kubelet.GetKubeUtil() + if err != nil { + return nil, err + } + kubeletPods, err := kubeUtil.GetLocalPodList(ctx) + if err != nil { + return nil, err + } + + pods := make([]localPod, 0, len(kubeletPods)) + for _, pod := range kubeletPods { + if pod == nil { + continue + } + owners := make([]podOwner, 0, len(pod.Owners())) + for _, owner := range pod.Owners() { + owners = append(owners, podOwner{ + kind: owner.Kind, + uid: owner.ID, + controller: owner.Controller != nil && *owner.Controller, + }) + } + containers := make([]localContainer, 0, len(pod.Status.Containers)) + for _, container := range pod.Status.Containers { + containers = append(containers, localContainer{ + name: container.Name, + terminated: container.State.Terminated != nil, + ready: container.Ready, + }) + } + declaredContainers := make([]string, 0, len(pod.Spec.Containers)) + for _, container := range pod.Spec.Containers { + declaredContainers = append(declaredContainers, container.Name) + } + pods = append(pods, localPod{ + uid: pod.Metadata.UID, + name: pod.Metadata.Name, + namespace: pod.Metadata.Namespace, + createdAt: pod.Metadata.CreationTimestamp, + deletionTimestamp: pod.Metadata.DeletionTimestamp, + owners: owners, + declaredContainers: declaredContainers, + containers: containers, + }) + } + return pods, nil +} diff --git a/comp/core/agentlifecycle/impl/local_pods_nokubelet.go b/comp/core/agentlifecycle/impl/local_pods_nokubelet.go new file mode 100644 index 000000000000..011ccc9ebe15 --- /dev/null +++ b/comp/core/agentlifecycle/impl/local_pods_nokubelet.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 2026-present Datadog, Inc. + +//go:build !kubelet + +package agentlifecycleimpl + +import ( + "context" + "errors" +) + +type unsupportedLocalPodSource struct{} + +func newLocalPodSource() localPodSource { + return unsupportedLocalPodSource{} +} + +func (unsupportedLocalPodSource) ListLocalPods(context.Context) ([]localPod, error) { + return nil, errors.New("experimental node Agent rollout requires an Agent built with kubelet support") +} diff --git a/pkg/config/schema/yaml/core_schema.yaml b/pkg/config/schema/yaml/core_schema.yaml index e9aff7c87968..17e9ece304ac 100644 --- a/pkg/config/schema/yaml/core_schema.yaml +++ b/pkg/config/schema/yaml/core_schema.yaml @@ -8312,6 +8312,26 @@ properties: tags: - golang_type:duration comment: Duration during which the host tags will be submitted with metrics. + experimental: + node_type: section + type: object + properties: + node_agent_rollout: + node_type: section + type: object + properties: + enabled: + node_type: setting + type: boolean + default: false + pod_uid: + node_type: setting + type: string + default: '' + state_path: + node_type: setting + type: string + default: '' external_metrics: node_type: section type: object diff --git a/pkg/config/setup/BUILD.bazel b/pkg/config/setup/BUILD.bazel index 2b3253de04b4..713a266311d8 100644 --- a/pkg/config/setup/BUILD.bazel +++ b/pkg/config/setup/BUILD.bazel @@ -4,6 +4,7 @@ load("//bazel/rules/go:dd_agent_go_test.bzl", "dd_agent_go_test") go_library( name = "setup", srcs = [ + "agent_lifecycle.go", "apm_settings.go", "common_settings.go", "config.go", diff --git a/pkg/config/setup/agent_lifecycle.go b/pkg/config/setup/agent_lifecycle.go new file mode 100644 index 000000000000..8d9d40e6963f --- /dev/null +++ b/pkg/config/setup/agent_lifecycle.go @@ -0,0 +1,20 @@ +// 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 2026-present Datadog, Inc. + +package setup + +import pkgconfigmodel "github.com/DataDog/datadog-agent/pkg/config/model" + +const ( + // ExperimentalNodeAgentRolloutEnabled enables the experimental prepared/active Agent lifecycle. + ExperimentalNodeAgentRolloutEnabled = "experimental.node_agent_rollout.enabled" + // ExperimentalNodeAgentRolloutPodUID identifies the Pod that must wait for older siblings owned by the same DaemonSet. + ExperimentalNodeAgentRolloutPodUID = "experimental.node_agent_rollout.pod_uid" +) + +func setupExperimentalNodeAgentRollout(config pkgconfigmodel.Setup) { + config.BindEnvAndSetDefault(ExperimentalNodeAgentRolloutEnabled, false) + config.BindEnvAndSetDefault(ExperimentalNodeAgentRolloutPodUID, "") +} diff --git a/pkg/config/setup/config.go b/pkg/config/setup/config.go index d87d9fb9cf35..27985e40d0c3 100644 --- a/pkg/config/setup/config.go +++ b/pkg/config/setup/config.go @@ -282,6 +282,7 @@ func initCommonConfigComponents(config pkgconfigmodel.Setup) { // settings that are only initialized by the full agent, not serverless func initFullAgentOnlyComponents(config pkgconfigmodel.Setup) { comps := []func(pkgconfigmodel.Setup){ + setupExperimentalNodeAgentRollout, setupProcesses, setupPrivateActionRunner, remoteflags, diff --git a/pkg/config/setup/config_test.go b/pkg/config/setup/config_test.go index de8fd47209fb..0cbda83c7f5b 100644 --- a/pkg/config/setup/config_test.go +++ b/pkg/config/setup/config_test.go @@ -772,6 +772,13 @@ func TestHealthPlatformDefaults(t *testing.T) { assert.Equal(t, true, config.GetBool("health_platform.invalidconfig_check.enabled")) } +func TestExperimentalNodeAgentRolloutDefaults(t *testing.T) { + config := confFromYAML(t, "") + + assert.False(t, config.GetBool(ExperimentalNodeAgentRolloutEnabled)) + assert.Empty(t, config.GetString(ExperimentalNodeAgentRolloutPodUID)) +} + func TestInfrastructureModeNoneDisablesECSTaskCollection(t *testing.T) { datadogYaml := ` infrastructure_mode: none diff --git a/pkg/util/fxutil/BUILD.bazel b/pkg/util/fxutil/BUILD.bazel index 70c03b5dd8d9..d2f2690b28e0 100644 --- a/pkg/util/fxutil/BUILD.bazel +++ b/pkg/util/fxutil/BUILD.bazel @@ -14,6 +14,7 @@ go_library( "provide_comp.go", "provide_optional.go", "run.go", + "startup_gate.go", "test.go", "timeout.go", ], @@ -37,6 +38,7 @@ dd_agent_go_test( "errorunwrapper_test.go", "group_test.go", "provide_comp_test.go", + "startup_gate_test.go", "test_test.go", ], embed = [":fxutil"], diff --git a/pkg/util/fxutil/errorunwrapper.go b/pkg/util/fxutil/errorunwrapper.go index a64f84895d76..a7edefcf744d 100644 --- a/pkg/util/fxutil/errorunwrapper.go +++ b/pkg/util/fxutil/errorunwrapper.go @@ -8,17 +8,18 @@ package fxutil import ( "errors" "reflect" - "regexp" ) // UnwrapIfErrArgumentsFailed unwrap the error if the error was returned by an FX invoke method otherwise return the error. func UnwrapIfErrArgumentsFailed(err error) error { // This is a workaround until https://github.com/uber-go/fx/issues/988 will be done. if reflect.TypeOf(err).Name() == "errArgumentsFailed" { - re := regexp.MustCompile(`.*received non-nil error from function.*\(.*\): (.*)`) - matches := re.FindStringSubmatch(err.Error()) - if len(matches) == 2 { - return errors.New(matches[1]) + for { + cause := errors.Unwrap(err) + if cause == nil { + return err + } + err = cause } } return err diff --git a/pkg/util/fxutil/errorunwrapper_test.go b/pkg/util/fxutil/errorunwrapper_test.go index 65b8c1d487c2..62fe3a85794b 100644 --- a/pkg/util/fxutil/errorunwrapper_test.go +++ b/pkg/util/fxutil/errorunwrapper_test.go @@ -19,4 +19,5 @@ func TestUnwrapIfErrArgumentsFailed(t *testing.T) { fx.Provide(func() (*struct{}, error) { return nil, expectedError }), ) require.Equal(t, expectedError.Error(), err.Error()) + require.ErrorIs(t, err, expectedError) } diff --git a/pkg/util/fxutil/oneshot.go b/pkg/util/fxutil/oneshot.go index f970f13efd4d..68f4ecd92431 100644 --- a/pkg/util/fxutil/oneshot.go +++ b/pkg/util/fxutil/oneshot.go @@ -23,6 +23,22 @@ func OneShot(oneShotFunc interface{}, opts ...fx.Option) error { if fxAppTestOverride != nil { return fxAppTestOverride(oneShotFunc, opts) } + return oneShot(oneShotFunc, nil, opts...) +} + +// OneShotWithStartupGate constructs an Fx application, waits for the supplied gate before +// starting its lifecycle, invokes oneShotFunc, and releases the gate after lifecycle shutdown. +func OneShotWithStartupGate[T StartupGate](oneShotFunc interface{}, opts ...fx.Option) error { + if fxAppTestOverride != nil { + return fxAppTestOverride(oneShotFunc, opts) + } + + var gate StartupGate + opts = append([]fx.Option{StartupGateOption[T](context.Background(), &gate)}, opts...) + return oneShot(oneShotFunc, func() StartupGate { return gate }, opts...) +} + +func oneShot(oneShotFunc interface{}, gateProvider func() StartupGate, opts ...fx.Option) error { // Use a delayed Fx invocation to capture arguments for oneShotFunc during // application setup, but not actually invoke the question until all @@ -43,20 +59,39 @@ func OneShot(oneShotFunc interface{}, opts ...fx.Option) error { ) app := fx.New(opts...) + var gate StartupGate + if gateProvider != nil { + gate = gateProvider() + } + if err := app.Err(); err != nil { + if gate != nil { + return errors.Join(UnwrapIfErrArgumentsFailed(err), gate.Close()) + } + return UnwrapIfErrArgumentsFailed(err) + } + // start the app startCtx, cancel := context.WithTimeout(context.Background(), app.StartTimeout()) defer cancel() if err := app.Start(startCtx); err != nil { - return errors.Join(UnwrapIfErrArgumentsFailed(err), stopApp(app)) + return errors.Join(UnwrapIfErrArgumentsFailed(err), stopAppAndCloseGate(app, gate)) } // call the original oneShotFunc with the args captured during app startup err := delayedCall.call() if err != nil { - return errors.Join(err, stopApp(app)) + return errors.Join(err, stopAppAndCloseGate(app, gate)) } - return stopApp(app) + return stopAppAndCloseGate(app, gate) +} + +func stopAppAndCloseGate(app *fx.App, gate StartupGate) error { + stopErr := stopApp(app) + if gate != nil { + return errors.Join(stopErr, gate.Close()) + } + return stopErr } func stopApp(app *fx.App) error { diff --git a/pkg/util/fxutil/run.go b/pkg/util/fxutil/run.go index d094927ee7e2..011a77d44126 100644 --- a/pkg/util/fxutil/run.go +++ b/pkg/util/fxutil/run.go @@ -20,6 +20,22 @@ func Run(opts ...fx.Option) error { if fxAppTestOverride != nil { return fxAppTestOverride(func() {}, opts) } + return run(nil, opts...) +} + +// RunWithStartupGate constructs an Fx application, waits for the supplied gate before +// starting its lifecycle, and releases it after shutdown. +func RunWithStartupGate[T StartupGate](opts ...fx.Option) error { + if fxAppTestOverride != nil { + return fxAppTestOverride(func() {}, opts) + } + + var gate StartupGate + opts = append([]fx.Option{StartupGateOption[T](context.Background(), &gate)}, opts...) + return run(func() StartupGate { return gate }, opts...) +} + +func run(gateProvider func() StartupGate, opts ...fx.Option) error { opts = append(opts, FxAgentBase()) // Temporarily increase timeout for all fxutil.Run calls until we can better characterize our @@ -30,18 +46,24 @@ func Run(opts ...fx.Option) error { ) app := fx.New(opts...) + var gate StartupGate + if gateProvider != nil { + gate = gateProvider() + } if err := app.Err(); err != nil { - return err + if gate != nil { + return errors.Join(UnwrapIfErrArgumentsFailed(err), gate.Close()) + } + return UnwrapIfErrArgumentsFailed(err) } startCtx, cancel := context.WithTimeout(context.Background(), app.StartTimeout()) defer cancel() if err := app.Start(startCtx); err != nil { - return errors.Join(UnwrapIfErrArgumentsFailed(err), stopApp(app)) + return errors.Join(UnwrapIfErrArgumentsFailed(err), stopAppAndCloseGate(app, gate)) } - <-app.Done() - return stopApp(app) + return stopAppAndCloseGate(app, gate) } diff --git a/pkg/util/fxutil/startup_gate.go b/pkg/util/fxutil/startup_gate.go new file mode 100644 index 000000000000..ec6ceeba6284 --- /dev/null +++ b/pkg/util/fxutil/startup_gate.go @@ -0,0 +1,27 @@ +// 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 2026-present Datadog, Inc. + +package fxutil + +import ( + "context" + + "go.uber.org/fx" +) + +// StartupGate delays lifecycle startup until an external condition is satisfied. +type StartupGate interface { + Wait(context.Context) error + Close() error +} + +// StartupGateOption installs the gate as the first Fx invoke. Fx providers are +// lazy, so later constructors and invokes are not evaluated until Wait returns. +func StartupGateOption[T StartupGate](ctx context.Context, captured *StartupGate) fx.Option { + return fx.Invoke(func(gate T) error { + *captured = gate + return gate.Wait(ctx) + }) +} diff --git a/pkg/util/fxutil/startup_gate_test.go b/pkg/util/fxutil/startup_gate_test.go new file mode 100644 index 000000000000..4cc6ec3ca080 --- /dev/null +++ b/pkg/util/fxutil/startup_gate_test.go @@ -0,0 +1,118 @@ +// 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 2026-present Datadog, Inc. + +package fxutil + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/stretchr/testify/require" + "go.uber.org/fx" +) + +type gateRecorder struct { + mu sync.Mutex + events []string +} + +func (r *gateRecorder) add(event string) { + r.mu.Lock() + defer r.mu.Unlock() + r.events = append(r.events, event) +} + +func (r *gateRecorder) snapshot() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.events...) +} + +type recordingGate struct{ recorder *gateRecorder } + +func (g *recordingGate) Wait(context.Context) error { g.recorder.add("wait"); return nil } +func (g *recordingGate) Close() error { g.recorder.add("close"); return nil } + +type recordingComponent struct{} + +func TestOneShotWithStartupGateOrdering(t *testing.T) { + recorder := &gateRecorder{} + gate := &recordingGate{recorder: recorder} + + err := OneShotWithStartupGate[*recordingGate]( + func(_ *recordingComponent) { recorder.add("run") }, + fx.Supply(gate), + fx.Provide(func(lc fx.Lifecycle) *recordingComponent { + recorder.add("construct") + lc.Append(fx.Hook{ + OnStart: func(context.Context) error { recorder.add("start"); return nil }, + OnStop: func(context.Context) error { recorder.add("stop"); return nil }, + }) + return &recordingComponent{} + }), + ) + require.NoError(t, err) + require.Equal(t, []string{"wait", "construct", "start", "run", "stop", "close"}, recorder.snapshot()) +} + +func TestRunWithStartupGateOrdering(t *testing.T) { + recorder := &gateRecorder{} + gate := &recordingGate{recorder: recorder} + + err := RunWithStartupGate[*recordingGate]( + fx.Supply(gate), + fx.Invoke(func(lc fx.Lifecycle, shutdowner fx.Shutdowner) { + lc.Append(fx.Hook{ + OnStart: func(context.Context) error { + recorder.add("start") + return shutdowner.Shutdown() + }, + OnStop: func(context.Context) error { recorder.add("stop"); return nil }, + }) + }), + ) + require.NoError(t, err) + require.Equal(t, []string{"wait", "start", "stop", "close"}, recorder.snapshot()) +} + +func TestOneShotWithStartupGateClosesAfterStopFailure(t *testing.T) { + recorder := &gateRecorder{} + gate := &recordingGate{recorder: recorder} + stopErr := errors.New("stop failed") + + err := OneShotWithStartupGate[*recordingGate]( + func(_ *recordingComponent) { recorder.add("run") }, + fx.Supply(gate), + fx.Provide(func(lc fx.Lifecycle) *recordingComponent { + lc.Append(fx.Hook{ + OnStart: func(context.Context) error { recorder.add("start"); return nil }, + OnStop: func(context.Context) error { recorder.add("stop"); return stopErr }, + }) + return &recordingComponent{} + }), + ) + require.ErrorIs(t, err, stopErr) + require.Equal(t, []string{"wait", "start", "run", "stop", "close"}, recorder.snapshot()) +} + +func TestRunWithStartupGateClosesAfterStopFailure(t *testing.T) { + recorder := &gateRecorder{} + gate := &recordingGate{recorder: recorder} + stopErr := errors.New("stop failed") + + err := RunWithStartupGate[*recordingGate]( + fx.Supply(gate), + fx.Invoke(func(lc fx.Lifecycle, shutdowner fx.Shutdowner) { + lc.Append(fx.Hook{ + OnStart: func(context.Context) error { return shutdowner.Shutdown() }, + OnStop: func(context.Context) error { recorder.add("stop"); return stopErr }, + }) + }), + ) + require.ErrorIs(t, err, stopErr) + require.Equal(t, []string{"wait", "stop", "close"}, recorder.snapshot()) +} diff --git a/tasks/build_tags.bzl b/tasks/build_tags.bzl index 3731fae968e7..0076b3a38b96 100644 --- a/tasks/build_tags.bzl +++ b/tasks/build_tags.bzl @@ -236,6 +236,7 @@ SERVERLESS_TAGS = set(["serverless", "otlp"]) SYSTEM_PROBE_TAGS = set([ "datadog.no_waf", "ec2", + "kubelet", "linux_bpf", "netcgo", "npm", @@ -278,7 +279,7 @@ LOADER_TAGS = set() # imported by https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/f963ab53ee55aeb56d58617ed12c840e8b07cc53/receiver/prometheusreceiver/factory.go#L10 HOST_PROFILER_TAGS = set(["remove_all_sd", "docker", "kubelet"]) -PRIVATEACTIONRUNNER_TAGS = set(["zlib", "zstd"]) +PRIVATEACTIONRUNNER_TAGS = set(["kubelet", "zlib", "zstd"]) SECRET_GENERIC_CONNECTOR_TAGS = set()