diff --git a/go.mod b/go.mod index 97d5c5d07d..20a049490c 100644 --- a/go.mod +++ b/go.mod @@ -29,6 +29,7 @@ require ( k8s.io/apimachinery v0.35.3 k8s.io/cli-runtime v0.35.3 k8s.io/client-go v0.35.3 + k8s.io/component-helpers v0.35.3 // indirect k8s.io/klog/v2 v2.140.0 k8s.io/kube-aggregator v0.35.3 k8s.io/kube-openapi v0.0.0-20251125145642-4e65d59e963e @@ -291,7 +292,6 @@ require ( k8s.io/apiserver v0.35.3 // indirect k8s.io/cloud-provider v0.35.0 // indirect k8s.io/component-base v0.35.3 // indirect - k8s.io/component-helpers v0.35.3 // indirect k8s.io/csi-translation-lib v0.35.0 // indirect k8s.io/gengo/v2 v2.0.0-20250922181213-ec3ebc5fd46b // indirect modernc.org/sqlite v1.34.1 // indirect diff --git a/internal/controller/datadogagentinternal/controller_reconcile_agent.go b/internal/controller/datadogagentinternal/controller_reconcile_agent.go index cb9f07eea2..786e782d4e 100644 --- a/internal/controller/datadogagentinternal/controller_reconcile_agent.go +++ b/internal/controller/datadogagentinternal/controller_reconcile_agent.go @@ -12,9 +12,11 @@ import ( edsv1alpha1 "github.com/DataDog/extendeddaemonset/api/v1alpha1" appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/reconcile" datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1" @@ -289,7 +291,48 @@ func (r *Reconciler) reconcileV2Agent(ctx context.Context, requiredComponents fe return reconcile.Result{}, nil } - return r.createOrUpdateDaemonset(ctx, ddai, daemonset, newStatus, updateDSStatusV2WithAgent) + rolloutBudget := preparedRolloutBudget(ddai, &r.options.ExtendedDaemonsetOptions) + rolloutEnabled := preparedRolloutEnabled(ddai) + var currentDaemonSet *appsv1.DaemonSet + if rolloutEnabled { + reader := r.apiReader + if reader == nil { + reader = r.client + } + currentDaemonSet = &appsv1.DaemonSet{} + if getErr := reader.Get(ctx, client.ObjectKeyFromObject(daemonset), currentDaemonSet); getErr != nil { + if !errors.IsNotFound(getErr) { + return reconcile.Result{}, getErr + } + currentDaemonSet = nil + } + } + affinityMigration, prepareErr := configurePreparedRollout(ddai, daemonset, currentDaemonSet, rolloutBudget) + if prepareErr != nil { + objLogger.Error(prepareErr, "Prepared Agent rollout request is incompatible with the rendered Pod template") + if r.recorder != nil { + r.recorder.Eventf(ddai, corev1.EventTypeWarning, "AgentPreparedRolloutRejected", "Prepared Agent rollout is disabled for this template: %v", prepareErr) + } + return reconcile.Result{}, prepareErr + } + result, err := r.createOrUpdateDaemonset(ctx, ddai, daemonset, newStatus, updateDSStatusV2WithAgent) + if err != nil || !rolloutEnabled { + return result, err + } + if affinityMigration { + if result.RequeueAfter == 0 || result.RequeueAfter > time.Second { + result.RequeueAfter = time.Second + } + return result, nil + } + fallbackResult, fallbackErr := r.reconcilePreparedRollout(ctx, ddai, daemonset, rolloutBudget) + if fallbackErr != nil { + return reconcile.Result{}, fallbackErr + } + if fallbackResult.RequeueAfter > 0 && (result.RequeueAfter == 0 || fallbackResult.RequeueAfter < result.RequeueAfter) { + result.RequeueAfter = fallbackResult.RequeueAfter + } + return result, nil } func updateDSStatusV2WithAgent(dsName string, ds *appsv1.DaemonSet, newStatus *datadoghqv1alpha1.DatadogAgentInternalStatus, updateTime metav1.Time, status metav1.ConditionStatus, reason, message string) { diff --git a/internal/controller/datadogagentinternal/controller_reconcile_agent_test.go b/internal/controller/datadogagentinternal/controller_reconcile_agent_test.go index d742818953..b3946e2f2d 100644 --- a/internal/controller/datadogagentinternal/controller_reconcile_agent_test.go +++ b/internal/controller/datadogagentinternal/controller_reconcile_agent_test.go @@ -1,23 +1,105 @@ package datadogagentinternal import ( + "context" "testing" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/tools/record" "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + apicommon "github.com/DataDog/datadog-operator/api/datadoghq/common" datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1" datadoghqv2alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v2alpha1" "github.com/DataDog/datadog-operator/internal/controller/datadogagent/component" + "github.com/DataDog/datadog-operator/internal/controller/datadogagent/defaults" + "github.com/DataDog/datadog-operator/internal/controller/datadogagent/feature" + "github.com/DataDog/datadog-operator/internal/controller/datadogagent/store" "github.com/DataDog/datadog-operator/pkg/constants" "github.com/DataDog/datadog-operator/pkg/kubernetes" + pkgtestutils "github.com/DataDog/datadog-operator/pkg/testutils" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) const defaultProvider = kubernetes.DefaultProvider const gkeCosProvider = kubernetes.GKECloudProvider + "-" + kubernetes.GKECosType +func TestReconcileV2AgentCreatesPreparedSurgeDaemonSet(t *testing.T) { + r, ddai := newPreparedRolloutReconciler(t, true) + status := &datadoghqv1alpha1.DatadogAgentInternalStatus{} + + result, err := r.reconcileV2Agent( + context.Background(), + preparedRolloutRequiredComponents(), + nil, + ddai, + feature.NewResourceManagers(store.NewStore(ddai, nil)), + status, + defaultProvider, + ) + + require.NoError(t, err) + assert.Equal(t, resourceFallbackPollInterval, result.RequeueAfter) + daemonSets := &appsv1.DaemonSetList{} + require.NoError(t, r.client.List(context.Background(), daemonSets)) + require.Len(t, daemonSets.Items, 1) + ds := &daemonSets.Items[0] + assert.Equal(t, preparedRolloutModeV1, ds.Spec.Template.Annotations[preparedRolloutModeAnnotation]) + require.NotNil(t, ds.Spec.UpdateStrategy.RollingUpdate) + assert.Equal(t, intstr.FromInt(1), *ds.Spec.UpdateStrategy.RollingUpdate.MaxSurge) + assert.Equal(t, intstr.FromInt(0), *ds.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable) +} + +func newPreparedRolloutReconciler(t *testing.T, hostNetwork bool) (*Reconciler, *datadoghqv1alpha1.DatadogAgentInternal) { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, appsv1.AddToScheme(scheme)) + require.NoError(t, datadoghqv1alpha1.AddToScheme(scheme)) + c := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(&appsv1.DaemonSet{}).Build() + one := intstr.FromInt(1) + ddai := pkgtestutils.NewDatadogAgentInternal("datadog-agent", "agent", nil) + ddai.UID = "ddai-uid" + ddai.Annotations = map[string]string{preparedRolloutModeAnnotation: preparedRolloutModeV1} + ddai.Spec.Features = &datadoghqv2alpha1.DatadogFeatures{} + ddai.Spec.Override = map[datadoghqv2alpha1.ComponentName]*datadoghqv2alpha1.DatadogAgentComponentOverride{ + datadoghqv2alpha1.NodeAgentComponentName: { + HostNetwork: ptr.To(hostNetwork), + UpdateStrategy: &apicommon.UpdateStrategy{ + Type: string(appsv1.RollingUpdateDaemonSetStrategyType), + RollingUpdate: &apicommon.RollingUpdate{ + MaxSurge: ptr.To(one), + MaxUnavailable: ptr.To(one), + }, + }, + }, + } + defaults.DefaultDatadogAgentSpec(&ddai.Spec) + return &Reconciler{ + client: c, + apiReader: c, + scheme: scheme, + recorder: record.NewFakeRecorder(10), + }, ddai +} + +func preparedRolloutRequiredComponents() feature.RequiredComponents { + return feature.RequiredComponents{Agent: feature.RequiredComponent{ + IsRequired: ptr.To(true), + Containers: []apicommon.AgentContainerName{ + apicommon.CoreAgentContainerName, + apicommon.TraceAgentContainerName, + }, + }} +} + // func Test_getValidDaemonSetNames(t *testing.T) { // testCases := []struct { // name string diff --git a/internal/controller/datadogagentinternal/prepared_rollout.go b/internal/controller/datadogagentinternal/prepared_rollout.go new file mode 100644 index 0000000000..a98f1109d1 --- /dev/null +++ b/internal/controller/datadogagentinternal/prepared_rollout.go @@ -0,0 +1,245 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// Copyright 2016-present Datadog, Inc. + +package datadogagentinternal + +import ( + "fmt" + "slices" + "strings" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" + "k8s.io/apimachinery/pkg/util/intstr" + + apicommon "github.com/DataDog/datadog-operator/api/datadoghq/common" + datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1" +) + +const ( + preparedRolloutModeAnnotation = "experimental.agent.datadoghq.com/node-agent-rollout-mode" + preparedRolloutModeV1 = "prepared-surge-v1" + + rolloutEnabledEnv = "DD_EXPERIMENTAL_NODE_AGENT_ROLLOUT_ENABLED" + rolloutPodUIDEnv = "DD_EXPERIMENTAL_NODE_AGENT_ROLLOUT_POD_UID" + kubeletHostEnv = "DD_KUBERNETES_KUBELET_HOST" +) + +var preparedRolloutContainerNames = map[string]struct{}{ + string(apicommon.CoreAgentContainerName): {}, + string(apicommon.TraceAgentContainerName): {}, + string(apicommon.ProcessAgentContainerName): {}, + string(apicommon.SystemProbeContainerName): {}, + string(apicommon.HostProfiler): {}, + string(apicommon.OtelAgent): {}, + string(apicommon.PrivateActionRunnerContainerName): {}, +} + +var preparedRolloutInitContainerNames = map[string]struct{}{ + string(apicommon.InitVolumeContainerName): {}, + string(apicommon.InitConfigContainerName): {}, + "seccomp-setup": {}, + "host-profiler-seccomp-setup": {}, +} + +func preparedRolloutEnabled(ddai *datadoghqv1alpha1.DatadogAgentInternal) bool { + return ddai != nil && ddai.Annotations[preparedRolloutModeAnnotation] == preparedRolloutModeV1 +} + +// configurePreparedRollout enables native DaemonSet surge. Existing profile +// DaemonSets need one ordinary rollout to narrow their anti-affinity before two +// revisions can share a node. +func configurePreparedRollout(ddai *datadoghqv1alpha1.DatadogAgentInternal, ds, current *appsv1.DaemonSet, budget intstr.IntOrString) (bool, error) { + if !preparedRolloutEnabled(ddai) { + return false, nil + } + if !positiveIntOrPercent(&budget) { + return false, fmt.Errorf("prepared Agent rollout requires a positive, valid maxUnavailable budget") + } + if ds.Spec.UpdateStrategy.Type != "" && ds.Spec.UpdateStrategy.Type != appsv1.RollingUpdateDaemonSetStrategyType { + return false, fmt.Errorf("prepared Agent rollout requires RollingUpdate strategy") + } + prepared := ds.DeepCopy() + if err := prepareAgentTemplate(prepared); err != nil { + return false, err + } + if !configurePreparedSurge(prepared, budget) { + return false, fmt.Errorf("prepared Agent rollout requires a positive, valid maxUnavailable budget") + } + + if current != nil && profileAffinityMigrationPending(current) { + migrationTemplate := current.Spec.Template.DeepCopy() + if apiequality.Semantic.DeepEqual(migrationTemplate.Spec.Affinity.PodAntiAffinity, broadAgentPodAntiAffinity()) && !prepareProfileAntiAffinityForSurge(migrationTemplate) { + return false, fmt.Errorf("prepared Agent rollout cannot migrate profile anti-affinity") + } + ds.Spec.Template = *migrationTemplate + configureConventionalMigration(ds, budget) + return true, nil + } + + ds.Spec.Template = prepared.Spec.Template + ds.Spec.UpdateStrategy = prepared.Spec.UpdateStrategy + return false, nil +} + +func profileAffinityMigrationPending(current *appsv1.DaemonSet) bool { + antiAffinity := current.Spec.Template.Spec.Affinity + if antiAffinity == nil || antiAffinity.PodAntiAffinity == nil { + return false + } + if apiequality.Semantic.DeepEqual(antiAffinity.PodAntiAffinity, broadAgentPodAntiAffinity()) { + return true + } + expected, ok := profileSurgePodAntiAffinity(current.Spec.Template.Labels) + return ok && apiequality.Semantic.DeepEqual(antiAffinity.PodAntiAffinity, expected) && + !hasRolloutMode(current.Spec.Template.Annotations) && !daemonSetFullyRolledOut(current) +} + +func daemonSetFullyRolledOut(ds *appsv1.DaemonSet) bool { + desired := ds.Status.DesiredNumberScheduled + return desired > 0 && ds.Status.ObservedGeneration == ds.Generation && + ds.Status.UpdatedNumberScheduled == desired && ds.Status.NumberAvailable == desired && ds.Status.NumberUnavailable == 0 +} + +func prepareAgentTemplate(ds *appsv1.DaemonSet) error { + spec := &ds.Spec.Template.Spec + if spec.OS != nil && spec.OS.Name != corev1.Linux || spec.NodeSelector[corev1.LabelOSStable] == "windows" || spec.NodeSelector["beta.kubernetes.io/os"] == "windows" { + return fmt.Errorf("prepared Agent rollout is Linux-only") + } + if err := validatePreparedContainers(spec); err != nil { + return err + } + if !prepareProfileAntiAffinityForSurge(&ds.Spec.Template) { + return fmt.Errorf("prepared Agent rollout does not support custom Pod anti-affinity") + } + if !spec.HostNetwork && podUsesHostPorts(spec) { + return fmt.Errorf("prepared Agent rollout cannot overlap Pod-networked containers that declare hostPort") + } + for i := range spec.Containers { + container := &spec.Containers[i] + if container.Name == string(apicommon.TraceAgentContainerName) { + traceIndex := slices.Index(container.Command, "trace-agent") + if traceIndex < 0 { + return fmt.Errorf("prepared Agent rollout cannot bypass an unknown trace-agent loader command") + } + container.Command = append([]string(nil), container.Command[traceIndex:]...) + } + configurePreparedContainer(container) + if spec.HostNetwork { + // On hostNetwork, declared container ports are scheduler host-port + // claims. The process still binds the same node ports after its older + // peer stops. Pod-networked containers keep their port metadata. + container.Ports = nil + } + } + if spec.HostNetwork { + for i := range spec.InitContainers { + spec.InitContainers[i].Ports = nil + } + } + if ds.Spec.Template.Annotations == nil { + ds.Spec.Template.Annotations = map[string]string{} + } + ds.Spec.Template.Annotations[preparedRolloutModeAnnotation] = preparedRolloutModeV1 + return nil +} + +func podUsesHostPorts(spec *corev1.PodSpec) bool { + for i := range spec.InitContainers { + for _, port := range spec.InitContainers[i].Ports { + if port.HostPort != 0 { + return true + } + } + } + for i := range spec.Containers { + for _, port := range spec.Containers[i].Ports { + if port.HostPort != 0 { + return true + } + } + } + return false +} + +func validatePreparedContainers(spec *corev1.PodSpec) error { + for i := range spec.Containers { + container := &spec.Containers[i] + if _, ok := preparedRolloutContainerNames[container.Name]; !ok { + return fmt.Errorf("prepared Agent rollout does not support container %q", container.Name) + } + if container.Lifecycle != nil { + return fmt.Errorf("prepared Agent rollout does not support lifecycle hooks on container %q", container.Name) + } + if !preparedContainerCommandSupported(container) { + return fmt.Errorf("prepared Agent rollout does not support command %q on container %q", container.Command, container.Name) + } + } + for i := range spec.InitContainers { + container := &spec.InitContainers[i] + if _, ok := preparedRolloutInitContainerNames[container.Name]; !ok { + return fmt.Errorf("prepared Agent rollout does not support init container %q", container.Name) + } + if container.Lifecycle != nil { + return fmt.Errorf("prepared Agent rollout does not support lifecycle hooks on init container %q", container.Name) + } + } + return nil +} + +func preparedContainerCommandSupported(container *corev1.Container) bool { + if len(container.Command) == 0 { + return false + } + expected := map[string]string{ + string(apicommon.CoreAgentContainerName): "agent", + string(apicommon.TraceAgentContainerName): "trace-agent", + string(apicommon.ProcessAgentContainerName): "process-agent", + string(apicommon.SystemProbeContainerName): "system-probe", + string(apicommon.HostProfiler): "host-profiler", + string(apicommon.OtelAgent): "otel-agent", + string(apicommon.PrivateActionRunnerContainerName): "/opt/datadog-agent/embedded/bin/privateactionrunner", + } + want := expected[container.Name] + if container.Name == string(apicommon.TraceAgentContainerName) { + return slices.Contains(container.Command, want) + } + if container.Command[0] != want { + return false + } + return true +} + +func configurePreparedContainer(container *corev1.Container) { + setContainerEnv(container, corev1.EnvVar{Name: rolloutEnabledEnv, Value: "true"}) + setContainerEnv(container, corev1.EnvVar{ + Name: rolloutPodUIDEnv, + ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{ + APIVersion: "v1", + FieldPath: "metadata.uid", + }}, + }) + setContainerEnv(container, corev1.EnvVar{ + Name: kubeletHostEnv, + ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{ + APIVersion: "v1", + FieldPath: "status.hostIP", + }}, + }) +} + +func setContainerEnv(container *corev1.Container, env corev1.EnvVar) { + for i := range container.Env { + if container.Env[i].Name == env.Name { + container.Env[i] = env + return + } + } + container.Env = append(container.Env, env) +} + +func hasRolloutMode(annotations map[string]string) bool { + return strings.EqualFold(annotations[preparedRolloutModeAnnotation], preparedRolloutModeV1) +} diff --git a/internal/controller/datadogagentinternal/prepared_rollout_support.go b/internal/controller/datadogagentinternal/prepared_rollout_support.go new file mode 100644 index 0000000000..71c27aa03b --- /dev/null +++ b/internal/controller/datadogagentinternal/prepared_rollout_support.go @@ -0,0 +1,230 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// Copyright 2016-present Datadog, Inc. + +package datadogagentinternal + +import ( + "context" + "encoding/json" + "fmt" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apiequality "k8s.io/apimachinery/pkg/api/equality" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/intstr" + "sigs.k8s.io/controller-runtime/pkg/client" + + datadoghqcommon "github.com/DataDog/datadog-operator/api/datadoghq/common" + datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1" + datadoghqv2alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v2alpha1" + componentagent "github.com/DataDog/datadog-operator/internal/controller/datadogagent/component/agent" + "github.com/DataDog/datadog-operator/pkg/constants" +) + +const ( + defaultPreparedRolloutMaxUnavailable = 1 +) + +func configurePreparedSurge(ds *appsv1.DaemonSet, budget intstr.IntOrString) bool { + strategy := &ds.Spec.UpdateStrategy + if strategy.Type != "" && strategy.Type != appsv1.RollingUpdateDaemonSetStrategyType { + return false + } + if strategy.RollingUpdate == nil { + strategy.RollingUpdate = &appsv1.RollingUpdateDaemonSet{} + } + if _, err := intstr.GetScaledValueFromIntOrPercent(&budget, 100, true); err != nil { + return false + } + + strategy.Type = appsv1.RollingUpdateDaemonSetStrategyType + zero := intstr.FromInt(0) + strategy.RollingUpdate.MaxUnavailable = &zero + surge := budget + strategy.RollingUpdate.MaxSurge = &surge + return positiveIntOrPercent(&surge) +} + +func configureConventionalMigration(ds *appsv1.DaemonSet, budget intstr.IntOrString) { + zero := intstr.FromInt(0) + value := budget + ds.Spec.UpdateStrategy = appsv1.DaemonSetUpdateStrategy{ + Type: appsv1.RollingUpdateDaemonSetStrategyType, + RollingUpdate: &appsv1.RollingUpdateDaemonSet{ + MaxUnavailable: &value, + MaxSurge: &zero, + }, + } +} + +// prepareProfileAntiAffinityForSurge narrows the standard DAP anti-affinity so +// old and new revisions of the same profile may overlap. Unknown user-supplied +// anti-affinity fails closed. +func prepareProfileAntiAffinityForSurge(template *corev1.PodTemplateSpec) bool { + if template.Spec.Affinity == nil || template.Spec.Affinity.PodAntiAffinity == nil { + return true + } + if !apiequality.Semantic.DeepEqual(template.Spec.Affinity.PodAntiAffinity, broadAgentPodAntiAffinity()) { + return false + } + narrowed, ok := profileSurgePodAntiAffinity(template.Labels) + if !ok { + return false + } + template.Spec.Affinity.PodAntiAffinity = narrowed + return true +} + +func broadAgentPodAntiAffinity() *corev1.PodAntiAffinity { + return &corev1.PodAntiAffinity{RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{{ + LabelSelector: &metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{{ + Key: datadoghqcommon.AgentDeploymentComponentLabelKey, Operator: metav1.LabelSelectorOpIn, Values: []string{constants.DefaultAgentResourceSuffix}, + }}}, + TopologyKey: corev1.LabelHostname, + }}} +} + +func profileSurgePodAntiAffinity(podLabels map[string]string) (*corev1.PodAntiAffinity, bool) { + ddaName := podLabels[datadoghqcommon.AgentDeploymentNameLabelKey] + if ddaName == "" { + return nil, false + } + + profileRequirement := metav1.LabelSelectorRequirement{Key: constants.ProfileLabelKey} + if profileName := podLabels[constants.ProfileLabelKey]; profileName != "" { + profileRequirement.Operator = metav1.LabelSelectorOpNotIn + profileRequirement.Values = []string{profileName} + } else { + profileRequirement.Operator = metav1.LabelSelectorOpExists + } + componentRequirement := metav1.LabelSelectorRequirement{ + Key: datadoghqcommon.AgentDeploymentComponentLabelKey, Operator: metav1.LabelSelectorOpIn, Values: []string{constants.DefaultAgentResourceSuffix}, + } + + return &corev1.PodAntiAffinity{RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{ + { + LabelSelector: &metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{ + componentRequirement, + {Key: datadoghqcommon.AgentDeploymentNameLabelKey, Operator: metav1.LabelSelectorOpIn, Values: []string{ddaName}}, + profileRequirement, + }}, + TopologyKey: corev1.LabelHostname, + }, + { + LabelSelector: &metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{ + componentRequirement, + {Key: datadoghqcommon.AgentDeploymentNameLabelKey, Operator: metav1.LabelSelectorOpNotIn, Values: []string{ddaName}}, + }}, + TopologyKey: corev1.LabelHostname, + }, + }}, true +} + +func positiveIntOrPercent(value *intstr.IntOrString) bool { + if value == nil { + return false + } + scaled, err := intstr.GetScaledValueFromIntOrPercent(value, 100, true) + return err == nil && scaled > 0 +} + +func preparedRolloutBudget(ddai *datadoghqv1alpha1.DatadogAgentInternal, options *componentagent.ExtendedDaemonsetOptions) intstr.IntOrString { + if override, ok := ddai.Spec.Override[datadoghqv2alpha1.NodeAgentComponentName]; ok && override != nil && override.UpdateStrategy != nil && override.UpdateStrategy.RollingUpdate != nil && override.UpdateStrategy.RollingUpdate.MaxUnavailable != nil { + return *override.UpdateStrategy.RollingUpdate.MaxUnavailable + } + if options != nil && options.MaxPodUnavailable != "" { + return intstr.Parse(options.MaxPodUnavailable) + } + return intstr.FromInt(defaultPreparedRolloutMaxUnavailable) +} + +func daemonSetControlledByDDAI(ds *appsv1.DaemonSet, ddai *datadoghqv1alpha1.DatadogAgentInternal) bool { + owner := metav1.GetControllerOf(ds) + return owner != nil && owner.APIVersion == datadoghqv1alpha1.GroupVersion.String() && owner.Kind == "DatadogAgentInternal" && owner.UID == ddai.UID +} + +func preparedRolloutDaemonSetEligible(ds *appsv1.DaemonSet) bool { + if ds.DeletionTimestamp != nil { + return false + } + return ds.Spec.UpdateStrategy.Type == appsv1.RollingUpdateDaemonSetStrategyType && + ds.Spec.UpdateStrategy.RollingUpdate != nil && positiveIntOrPercent(ds.Spec.UpdateStrategy.RollingUpdate.MaxSurge) +} + +func currentDaemonSetRevision(ctx context.Context, reader client.Reader, ds *appsv1.DaemonSet) (string, error) { + revisions := &appsv1.ControllerRevisionList{} + if err := reader.List(ctx, revisions, client.InNamespace(ds.Namespace)); err != nil { + return "", fmt.Errorf("list revisions for Agent DaemonSet %s/%s: %w", ds.Namespace, ds.Name, err) + } + + var current *appsv1.ControllerRevision + for i := range revisions.Items { + revision := &revisions.Items[i] + if !controlledByUID(revision, ds.UID) { + continue + } + matches, err := controllerRevisionMatchesTemplate(revision, &ds.Spec.Template) + if err != nil { + return "", fmt.Errorf("decode revision %s for Agent DaemonSet %s/%s: %w", revision.Name, ds.Namespace, ds.Name, err) + } + if matches && (current == nil || revision.Revision > current.Revision) { + current = revision + } + } + if current == nil { + return "", nil + } + return current.Labels[appsv1.DefaultDaemonSetUniqueLabelKey], nil +} + +func controllerRevisionMatchesTemplate(revision *appsv1.ControllerRevision, template *corev1.PodTemplateSpec) (bool, error) { + var patch struct { + Spec struct { + Template corev1.PodTemplateSpec `json:"template"` + } `json:"spec"` + } + if err := json.Unmarshal(revision.Data.Raw, &patch); err != nil { + return false, err + } + return apiequality.Semantic.DeepEqual(patch.Spec.Template, *template), nil +} + +func daemonSetPods(ctx context.Context, reader client.Reader, ds *appsv1.DaemonSet) ([]corev1.Pod, error) { + selector, err := metav1.LabelSelectorAsSelector(ds.Spec.Selector) + if err != nil { + return nil, fmt.Errorf("build selector for Agent DaemonSet %s/%s: %w", ds.Namespace, ds.Name, err) + } + list := &corev1.PodList{} + if err := reader.List(ctx, list, client.InNamespace(ds.Namespace), client.MatchingLabelsSelector{Selector: selector}); err != nil { + return nil, fmt.Errorf("list Pods for Agent DaemonSet %s/%s: %w", ds.Namespace, ds.Name, err) + } + result := make([]corev1.Pod, 0, len(list.Items)) + for i := range list.Items { + if controlledByUID(&list.Items[i], ds.UID) { + result = append(result, list.Items[i]) + } + } + return result, nil +} + +func controlledByUID(obj metav1.Object, uid types.UID) bool { + owner := metav1.GetControllerOf(obj) + return owner != nil && owner.UID == uid +} + +func podAvailable(pod *corev1.Pod, minReadySeconds int32, now time.Time) bool { + if pod.DeletionTimestamp != nil || pod.Status.Phase != corev1.PodRunning { + return false + } + for i := range pod.Status.Conditions { + condition := &pod.Status.Conditions[i] + if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue { + return minReadySeconds == 0 || !condition.LastTransitionTime.IsZero() && condition.LastTransitionTime.Add(time.Duration(minReadySeconds)*time.Second).Before(now) + } + } + return false +} diff --git a/internal/controller/datadogagentinternal/prepared_rollout_test.go b/internal/controller/datadogagentinternal/prepared_rollout_test.go new file mode 100644 index 0000000000..509dc404eb --- /dev/null +++ b/internal/controller/datadogagentinternal/prepared_rollout_test.go @@ -0,0 +1,235 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// Copyright 2016-present Datadog, Inc. + +package datadogagentinternal + +import ( + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" + + apicommon "github.com/DataDog/datadog-operator/api/datadoghq/common" + datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1" + "github.com/DataDog/datadog-operator/pkg/constants" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPreparedRolloutRequiresExplicitMode(t *testing.T) { + ddai := &datadoghqv1alpha1.DatadogAgentInternal{ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{}}} + assert.False(t, preparedRolloutEnabled(ddai)) + ddai.Annotations[preparedRolloutModeAnnotation] = "true" + assert.False(t, preparedRolloutEnabled(ddai)) + ddai.Annotations[preparedRolloutModeAnnotation] = preparedRolloutModeV1 + assert.True(t, preparedRolloutEnabled(ddai)) +} + +func TestPrepareAgentTemplatePreservesHostNetworkAndUDS(t *testing.T) { + ds := preparedTestDaemonSet(true) + require.NoError(t, prepareAgentTemplate(ds)) + assert.True(t, ds.Spec.Template.Spec.HostNetwork) + for i := range ds.Spec.Template.Spec.Containers { + assert.Empty(t, ds.Spec.Template.Spec.Containers[i].Ports) + } + assert.True(t, hasHostPath(ds.Spec.Template.Spec.Volumes, "/var/run/datadog")) + + podNetwork := preparedTestDaemonSet(false) + for i := range podNetwork.Spec.Template.Spec.Containers { + for j := range podNetwork.Spec.Template.Spec.Containers[i].Ports { + podNetwork.Spec.Template.Spec.Containers[i].Ports[j].HostPort = 0 + } + } + require.NoError(t, prepareAgentTemplate(podNetwork)) + assert.False(t, podNetwork.Spec.Template.Spec.HostNetwork) + for i := range podNetwork.Spec.Template.Spec.Containers { + assert.NotEmpty(t, podNetwork.Spec.Template.Spec.Containers[i].Ports) + } + + podNetworkWithHostPort := preparedTestDaemonSet(false) + require.ErrorContains(t, prepareAgentTemplate(podNetworkWithHostPort), "declare hostPort") +} + +func TestConfigurePreparedRolloutUsesExistingUnavailableBudgetAsSurge(t *testing.T) { + ddai := preparedRolloutDDAI() + ds := preparedTestDaemonSet(true) + budget := intstr.FromString("10%") + migrating, err := configurePreparedRollout(ddai, ds, nil, budget) + require.NoError(t, err) + assert.False(t, migrating) + assert.Equal(t, budget, *ds.Spec.UpdateStrategy.RollingUpdate.MaxSurge) + assert.Equal(t, intstr.FromInt(0), *ds.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable) + assert.Equal(t, preparedRolloutModeV1, ds.Spec.Template.Annotations[preparedRolloutModeAnnotation]) +} + +func TestPreparedRolloutMigratesExistingProfileAntiAffinityBeforeSurge(t *testing.T) { + budget := intstr.FromInt(1) + current := preparedTestDaemonSet(true) + current.Generation = 1 + current.Spec.Template.Spec.Affinity = &corev1.Affinity{PodAntiAffinity: broadAgentPodAntiAffinity()} + + desired := preparedTestDaemonSet(true) + migrating, err := configurePreparedRollout(preparedRolloutDDAI(), desired, current, budget) + require.NoError(t, err) + require.True(t, migrating) + assert.Equal(t, intstr.FromInt(0), *desired.Spec.UpdateStrategy.RollingUpdate.MaxSurge) + assert.Equal(t, budget, *desired.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable) + assert.Nil(t, containerEnv(&desired.Spec.Template.Spec.Containers[0], rolloutEnabledEnv)) + + current = desired.DeepCopy() + current.Generation = 2 + current.Status = appsv1.DaemonSetStatus{ObservedGeneration: 2, DesiredNumberScheduled: 2, UpdatedNumberScheduled: 2, NumberAvailable: 2} + desired = preparedTestDaemonSet(true) + migrating, err = configurePreparedRollout(preparedRolloutDDAI(), desired, current, budget) + require.NoError(t, err) + assert.False(t, migrating) + assert.Equal(t, budget, *desired.Spec.UpdateStrategy.RollingUpdate.MaxSurge) +} + +func TestPreparedRolloutPreservesProbes(t *testing.T) { + ds := preparedTestDaemonSet(true) + original := make(map[string]struct { + startup *corev1.Probe + liveness *corev1.Probe + readiness *corev1.Probe + }, len(ds.Spec.Template.Spec.Containers)) + for i := range ds.Spec.Template.Spec.Containers { + container := &ds.Spec.Template.Spec.Containers[i] + original[container.Name] = struct { + startup *corev1.Probe + liveness *corev1.Probe + readiness *corev1.Probe + }{container.StartupProbe.DeepCopy(), container.LivenessProbe.DeepCopy(), container.ReadinessProbe.DeepCopy()} + } + require.NoError(t, prepareAgentTemplate(ds)) + for i := range ds.Spec.Template.Spec.Containers { + container := &ds.Spec.Template.Spec.Containers[i] + assert.Equal(t, original[container.Name].startup, container.StartupProbe) + assert.Equal(t, original[container.Name].liveness, container.LivenessProbe) + assert.Equal(t, original[container.Name].readiness, container.ReadinessProbe) + assert.Equal(t, "metadata.uid", containerEnv(container, rolloutPodUIDEnv).ValueFrom.FieldRef.FieldPath) + assert.Equal(t, "status.hostIP", containerEnv(container, kubeletHostEnv).ValueFrom.FieldRef.FieldPath) + } +} + +func TestPreparedRolloutSupportsAllRenderedAgentContainers(t *testing.T) { + ds := preparedTestDaemonSet(true) + ds.Spec.Template.Spec.Containers = []corev1.Container{ + {Name: string(apicommon.CoreAgentContainerName), Command: []string{"agent", "run"}}, + {Name: string(apicommon.TraceAgentContainerName), Command: []string{"/entrypoint.sh", "trace-agent"}}, + {Name: string(apicommon.ProcessAgentContainerName), Command: []string{"process-agent"}}, + {Name: string(apicommon.SystemProbeContainerName), Command: []string{"system-probe"}}, + {Name: string(apicommon.HostProfiler), Command: []string{"host-profiler", "--core-config=/etc/datadog-agent/datadog.yaml"}}, + {Name: string(apicommon.OtelAgent), Command: []string{"otel-agent"}}, + {Name: string(apicommon.PrivateActionRunnerContainerName), Command: []string{"/opt/datadog-agent/embedded/bin/privateactionrunner"}}, + } + ds.Spec.Template.Spec.InitContainers = append(ds.Spec.Template.Spec.InitContainers, + corev1.Container{Name: "seccomp-setup"}, corev1.Container{Name: "host-profiler-seccomp-setup"}) + require.NoError(t, prepareAgentTemplate(ds)) + assert.Equal(t, []string{"trace-agent"}, ds.Spec.Template.Spec.Containers[1].Command) +} + +func TestPreparedRolloutSupportsStandaloneHostProfiler(t *testing.T) { + ds := preparedTestDaemonSet(true) + ds.Spec.Template.Spec.Containers = append(ds.Spec.Template.Spec.Containers, + corev1.Container{Name: string(apicommon.HostProfiler), Command: []string{"host-profiler"}}) + require.NoError(t, prepareAgentTemplate(ds)) +} + +func TestPreparedRolloutRejectsUnknownSidecar(t *testing.T) { + ds := preparedTestDaemonSet(true) + ds.Spec.Template.Spec.Containers = append(ds.Spec.Template.Spec.Containers, corev1.Container{Name: "unknown-sidecar", Command: []string{"sidecar"}}) + require.ErrorContains(t, prepareAgentTemplate(ds), "does not support container") +} + +func TestProfileSurgeAntiAffinityAllowsOnlySameProfileOverlap(t *testing.T) { + antiAffinity, ok := profileSurgePodAntiAffinity(map[string]string{ + apicommon.AgentDeploymentNameLabelKey: "agent-a", + constants.ProfileLabelKey: "blue", + }) + require.True(t, ok) + blocked := func(podLabels map[string]string) bool { + for _, term := range antiAffinity.RequiredDuringSchedulingIgnoredDuringExecution { + selector, err := metav1.LabelSelectorAsSelector(term.LabelSelector) + require.NoError(t, err) + if selector.Matches(labels.Set(podLabels)) { + return true + } + } + return false + } + base := map[string]string{apicommon.AgentDeploymentComponentLabelKey: constants.DefaultAgentResourceSuffix} + assert.False(t, blocked(mergeLabels(base, map[string]string{apicommon.AgentDeploymentNameLabelKey: "agent-a", constants.ProfileLabelKey: "blue"}))) + assert.True(t, blocked(mergeLabels(base, map[string]string{apicommon.AgentDeploymentNameLabelKey: "agent-a", constants.ProfileLabelKey: "green"}))) +} + +func preparedRolloutDDAI() *datadoghqv1alpha1.DatadogAgentInternal { + return &datadoghqv1alpha1.DatadogAgentInternal{ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{ + preparedRolloutModeAnnotation: preparedRolloutModeV1, + }}} +} + +func preparedTestDaemonSet(hostNetwork bool) *appsv1.DaemonSet { + port := corev1.ContainerPort{Name: "declared", ContainerPort: 8126, HostPort: 8126, Protocol: corev1.ProtocolTCP} + return &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: "agent", Namespace: "default"}, + Spec: appsv1.DaemonSetSpec{ + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "agent"}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ + "app": "agent", + apicommon.AgentDeploymentNameLabelKey: "agent", + apicommon.AgentDeploymentComponentLabelKey: constants.DefaultAgentResourceSuffix, + }}, + Spec: corev1.PodSpec{ + HostNetwork: hostNetwork, + NodeSelector: map[string]string{corev1.LabelOSStable: "linux"}, + InitContainers: []corev1.Container{ + {Name: string(apicommon.InitVolumeContainerName)}, + {Name: string(apicommon.InitConfigContainerName)}, + }, + Containers: []corev1.Container{ + {Name: string(apicommon.CoreAgentContainerName), Command: []string{"agent", "run"}, Ports: []corev1.ContainerPort{port}, StartupProbe: &corev1.Probe{}, LivenessProbe: &corev1.Probe{}, ReadinessProbe: &corev1.Probe{}}, + {Name: string(apicommon.TraceAgentContainerName), Command: []string{"/entrypoint.sh", "trace-agent"}, Ports: []corev1.ContainerPort{port}, StartupProbe: &corev1.Probe{}, LivenessProbe: &corev1.Probe{}, ReadinessProbe: &corev1.Probe{}}, + }, + Volumes: []corev1.Volume{{Name: "sockets", VolumeSource: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: "/var/run/datadog", Type: ptr.To(corev1.HostPathDirectoryOrCreate)}}}}, + }, + }, + UpdateStrategy: appsv1.DaemonSetUpdateStrategy{Type: appsv1.RollingUpdateDaemonSetStrategyType, RollingUpdate: &appsv1.RollingUpdateDaemonSet{MaxUnavailable: ptr.To(intstr.FromInt(1)), MaxSurge: ptr.To(intstr.FromInt(1))}}, + }, + } +} + +func hasHostPath(volumes []corev1.Volume, path string) bool { + for i := range volumes { + if volumes[i].HostPath != nil && volumes[i].HostPath.Path == path { + return true + } + } + return false +} + +func containerEnv(container *corev1.Container, name string) *corev1.EnvVar { + for i := range container.Env { + if container.Env[i].Name == name { + return &container.Env[i] + } + } + return nil +} + +func mergeLabels(left, right map[string]string) map[string]string { + result := make(map[string]string, len(left)+len(right)) + for key, value := range left { + result[key] = value + } + for key, value := range right { + result[key] = value + } + return result +} diff --git a/internal/controller/datadogagentinternal/resource_fallback.go b/internal/controller/datadogagentinternal/resource_fallback.go new file mode 100644 index 0000000000..09b9d28feb --- /dev/null +++ b/internal/controller/datadogagentinternal/resource_fallback.go @@ -0,0 +1,408 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// Copyright 2016-present Datadog, Inc. + +package datadogagentinternal + +import ( + "context" + "fmt" + "sort" + "strconv" + "strings" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1" +) + +type fallbackCandidate struct { + pending *corev1.Pod + old *corev1.Pod + nodeName string +} + +type runningHandoffCandidate struct { + replacement *corev1.Pod + old *corev1.Pod + nodeName string +} + +const resourceFallbackPollInterval = 5 * time.Second + +// reconcilePreparedRollout authorizes one node handoff when a surged replacement +// is running but intentionally unready. If no replacement fits on its node, it +// retains the CPU/memory-only delete-before-create fallback. +func (r *Reconciler) reconcilePreparedRollout(ctx context.Context, ddai *datadoghqv1alpha1.DatadogAgentInternal, expectedDS *appsv1.DaemonSet, budgetValue intstr.IntOrString) (reconcile.Result, error) { + reader := r.apiReader + if reader == nil { + reader = r.client + } + ds := &appsv1.DaemonSet{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(expectedDS), ds); err != nil { + return reconcile.Result{}, client.IgnoreNotFound(err) + } + if !daemonSetControlledByDDAI(ds, ddai) || !preparedRolloutDaemonSetEligible(ds) || !hasRolloutMode(ds.Spec.Template.Annotations) { + return reconcile.Result{}, nil + } + if ds.Status.DesiredNumberScheduled <= 0 || ds.Status.ObservedGeneration != ds.Generation { + return resourceFallbackPollResult(ds), nil + } + budget, budgetErr := intstr.GetScaledValueFromIntOrPercent(&budgetValue, int(ds.Status.DesiredNumberScheduled), true) + if budgetErr != nil { + return reconcile.Result{}, fmt.Errorf("resolve Agent resource fallback budget: %w", budgetErr) + } + if budget <= 0 { + return reconcile.Result{}, nil + } + pods, podsErr := daemonSetPods(ctx, reader, ds) + if podsErr != nil { + return reconcile.Result{}, podsErr + } + if consumedPreparedRolloutBudget(ds, pods) >= budget { + return resourceFallbackPollResult(ds), nil + } + desiredRevision, revisionErr := currentDaemonSetRevision(ctx, reader, ds) + if revisionErr != nil { + return reconcile.Result{}, revisionErr + } + if desiredRevision == "" { + return resourceFallbackPollResult(ds), nil + } + + runningCandidates := runningHandoffCandidates(ds, pods, desiredRevision, time.Now()) + if len(runningCandidates) > 0 { + candidate := runningCandidates[0] + replacement := &corev1.Pod{} + old := &corev1.Pod{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(candidate.replacement), replacement); err != nil { + return reconcile.Result{}, client.IgnoreNotFound(err) + } + if err := reader.Get(ctx, client.ObjectKeyFromObject(candidate.old), old); err != nil { + return reconcile.Result{}, client.IgnoreNotFound(err) + } + if replacement.UID != candidate.replacement.UID || old.UID != candidate.old.UID || + !controlledByUID(replacement, ds.UID) || !controlledByUID(old, ds.UID) || + !replacementRunningForHandoff(replacement, ds, desiredRevision) || + old.Spec.NodeName != candidate.nodeName || replacement.Spec.NodeName != candidate.nodeName || + old.DeletionTimestamp != nil || !podAvailable(old, ds.Spec.MinReadySeconds, time.Now()) || + podRevision(old) == "" || podRevision(old) == desiredRevision { + return resourceFallbackPollResult(ds), nil + } + allowed, err := preparedRolloutDeletionAllowed(ctx, reader, ds, budget, desiredRevision) + if err != nil { + return reconcile.Result{}, err + } + if !allowed { + return resourceFallbackPollResult(ds), nil + } + + uid := old.UID + if err := r.client.Delete(ctx, old, &client.DeleteOptions{Preconditions: &metav1.Preconditions{UID: &uid}}); err != nil && !apierrors.IsNotFound(err) { + return reconcile.Result{}, fmt.Errorf("delete old Agent Pod %s/%s for prepared handoff: %w", old.Namespace, old.Name, err) + } + ctrl.LoggerFrom(ctx).WithValues("daemonset", ds.Name, "node", candidate.nodeName, "oldPod", old.Name, "replacementPod", replacement.Name).Info("Deleted old Agent Pod after every replacement container started") + if r.recorder != nil { + r.recorder.Eventf(ddai, corev1.EventTypeNormal, "AgentPreparedHandoff", "Deleted old Agent Pod %s on node %s after every container in replacement %s started", old.Name, candidate.nodeName, replacement.Name) + } + return reconcile.Result{RequeueAfter: time.Second}, nil + } + + candidates := fallbackCandidates(ds, pods, desiredRevision, time.Now()) + if len(candidates) == 0 { + return resourceFallbackPollResult(ds), nil + } + candidate := candidates[0] + + // Re-read both Pods immediately before deletion. The scheduler condition, + // target node and old Pod identity must still describe the same handoff. + pending := &corev1.Pod{} + old := &corev1.Pod{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(candidate.pending), pending); err != nil { + return reconcile.Result{}, client.IgnoreNotFound(err) + } + if err := reader.Get(ctx, client.ObjectKeyFromObject(candidate.old), old); err != nil { + return reconcile.Result{}, client.IgnoreNotFound(err) + } + if pending.UID != candidate.pending.UID || old.UID != candidate.old.UID || !controlledByUID(pending, ds.UID) || !controlledByUID(old, ds.UID) { + return resourceFallbackPollResult(ds), nil + } + if _, ok := resourceOnlyUnschedulable(pending); !ok || podRevision(pending) != desiredRevision || !replacementSpecMatchesTemplate(pending, ds) { + return resourceFallbackPollResult(ds), nil + } + nodeName, ok := targetNodeFromDaemonSetAffinity(pending) + if !ok || nodeName != candidate.nodeName || pending.Spec.NodeName != "" || pending.DeletionTimestamp != nil || old.Spec.NodeName != nodeName || old.DeletionTimestamp != nil || !podAvailable(old, ds.Spec.MinReadySeconds, time.Now()) || podRevision(old) == "" || podRevision(old) == desiredRevision { + return resourceFallbackPollResult(ds), nil + } + allowed, err := preparedRolloutDeletionAllowed(ctx, reader, ds, budget, desiredRevision) + if err != nil { + return reconcile.Result{}, err + } + if !allowed { + return resourceFallbackPollResult(ds), nil + } + + uid := old.UID + if err := r.client.Delete(ctx, old, &client.DeleteOptions{Preconditions: &metav1.Preconditions{UID: &uid}}); err != nil && !apierrors.IsNotFound(err) { + return reconcile.Result{}, fmt.Errorf("delete old Agent Pod %s/%s for resource fallback: %w", old.Namespace, old.Name, err) + } + ctrl.LoggerFrom(ctx).WithValues("daemonset", ds.Name, "node", nodeName, "oldPod", old.Name, "replacementPod", pending.Name).Info("Deleted old Agent Pod because the surged replacement was blocked by node CPU or memory") + if r.recorder != nil { + r.recorder.Eventf(ddai, corev1.EventTypeWarning, "AgentResourceFallback", "Deleted old Agent Pod %s on node %s because replacement %s was blocked by CPU or memory", old.Name, nodeName, pending.Name) + } + return reconcile.Result{RequeueAfter: time.Second}, nil +} + +// preparedRolloutDeletionAllowed narrows the unavoidable observation-to-delete +// race by rechecking the live rollout generation, revision and unavailability +// immediately before deleting an old Pod. +func preparedRolloutDeletionAllowed(ctx context.Context, reader client.Reader, expectedDS *appsv1.DaemonSet, budget int, desiredRevision string) (bool, error) { + ds := &appsv1.DaemonSet{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(expectedDS), ds); err != nil { + return false, client.IgnoreNotFound(err) + } + if ds.UID != expectedDS.UID || ds.Generation != expectedDS.Generation || ds.Status.ObservedGeneration != ds.Generation || !preparedRolloutDaemonSetEligible(ds) || !hasRolloutMode(ds.Spec.Template.Annotations) { + return false, nil + } + liveRevision, err := currentDaemonSetRevision(ctx, reader, ds) + if err != nil { + return false, err + } + if liveRevision == "" || liveRevision != desiredRevision { + return false, nil + } + pods, err := daemonSetPods(ctx, reader, ds) + if err != nil { + return false, err + } + return consumedPreparedRolloutBudget(ds, pods) < budget, nil +} + +func runningHandoffCandidates(ds *appsv1.DaemonSet, pods []corev1.Pod, desiredRevision string, now time.Time) []runningHandoffCandidate { + newByNode := map[string][]*corev1.Pod{} + oldByNode := map[string][]*corev1.Pod{} + for i := range pods { + pod := &pods[i] + if replacementRunningForHandoff(pod, ds, desiredRevision) { + newByNode[pod.Spec.NodeName] = append(newByNode[pod.Spec.NodeName], pod) + continue + } + if pod.Spec.NodeName != "" && pod.DeletionTimestamp == nil && podRevision(pod) != "" && podRevision(pod) != desiredRevision && podAvailable(pod, ds.Spec.MinReadySeconds, now) { + oldByNode[pod.Spec.NodeName] = append(oldByNode[pod.Spec.NodeName], pod) + } + } + + var candidates []runningHandoffCandidate + for nodeName, replacements := range newByNode { + olds := oldByNode[nodeName] + if len(replacements) == 1 && len(olds) == 1 { + candidates = append(candidates, runningHandoffCandidate{replacement: replacements[0], old: olds[0], nodeName: nodeName}) + } + } + sort.Slice(candidates, func(i, j int) bool { return candidates[i].nodeName < candidates[j].nodeName }) + return candidates +} + +func replacementRunningForHandoff(pod *corev1.Pod, ds *appsv1.DaemonSet, desiredRevision string) bool { + if pod.DeletionTimestamp != nil || pod.Spec.NodeName == "" || pod.Status.Phase != corev1.PodRunning || podRevision(pod) != desiredRevision || !podInitialized(pod) { + return false + } + if !replacementSpecMatchesTemplate(pod, ds) { + return false + } + + desiredImages := make(map[string]string, len(ds.Spec.Template.Spec.Containers)) + for i := range ds.Spec.Template.Spec.Containers { + container := &ds.Spec.Template.Spec.Containers[i] + if container.Name == "" { + return false + } + desiredImages[container.Name] = container.Image + } + if len(pod.Status.ContainerStatuses) != len(desiredImages) { + return false + } + + seen := make(map[string]struct{}, len(pod.Status.ContainerStatuses)) + for i := range pod.Status.ContainerStatuses { + status := &pod.Status.ContainerStatuses[i] + if _, ok := desiredImages[status.Name]; !ok || status.State.Running == nil || status.RestartCount != 0 || status.ContainerID == "" || status.ImageID == "" { + return false + } + if _, duplicate := seen[status.Name]; duplicate { + return false + } + seen[status.Name] = struct{}{} + } + return len(seen) == len(desiredImages) +} + +func replacementSpecMatchesTemplate(pod *corev1.Pod, ds *appsv1.DaemonSet) bool { + desiredImages := make(map[string]string, len(ds.Spec.Template.Spec.Containers)) + for i := range ds.Spec.Template.Spec.Containers { + container := &ds.Spec.Template.Spec.Containers[i] + if container.Name == "" { + return false + } + desiredImages[container.Name] = container.Image + } + if len(pod.Spec.Containers) != len(desiredImages) { + return false + } + for i := range pod.Spec.Containers { + container := &pod.Spec.Containers[i] + if desiredImage, ok := desiredImages[container.Name]; !ok || desiredImage != container.Image { + return false + } + } + return true +} + +func podInitialized(pod *corev1.Pod) bool { + for i := range pod.Status.Conditions { + condition := &pod.Status.Conditions[i] + if condition.Type == corev1.PodInitialized { + return condition.Status == corev1.ConditionTrue + } + } + return false +} + +func podRevision(pod *corev1.Pod) string { + return pod.Labels[appsv1.DefaultDaemonSetUniqueLabelKey] +} + +func resourceFallbackPollResult(ds *appsv1.DaemonSet) reconcile.Result { + if daemonSetFullyRolledOut(ds) { + return reconcile.Result{} + } + return reconcile.Result{RequeueAfter: resourceFallbackPollInterval} +} + +func fallbackCandidates(ds *appsv1.DaemonSet, pods []corev1.Pod, desiredRevision string, now time.Time) []fallbackCandidate { + var candidates []fallbackCandidate + for i := range pods { + pending := &pods[i] + if _, ok := resourceOnlyUnschedulable(pending); !ok || pending.DeletionTimestamp != nil || pending.Spec.NodeName != "" || pending.Status.NominatedNodeName != "" || podRevision(pending) != desiredRevision || !replacementSpecMatchesTemplate(pending, ds) { + continue + } + nodeName, ok := targetNodeFromDaemonSetAffinity(pending) + if !ok { + continue + } + var oldPods []*corev1.Pod + for j := range pods { + old := &pods[j] + if old.Spec.NodeName == nodeName && old.DeletionTimestamp == nil && podRevision(old) != "" && podRevision(old) != desiredRevision && podAvailable(old, ds.Spec.MinReadySeconds, now) { + oldPods = append(oldPods, old) + } + } + if len(oldPods) == 1 { + candidates = append(candidates, fallbackCandidate{pending: pending, old: oldPods[0], nodeName: nodeName}) + } + } + sort.Slice(candidates, func(i, j int) bool { return candidates[i].nodeName < candidates[j].nodeName }) + return candidates +} + +func consumedPreparedRolloutBudget(ds *appsv1.DaemonSet, pods []corev1.Pod) int { + terminatingNodes := map[string]struct{}{} + for i := range pods { + if pods[i].DeletionTimestamp != nil && pods[i].Spec.NodeName != "" { + terminatingNodes[pods[i].Spec.NodeName] = struct{}{} + } + } + // NumberUnavailable may already include some terminating Pods. Counting both + // is deliberately conservative: fallback must never delete more old Agents + // merely because DaemonSet status and Pod deletion are observed at different + // times. + return int(ds.Status.NumberUnavailable) + len(terminatingNodes) +} + +type resourceShortage struct { + cpu bool + memory bool +} + +func resourceOnlyUnschedulable(pod *corev1.Pod) (resourceShortage, bool) { + condition := scheduledCondition(pod) + if condition == nil || condition.Status != corev1.ConditionFalse || condition.Reason != corev1.PodReasonUnschedulable { + return resourceShortage{}, false + } + primary := condition.Message + lower := strings.ToLower(primary) + if i := strings.Index(lower, "preemption:"); i >= 0 { + primary = primary[:i] + lower = lower[:i] + } + if i := strings.Index(lower, "nodes are available:"); i >= 0 { + primary = primary[i+len("nodes are available:"):] + } + primary = strings.TrimSuffix(strings.TrimSpace(primary), ".") + var shortage resourceShortage + for reason := range strings.SplitSeq(primary, ", ") { + fields := strings.Fields(strings.ToLower(strings.TrimSpace(reason))) + if len(fields) < 2 { + return resourceShortage{}, false + } + if _, err := strconv.Atoi(fields[0]); err != nil { + return resourceShortage{}, false + } + switch strings.Join(fields[1:], " ") { + case "insufficient cpu": + shortage.cpu = true + case "insufficient memory": + shortage.memory = true + case "node(s) didn't match pod's node affinity/selector", "node(s) didn't satisfy plugin(s) [nodeaffinity]": + // Expected on non-target nodes for DaemonSet surge Pods. + default: + return resourceShortage{}, false + } + } + return shortage, shortage.cpu || shortage.memory +} + +func scheduledCondition(pod *corev1.Pod) *corev1.PodCondition { + for i := range pod.Status.Conditions { + if pod.Status.Conditions[i].Type == corev1.PodScheduled { + return &pod.Status.Conditions[i] + } + } + return nil +} + +func targetNodeFromDaemonSetAffinity(pod *corev1.Pod) (string, bool) { + if pod.Spec.Affinity == nil || pod.Spec.Affinity.NodeAffinity == nil || pod.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil { + return "", false + } + terms := pod.Spec.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms + if len(terms) == 0 { + return "", false + } + var target string + for _, term := range terms { + var termTarget string + for _, requirement := range term.MatchFields { + if requirement.Key != metav1.ObjectNameField { + continue + } + if requirement.Operator != corev1.NodeSelectorOpIn || len(requirement.Values) != 1 || requirement.Values[0] == "" || termTarget != "" { + return "", false + } + termTarget = requirement.Values[0] + } + if termTarget == "" || target != "" && termTarget != target { + return "", false + } + target = termTarget + } + return target, target != "" +} diff --git a/internal/controller/datadogagentinternal/resource_fallback_test.go b/internal/controller/datadogagentinternal/resource_fallback_test.go new file mode 100644 index 0000000000..d6129ca6a9 --- /dev/null +++ b/internal/controller/datadogagentinternal/resource_fallback_test.go @@ -0,0 +1,285 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// Copyright 2016-present Datadog, Inc. + +package datadogagentinternal + +import ( + "context" + "encoding/json" + "testing" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + datadoghqv1alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v1alpha1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestResourceFallbackDeletesOneOldPodForResourceOnlyFailure(t *testing.T) { + fixture := newResourceFallbackFixture(t) + result, err := fixture.reconciler.reconcilePreparedRollout(context.Background(), fixture.ddai, fixture.ds, intstr.FromInt(1)) + require.NoError(t, err) + assert.Equal(t, time.Second, result.RequeueAfter) + err = fixture.client.Get(context.Background(), client.ObjectKeyFromObject(fixture.old), &corev1.Pod{}) + assert.True(t, apierrors.IsNotFound(err)) + assert.NoError(t, fixture.client.Get(context.Background(), client.ObjectKeyFromObject(fixture.pending), &corev1.Pod{})) +} + +func TestResourceFallbackIgnoresOtherSchedulingFailures(t *testing.T) { + fixture := newResourceFallbackFixture(t) + fixture.pending.Status.Conditions[0].Message = "0/1 nodes are available: 1 node(s) didn't have free ports for the requested pod ports, 1 Insufficient cpu." + updatePodStatus(t, fixture.client, fixture.pending) + + result, err := fixture.reconciler.reconcilePreparedRollout(context.Background(), fixture.ddai, fixture.ds, intstr.FromInt(1)) + require.NoError(t, err) + assert.Equal(t, resourceFallbackPollInterval, result.RequeueAfter) + assert.NoError(t, fixture.client.Get(context.Background(), client.ObjectKeyFromObject(fixture.old), &corev1.Pod{})) +} + +func TestResourceFallbackRespectsExistingUnavailableBudget(t *testing.T) { + fixture := newResourceFallbackFixture(t) + fixture.ds.Status.NumberUnavailable = 1 + updateDaemonSetStatus(t, fixture.client, fixture.ds) + + result, err := fixture.reconciler.reconcilePreparedRollout(context.Background(), fixture.ddai, fixture.ds, intstr.FromInt(1)) + require.NoError(t, err) + assert.Equal(t, resourceFallbackPollInterval, result.RequeueAfter) + assert.NoError(t, fixture.client.Get(context.Background(), client.ObjectKeyFromObject(fixture.old), &corev1.Pod{})) +} + +func TestResourceFallbackRequiresAnOldRevisionOnTheTargetNode(t *testing.T) { + fixture := newResourceFallbackFixture(t) + fixture.old.Labels[appsv1.DefaultDaemonSetUniqueLabelKey] = "new-hash" + require.NoError(t, fixture.client.Update(context.Background(), fixture.old)) + + result, err := fixture.reconciler.reconcilePreparedRollout(context.Background(), fixture.ddai, fixture.ds, intstr.FromInt(1)) + require.NoError(t, err) + assert.Equal(t, resourceFallbackPollInterval, result.RequeueAfter) + assert.NoError(t, fixture.client.Get(context.Background(), client.ObjectKeyFromObject(fixture.old), &corev1.Pod{})) +} + +func TestPreparedRolloutDeletionRechecksLiveBudgetAndGeneration(t *testing.T) { + fixture := newResourceFallbackFixture(t) + + allowed, err := preparedRolloutDeletionAllowed(context.Background(), fixture.client, fixture.ds, 1, "new-hash") + require.NoError(t, err) + assert.True(t, allowed) + + fixture.ds.Status.NumberUnavailable = 1 + updateDaemonSetStatus(t, fixture.client, fixture.ds) + allowed, err = preparedRolloutDeletionAllowed(context.Background(), fixture.client, fixture.ds, 1, "new-hash") + require.NoError(t, err) + assert.False(t, allowed) + + fixture.ds.Status.NumberUnavailable = 0 + fixture.ds.Status.ObservedGeneration-- + updateDaemonSetStatus(t, fixture.client, fixture.ds) + allowed, err = preparedRolloutDeletionAllowed(context.Background(), fixture.client, fixture.ds, 1, "new-hash") + require.NoError(t, err) + assert.False(t, allowed) +} + +func TestResourceFallbackStopsPollingAfterRollout(t *testing.T) { + fixture := newResourceFallbackFixture(t) + fixture.ds.Status.UpdatedNumberScheduled = 1 + updateDaemonSetStatus(t, fixture.client, fixture.ds) + fixture.pending.Status.Conditions[0].Message = "0/1 nodes are available: 1 node(s) had untolerated taint." + updatePodStatus(t, fixture.client, fixture.pending) + + result, err := fixture.reconciler.reconcilePreparedRollout(context.Background(), fixture.ddai, fixture.ds, intstr.FromInt(1)) + require.NoError(t, err) + assert.Zero(t, result.RequeueAfter) +} + +func TestPreparedRolloutPollsUntilDaemonSetStatusObservesTemplate(t *testing.T) { + fixture := newResourceFallbackFixture(t) + fixture.ds.Status.ObservedGeneration-- + updateDaemonSetStatus(t, fixture.client, fixture.ds) + + result, err := fixture.reconciler.reconcilePreparedRollout(context.Background(), fixture.ddai, fixture.ds, intstr.FromInt(1)) + require.NoError(t, err) + assert.Equal(t, resourceFallbackPollInterval, result.RequeueAfter) +} + +func TestConsumedFallbackBudgetCountsTerminatingNodes(t *testing.T) { + fixture := newResourceFallbackFixture(t) + fixture.ds.Status.NumberUnavailable = 1 + now := metav1.Now() + terminatingA := fixture.old.DeepCopy() + terminatingA.DeletionTimestamp = &now + terminatingB := fixture.old.DeepCopy() + terminatingB.Name = "old-b" + terminatingB.UID = "old-b-uid" + terminatingB.Spec.NodeName = "node-b" + terminatingB.DeletionTimestamp = &now + assert.Equal(t, 3, consumedPreparedRolloutBudget(fixture.ds, []corev1.Pod{*terminatingA, *terminatingB})) +} + +func TestPreparedRolloutDeletesOldPodAfterReplacementContainersAreRunning(t *testing.T) { + fixture := newResourceFallbackFixture(t) + replacement := fixture.pending.DeepCopy() + replacement.Spec.NodeName = "node-a" + replacement.Spec.Affinity = nil + replacement.Labels[appsv1.DefaultDaemonSetUniqueLabelKey] = "new-hash" + replacement.Status = corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{Type: corev1.PodInitialized, Status: corev1.ConditionTrue}}, + ContainerStatuses: []corev1.ContainerStatus{{ + Name: "agent", Image: "agent:new", ImageID: "sha256:new", ContainerID: "containerd://new", + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{StartedAt: metav1.Now()}}, + }}, + } + require.NoError(t, fixture.client.Delete(context.Background(), fixture.pending)) + replacement.ResourceVersion = "" + require.NoError(t, fixture.client.Create(context.Background(), replacement)) + require.NoError(t, fixture.client.Status().Update(context.Background(), replacement)) + + result, err := fixture.reconciler.reconcilePreparedRollout(context.Background(), fixture.ddai, fixture.ds, intstr.FromInt(1)) + require.NoError(t, err) + assert.Equal(t, time.Second, result.RequeueAfter) + err = fixture.client.Get(context.Background(), client.ObjectKeyFromObject(fixture.old), &corev1.Pod{}) + assert.True(t, apierrors.IsNotFound(err)) +} + +func TestPreparedRolloutRequiresEveryReplacementContainerToBePristineAndRunning(t *testing.T) { + tests := map[string]func(*corev1.Pod){ + "not initialized": func(p *corev1.Pod) { p.Status.Conditions = nil }, + "container terminated": func(p *corev1.Pod) { + p.Status.ContainerStatuses[0].State = corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{}} + }, + "container restarted": func(p *corev1.Pod) { p.Status.ContainerStatuses[0].RestartCount = 1 }, + "missing container id": func(p *corev1.Pod) { p.Status.ContainerStatuses[0].ContainerID = "" }, + "missing image id": func(p *corev1.Pod) { p.Status.ContainerStatuses[0].ImageID = "" }, + "wrong image": func(p *corev1.Pod) { p.Spec.Containers[0].Image = "agent:wrong" }, + "wrong revision": func(p *corev1.Pod) { p.Labels[appsv1.DefaultDaemonSetUniqueLabelKey] = "old-hash" }, + } + for name, mutate := range tests { + t.Run(name, func(t *testing.T) { + fixture := newResourceFallbackFixture(t) + replacement := runningReplacement(fixture) + mutate(replacement) + require.False(t, replacementRunningForHandoff(replacement, fixture.ds, "new-hash")) + }) + } +} + +func TestResourceOnlyUnschedulableParsing(t *testing.T) { + pod := &corev1.Pod{Status: corev1.PodStatus{Conditions: []corev1.PodCondition{{ + Type: corev1.PodScheduled, Status: corev1.ConditionFalse, Reason: corev1.PodReasonUnschedulable, + Message: "0/3 nodes are available: 1 Insufficient cpu, 2 node(s) didn't match Pod's node affinity/selector. preemption: 0/3 nodes are available", + }}}} + shortage, ok := resourceOnlyUnschedulable(pod) + assert.True(t, ok) + assert.True(t, shortage.cpu) + assert.False(t, shortage.memory) + pod.Status.Conditions[0].Message = "0/1 nodes are available: 1 node(s) had untolerated taint." + _, ok = resourceOnlyUnschedulable(pod) + assert.False(t, ok) +} + +type resourceFallbackFixture struct { + client client.Client + reconciler *Reconciler + ddai *datadoghqv1alpha1.DatadogAgentInternal + ds *appsv1.DaemonSet + old *corev1.Pod + pending *corev1.Pod +} + +func newResourceFallbackFixture(t *testing.T) resourceFallbackFixture { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, appsv1.AddToScheme(scheme)) + require.NoError(t, datadoghqv1alpha1.AddToScheme(scheme)) + + ddai := &datadoghqv1alpha1.DatadogAgentInternal{ObjectMeta: metav1.ObjectMeta{Name: "agent", Namespace: "default", UID: "ddai-uid"}} + ds := &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: "agent", Namespace: "default", UID: "ds-uid", Generation: 2, OwnerReferences: []metav1.OwnerReference{{ + APIVersion: datadoghqv1alpha1.GroupVersion.String(), Kind: "DatadogAgentInternal", Name: ddai.Name, UID: ddai.UID, Controller: ptr.To(true), + }}}, + Spec: appsv1.DaemonSetSpec{ + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "agent"}}, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "agent"}, Annotations: map[string]string{preparedRolloutModeAnnotation: preparedRolloutModeV1}}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "agent", Image: "agent:new"}}}, + }, + UpdateStrategy: appsv1.DaemonSetUpdateStrategy{Type: appsv1.RollingUpdateDaemonSetStrategyType, RollingUpdate: &appsv1.RollingUpdateDaemonSet{MaxSurge: ptr.To(intstr.FromInt(1)), MaxUnavailable: ptr.To(intstr.FromInt(0))}}, + }, + Status: appsv1.DaemonSetStatus{ObservedGeneration: 2, DesiredNumberScheduled: 1, NumberReady: 1, NumberAvailable: 1}, + } + owner := metav1.OwnerReference{APIVersion: appsv1.SchemeGroupVersion.String(), Kind: "DaemonSet", Name: ds.Name, UID: ds.UID, Controller: ptr.To(true)} + old := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "old", Namespace: "default", UID: "old-uid", Labels: map[string]string{"app": "agent", appsv1.DefaultDaemonSetUniqueLabelKey: "old-hash"}, OwnerReferences: []metav1.OwnerReference{owner}}, + Spec: corev1.PodSpec{NodeName: "node-a", Containers: []corev1.Container{{Name: "agent", Image: "agent:old"}}}, + Status: corev1.PodStatus{Phase: corev1.PodRunning, Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue, LastTransitionTime: metav1.NewTime(time.Now().Add(-time.Minute))}}}, + } + pending := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "new", Namespace: "default", UID: "new-uid", Labels: map[string]string{"app": "agent", appsv1.DefaultDaemonSetUniqueLabelKey: "new-hash"}, OwnerReferences: []metav1.OwnerReference{owner}}, + Spec: corev1.PodSpec{ + Affinity: &corev1.Affinity{NodeAffinity: &corev1.NodeAffinity{RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{NodeSelectorTerms: []corev1.NodeSelectorTerm{{MatchFields: []corev1.NodeSelectorRequirement{{Key: metav1.ObjectNameField, Operator: corev1.NodeSelectorOpIn, Values: []string{"node-a"}}}}}}}}, + Containers: []corev1.Container{{Name: "agent", Image: "agent:new"}}, + }, + Status: corev1.PodStatus{Conditions: []corev1.PodCondition{{Type: corev1.PodScheduled, Status: corev1.ConditionFalse, Reason: corev1.PodReasonUnschedulable, Message: "0/1 nodes are available: 1 Insufficient cpu."}}}, + } + revisionData, err := json.Marshal(struct { + Spec struct { + Template corev1.PodTemplateSpec `json:"template"` + } `json:"spec"` + }{Spec: struct { + Template corev1.PodTemplateSpec `json:"template"` + }{Template: ds.Spec.Template}}) + require.NoError(t, err) + revision := &appsv1.ControllerRevision{ + ObjectMeta: metav1.ObjectMeta{ + Name: "agent-new-hash", Namespace: ds.Namespace, + Labels: map[string]string{appsv1.DefaultDaemonSetUniqueLabelKey: "new-hash"}, + OwnerReferences: []metav1.OwnerReference{owner}, + }, + Data: runtime.RawExtension{Raw: revisionData}, + Revision: 2, + } + c := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(&corev1.Pod{}, &appsv1.DaemonSet{}).WithObjects(ddai, ds, old, pending, revision).Build() + return resourceFallbackFixture{client: c, reconciler: &Reconciler{client: c, apiReader: c}, ddai: ddai, ds: ds, old: old, pending: pending} +} + +func runningReplacement(fixture resourceFallbackFixture) *corev1.Pod { + pod := fixture.pending.DeepCopy() + pod.Spec.NodeName = "node-a" + pod.Spec.Affinity = nil + pod.Status = corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{Type: corev1.PodInitialized, Status: corev1.ConditionTrue}}, + ContainerStatuses: []corev1.ContainerStatus{{ + Name: "agent", Image: "agent:new", ImageID: "sha256:new", ContainerID: "containerd://new", + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{StartedAt: metav1.Now()}}, + }}, + } + return pod +} + +func updatePodStatus(t *testing.T, c client.Client, pod *corev1.Pod) { + t.Helper() + live := &corev1.Pod{} + require.NoError(t, c.Get(context.Background(), client.ObjectKeyFromObject(pod), live)) + live.Status = *pod.Status.DeepCopy() + require.NoError(t, c.Status().Update(context.Background(), live)) +} + +func updateDaemonSetStatus(t *testing.T, c client.Client, ds *appsv1.DaemonSet) { + t.Helper() + live := &appsv1.DaemonSet{} + require.NoError(t, c.Get(context.Background(), client.ObjectKeyFromObject(ds), live)) + live.Status = ds.Status + require.NoError(t, c.Status().Update(context.Background(), live)) +} diff --git a/internal/controller/datadogagentinternal_controller.go b/internal/controller/datadogagentinternal_controller.go index 53429b537d..e2a6de7dbb 100644 --- a/internal/controller/datadogagentinternal_controller.go +++ b/internal/controller/datadogagentinternal_controller.go @@ -46,6 +46,8 @@ type DatadogAgentInternalReconciler struct { // +kubebuilder:rbac:groups=datadoghq.com,resources=datadogagentinternals,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=datadoghq.com,resources=datadogagentinternals/status,verbs=get;update;patch // +kubebuilder:rbac:groups=datadoghq.com,resources=datadogagentinternals/finalizers,verbs=get;list;watch;create;update;patch;delete +// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;delete +// +kubebuilder:rbac:groups=apps,resources=controllerrevisions,verbs=get;list // Reconcile loop for DatadogAgent. func (r *DatadogAgentInternalReconciler) Reconcile(ctx context.Context, ddai *v1alpha1.DatadogAgentInternal) (ctrl.Result, error) { @@ -58,10 +60,11 @@ func (r *DatadogAgentInternalReconciler) Reconcile(ctx context.Context, ddai *v1 // SetupWithManager creates a new DatadogAgent controller. func (r *DatadogAgentInternalReconciler) SetupWithManager(mgr ctrl.Manager, metricForwardersMgr datadog.MetricsForwardersManager) error { + generationChanged := ctrlbuilder.WithPredicates(predicate.GenerationChangedPredicate{}) builder := ctrl.NewControllerManagedBy(mgr). Owns(&corev1.Secret{}). Owns(&corev1.ConfigMap{}). - Owns(&appsv1.DaemonSet{}). + Owns(&appsv1.DaemonSet{}, generationChanged). Owns(&appsv1.Deployment{}). Owns(&rbacv1.Role{}). Owns(&rbacv1.RoleBinding{}). @@ -76,7 +79,6 @@ func (r *DatadogAgentInternalReconciler) SetupWithManager(mgr ctrl.Manager, metr handlerEnqueue := handler.EnqueueRequestsFromMapFunc(enqueueIfOwnedByDatadogAgentInternal) builder.Watches(&rbacv1.ClusterRole{}, handlerEnqueue) builder.Watches(&rbacv1.ClusterRoleBinding{}, handlerEnqueue) - if r.Options.ExtendedDaemonsetOptions.Enabled { builder = builder.Owns(&edsdatadoghqv1alpha1.ExtendedDaemonSet{}) } @@ -101,17 +103,27 @@ func (r *DatadogAgentInternalReconciler) SetupWithManager(mgr ctrl.Manager, metr }, })) } + builderOptions = append(builderOptions, ctrlbuilder.WithPredicates(datadogAgentInternalEventPredicate())) or := reconcile.AsReconciler[*v1alpha1.DatadogAgentInternal](r.Client, r) - if err := builder.For(&datadoghqv1alpha1.DatadogAgentInternal{}, builderOptions...).WithEventFilter(predicate.GenerationChangedPredicate{}).Complete(or); err != nil { + if err := builder.For(&datadoghqv1alpha1.DatadogAgentInternal{}, builderOptions...).Complete(or); err != nil { return err } - r.internal = datadogagentinternal.NewReconciler(r.Options, r.Client, r.PlatformInfo, r.Scheme, r.Recorder, metricForwardersMgr) + internalOptions := r.Options + internalOptions.APIReader = mgr.GetAPIReader() + r.internal = datadogagentinternal.NewReconciler(internalOptions, r.Client, r.PlatformInfo, r.Scheme, r.Recorder, metricForwardersMgr) return nil } +func datadogAgentInternalEventPredicate() predicate.Predicate { + return predicate.Or( + predicate.GenerationChangedPredicate{}, + datadogAnnotationChangedPredicate(), + ) +} + func enqueueIfOwnedByDatadogAgentInternal(ctx context.Context, obj client.Object) []reconcile.Request { labels := obj.GetLabels() diff --git a/internal/controller/datadogagentinternal_controller_test.go b/internal/controller/datadogagentinternal_controller_test.go new file mode 100644 index 0000000000..b2d0efc321 --- /dev/null +++ b/internal/controller/datadogagentinternal_controller_test.go @@ -0,0 +1,32 @@ +// Unless explicitly stated otherwise all files in this repository are licensed +// under the Apache License Version 2.0. +// Copyright 2016-present Datadog, Inc. + +package controller + +import ( + "context" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/DataDog/datadog-operator/internal/controller/datadogagent/object" + "github.com/DataDog/datadog-operator/pkg/kubernetes" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestEnqueueIfOwnedByDatadogAgentInternal(t *testing.T) { + unmanaged := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ + kubernetes.AppKubernetesPartOfLabelKey: "default-profile--ddai", + }}} + assert.Empty(t, enqueueIfOwnedByDatadogAgentInternal(context.Background(), unmanaged)) + + managed := unmanaged.DeepCopy() + managed.Labels[kubernetes.AppKubernetesManageByLabelKey] = "datadog-operator" + requests := enqueueIfOwnedByDatadogAgentInternal(context.Background(), managed) + require.Len(t, requests, 1) + owner := object.PartOfLabelValue{Value: "default-profile--ddai"} + assert.Equal(t, owner.NamespacedName(), requests[0].NamespacedName) +} diff --git a/internal/controller/testutils/renderer/render_e2e_test.go b/internal/controller/testutils/renderer/render_e2e_test.go index 89dbe07234..56fa0e1e9b 100644 --- a/internal/controller/testutils/renderer/render_e2e_test.go +++ b/internal/controller/testutils/renderer/render_e2e_test.go @@ -9,6 +9,13 @@ import ( "strings" "testing" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" + + common "github.com/DataDog/datadog-operator/api/datadoghq/common" + datadoghqv2alpha1 "github.com/DataDog/datadog-operator/api/datadoghq/v2alpha1" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/runtime" @@ -159,6 +166,130 @@ func TestRender_AppArmorProfileVersionGate(t *testing.T) { } } +func TestRender_PreparedRolloutUsesNativeSurge(t *testing.T) { + renderAgentDaemonSet := func(t *testing.T, prepared bool) *appsv1.DaemonSet { + t.Helper() + dda, err := LoadDDA("testdata/minimal-dda.yaml") + require.NoError(t, err) + dda.Spec.Features = preparedRolloutTestFeatures() + + override := &datadoghqv2alpha1.DatadogAgentComponentOverride{HostNetwork: ptr.To(true)} + override.UpdateStrategy = &common.UpdateStrategy{ + Type: string(appsv1.RollingUpdateDaemonSetStrategyType), + RollingUpdate: &common.RollingUpdate{ + MaxUnavailable: ptr.To(intstr.FromInt(1)), + MaxSurge: ptr.To(intstr.FromInt(1)), + }, + } + if prepared { + dda.Annotations = map[string]string{"experimental.agent.datadoghq.com/node-agent-rollout-mode": "prepared-surge-v1"} + } + dda.Spec.Override = map[datadoghqv2alpha1.ComponentName]*datadoghqv2alpha1.DatadogAgentComponentOverride{ + datadoghqv2alpha1.NodeAgentComponentName: override, + } + + objects, _, err := Render(Options{DDA: dda}) + require.NoError(t, err) + for _, object := range objects { + if ds, ok := object.(*appsv1.DaemonSet); ok { + return ds + } + } + t.Fatal("render produced no Agent DaemonSet") + return nil + } + + baseline := renderAgentDaemonSet(t, false) + require.True(t, baseline.Spec.Template.Spec.HostNetwork) + require.NotNil(t, baseline.Spec.UpdateStrategy.RollingUpdate) + assert.Equal(t, intstr.FromInt(1), *baseline.Spec.UpdateStrategy.RollingUpdate.MaxSurge) + assert.Equal(t, intstr.FromInt(1), *baseline.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable, + "ordinary native surge must remain unchanged when prepared rollout is disabled") + baselinePortCount := 0 + for _, container := range baseline.Spec.Template.Spec.Containers { + baselinePortCount += len(container.Ports) + } + require.Positive(t, baselinePortCount, "the host-network baseline must exercise Kubernetes's implicit hostPort defaulting") + + prepared := renderAgentDaemonSet(t, true) + require.True(t, prepared.Spec.Template.Spec.HostNetwork) + require.NotNil(t, prepared.Spec.UpdateStrategy.RollingUpdate) + assert.Equal(t, intstr.FromInt(1), *prepared.Spec.UpdateStrategy.RollingUpdate.MaxSurge) + assert.Equal(t, intstr.FromInt(0), *prepared.Spec.UpdateStrategy.RollingUpdate.MaxUnavailable) + assert.Equal(t, "prepared-surge-v1", prepared.Spec.Template.Annotations["experimental.agent.datadoghq.com/node-agent-rollout-mode"]) + baselineByName := make(map[string]corev1.Container, len(baseline.Spec.Template.Spec.Containers)) + for _, container := range baseline.Spec.Template.Spec.Containers { + baselineByName[container.Name] = container + } + preparedPortCount := 0 + for _, container := range prepared.Spec.Template.Spec.Containers { + preparedPortCount += len(container.Ports) + baselineContainer, found := baselineByName[container.Name] + require.True(t, found) + assert.Equal(t, baselineContainer.StartupProbe, container.StartupProbe) + assert.Equal(t, baselineContainer.LivenessProbe, container.LivenessProbe) + assert.Equal(t, baselineContainer.ReadinessProbe, container.ReadinessProbe) + if container.Name == "trace-agent" { + assert.Equal(t, "trace-agent", container.Command[0], "prepared mode must bypass trace-loader") + } + } + assert.Zero(t, preparedPortCount, "host-network replacements must not claim ports in the scheduler") +} + +func TestRender_PreparedHostNetworkSurgeWithProfiles(t *testing.T) { + dda, err := LoadDDA("testdata/minimal-dda.yaml") + require.NoError(t, err) + dda.Spec.Features = preparedRolloutTestFeatures() + dda.Annotations = map[string]string{"experimental.agent.datadoghq.com/node-agent-rollout-mode": "prepared-surge-v1"} + + surgeOverride := func() *datadoghqv2alpha1.DatadogAgentComponentOverride { + return &datadoghqv2alpha1.DatadogAgentComponentOverride{ + HostNetwork: ptr.To(true), + UpdateStrategy: &common.UpdateStrategy{ + Type: string(appsv1.RollingUpdateDaemonSetStrategyType), + RollingUpdate: &common.RollingUpdate{ + MaxUnavailable: ptr.To(intstr.FromInt(1)), + MaxSurge: ptr.To(intstr.FromInt(1)), + }, + }, + } + } + dda.Spec.Override = map[datadoghqv2alpha1.ComponentName]*datadoghqv2alpha1.DatadogAgentComponentOverride{ + datadoghqv2alpha1.NodeAgentComponentName: surgeOverride(), + } + + daps, err := LoadDAPs([]string{"testdata/linux-profile.yaml", "testdata/gpu-profile.yaml"}) + require.NoError(t, err) + + objects, _, err := Render(Options{DDA: dda, DAPs: daps, ProfileEnabled: true}) + require.NoError(t, err) + daemonSets := 0 + for _, object := range objects { + ds, ok := object.(*appsv1.DaemonSet) + if !ok { + continue + } + daemonSets++ + require.True(t, ds.Spec.Template.Spec.HostNetwork) + assert.Equal(t, "prepared-surge-v1", ds.Spec.Template.Annotations["experimental.agent.datadoghq.com/node-agent-rollout-mode"]) + require.NotNil(t, ds.Spec.Template.Spec.Affinity) + require.NotNil(t, ds.Spec.Template.Spec.Affinity.PodAntiAffinity) + assert.Len(t, ds.Spec.Template.Spec.Affinity.PodAntiAffinity.RequiredDuringSchedulingIgnoredDuringExecution, 2, + "DaemonSet %s must scope overlap to the same DDA and profile", ds.Name) + } + assert.Equal(t, 3, daemonSets) +} + +func preparedRolloutTestFeatures() *datadoghqv2alpha1.DatadogFeatures { + return &datadoghqv2alpha1.DatadogFeatures{ + APM: &datadoghqv2alpha1.APMFeatureConfig{Enabled: ptr.To(true)}, + LiveProcessCollection: &datadoghqv2alpha1.LiveProcessCollectionFeatureConfig{Enabled: ptr.To(false)}, + LiveContainerCollection: &datadoghqv2alpha1.LiveContainerCollectionFeatureConfig{Enabled: ptr.To(false)}, + ProcessDiscovery: &datadoghqv2alpha1.ProcessDiscoveryFeatureConfig{Enabled: ptr.To(false)}, + ServiceDiscovery: &datadoghqv2alpha1.ServiceDiscoveryFeatureConfig{Enabled: ptr.To(false)}, + } +} + // kindSequence extracts the ordered list of "kind: X" values from serialized YAML. func kindSequence(yaml string) []string { var kinds []string diff --git a/internal/controller/testutils/renderer/renderer.go b/internal/controller/testutils/renderer/renderer.go index 6212558b40..d781835ae4 100644 --- a/internal/controller/testutils/renderer/renderer.go +++ b/internal/controller/testutils/renderer/renderer.go @@ -229,6 +229,7 @@ func Render(opts Options) ([]client.Object, *runtime.Scheme, error) { ddaiOpts := datadogagentinternal.ReconcilerOptions{ SupportCilium: opts.SupportCilium, + APIReader: fakeClient, } ddaiReconciler := datadogagentinternal.NewReconciler(ddaiOpts, fakeClient, platformInfo, scheme, recorder, noopForwarder{})