diff --git a/api/shared/nodenetworkconfigurationenactment_types.go b/api/shared/nodenetworkconfigurationenactment_types.go index bed2524b3f..f281667902 100644 --- a/api/shared/nodenetworkconfigurationenactment_types.go +++ b/api/shared/nodenetworkconfigurationenactment_types.go @@ -83,6 +83,13 @@ const ( NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached ConditionReason = "MaxUnavailableLimitReached" NodeNetworkConfigurationEnactmentConditionConfigurationProgressing ConditionReason = "ConfigurationProgressing" NodeNetworkConfigurationEnactmentConditionConfigurationAborted ConditionReason = "ConfigurationAborted" + NodeNetworkConfigurationEnactmentConditionConfigurationInterrupted ConditionReason = "ConfigurationInterrupted" + // NodeNetworkConfigurationEnactmentConditionConfigurationFinalizing marks the + // post-apply phase: the desired state was applied and only releasing the + // maxUnavailable slot and recording success remain. A reconcile that finds + // the enactment in this phase must not re-apply the (already committed) + // configuration. + NodeNetworkConfigurationEnactmentConditionConfigurationFinalizing ConditionReason = "ConfigurationFinalizing" ) func EnactmentKey(node, policy string) types.NamespacedName { diff --git a/cmd/handler/main.go b/cmd/handler/main.go index 4d329a09b9..03ca808e78 100644 --- a/cmd/handler/main.go +++ b/cmd/handler/main.go @@ -28,7 +28,6 @@ import ( "time" corev1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/fields" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" @@ -64,11 +63,13 @@ import ( controllersmetrics "github.com/nmstate/kubernetes-nmstate/controllers/metrics" "github.com/nmstate/kubernetes-nmstate/pkg/cluster" "github.com/nmstate/kubernetes-nmstate/pkg/enactmentstatus" + enactmentconditions "github.com/nmstate/kubernetes-nmstate/pkg/enactmentstatus/conditions" "github.com/nmstate/kubernetes-nmstate/pkg/environment" "github.com/nmstate/kubernetes-nmstate/pkg/file" nmstatelog "github.com/nmstate/kubernetes-nmstate/pkg/log" "github.com/nmstate/kubernetes-nmstate/pkg/monitoring" "github.com/nmstate/kubernetes-nmstate/pkg/nmstatectl" + "github.com/nmstate/kubernetes-nmstate/pkg/node" "github.com/nmstate/kubernetes-nmstate/pkg/webhook" ) @@ -302,15 +303,15 @@ func setupWebhookEnvironment(mgr manager.Manager, tlsOpts func(*tls.Config)) err return nil } -// setupHandlerEnvironment cleans up unavailableNodeCounts after unexpected restart, +// setupHandlerEnvironment reclaims maxUnavailable slots held by this node's +// enactments interrupted by an unexpected restart (reclaimInterruptedSlots), // configures the handler controllers and performs health checks func setupHandlerEnvironment(mgr manager.Manager) error { - // Clean stale unavailable counts from node before starting controllers - // Prevents deadlock after unexpected cluster reboot where nodes were - // processing NNCP and left stale counts in etcd. - if err := cleanStaleUnavailableCounts(mgr); err != nil { - setupLog.Error(err, "Failed to cleanup stale unavailable counts, continuing anyway") - // Don't error this is best-effort (NNCP needs manual restart) + // Reclaim maxUnavailable slots held by enactments interrupted by an + // unexpected handler/cluster restart. Best-effort: the reconciler's + // audit-on-block heals the same state if this fails. + if err := reclaimInterruptedSlots(mgr); err != nil { + setupLog.Error(err, "Failed to reclaim interrupted slots, continuing anyway") } if err := setupHandlerControllers(mgr); err != nil { @@ -333,18 +334,38 @@ func startManager(mgr manager.Manager, ctx context.Context) int { return 0 } -// cleanStaleUnavailableCounts cleans up stale unavailable node counts that have been -// left in NNCP status after unexpected handler restarts or cluster reboots. +// startupListBackoff retries the initial enactment List for ~2 minutes: at +// handler startup after an ungraceful reboot, the apiserver is frequently +// not yet ready, and failing silently here would leave ghost slots in +// place until the audit's staleness threshold elapses. +var startupListBackoff = wait.Backoff{ + Duration: 2 * time.Second, + Factor: 2.0, + Jitter: 0.1, + Steps: 7, + Cap: 60 * time.Second, +} + +// reclaimInterruptedSlots recovers maxUnavailable slots held by this node's +// enactments that were still Progressing when the handler last died. At +// handler startup no applies are in progress for this node, so any own +// enactment still Progressing is provably dead and is recovered by phase: +// +// - Finalizing (ConfigurationFinalizing): the desired state was already +// applied and committed before the crash and the node's networking +// persists across a handler restart, so it is finalized without +// re-applying: record success, then release the slot with the idempotent +// set-to-truth audit. +// - Progressing (ConfigurationProgressing): the apply was interrupted +// mid-flight, so mark it interrupted (Pending, retry count reset) to be +// re-applied and audit the policy's unavailable-node counter. // -// At handler startup, no applies are in progress for this node. Any enactment that -// is NOT in Available=True state may have a stale UnavailableNodeCountMap entry from -// a previous interrupted reconcile. We check !IsAvailable rather than IsProgressing -// because crashes can leave enactments in various non-progressing states (Failing, -// Pending, empty conditions) that still have stale counts. -func cleanStaleUnavailableCounts(mgr manager.Manager) error { +// This is a fast-path optimization: if it fails, the audit-on-block in the +// NNCP reconciler heals the same state within one staleness threshold. +func reclaimInterruptedSlots(mgr manager.Manager) error { ctx := context.Background() nodeName := environment.NodeName() - setupLog.Info("Cleaning up stale unavailable counts for node", "node", nodeName) + setupLog.Info("Reclaiming interrupted unavailable-node slots", "node", nodeName) apiClient, err := client.New(mgr.GetConfig(), client.Options{Scheme: mgr.GetScheme()}) if err != nil { @@ -353,92 +374,59 @@ func cleanStaleUnavailableCounts(mgr manager.Manager) error { enactmentList := &nmstatev1beta1.NodeNetworkConfigurationEnactmentList{} nodeLabel := client.MatchingLabels{nmstateapi.EnactmentNodeLabel: nodeName} - if err := apiClient.List(ctx, enactmentList, nodeLabel); err != nil { + if err := retry.OnError(startupListBackoff, func(error) bool { return true }, func() error { + return apiClient.List(ctx, enactmentList, nodeLabel) + }); err != nil { return err } - // For each enactment that is not Available (may have a stale count from an interrupted apply) for i := range enactmentList.Items { enactment := &enactmentList.Items[i] - if !enactmentstatus.IsAvailable(&enactment.Status.Conditions) { - policyName := enactment.Labels[nmstateapi.EnactmentPolicyLabel] - if policyName == "" { - continue - } - generationKey := strconv.FormatInt(enactment.Status.PolicyGeneration, 10) - - setupLog.Info("detected stale non-available enactment, cleaning up", - "enactment", enactment.Name, - "policy", policyName, - "generation", generationKey) - - // Decrement counter for this policy and generation - if err := decrementStaleUnavailableCount(ctx, apiClient, policyName, generationKey); err != nil { - setupLog.Error(err, "Failed to decrement stale count", "policy", policyName) - // no return to continue with other enactments - } - - // Reset retry count for this enactment and generation - if err := resetStaleRetryCount(ctx, apiClient, enactment.Name, generationKey); err != nil { - setupLog.Error(err, "Failed to reset stale retry count", "enactment", enactment.Name) - // no return to continue with other enactments - } - } - } - - setupLog.Info("Finished cleaning up stale unavailable counts", "node", nodeName) - return nil -} - -// decrementStaleUnavailableCount decrements the UnavailableNodeCountMap of a specific -// policy and generation for startup cleanup. -func decrementStaleUnavailableCount(ctx context.Context, cli client.Client, policyName, generationKey string) error { - return retry.RetryOnConflict(retry.DefaultRetry, func() error { - policy := &nmstatev1.NodeNetworkConfigurationPolicy{} - if err := cli.Get(ctx, types.NamespacedName{Name: policyName}, policy); err != nil { - if apierrors.IsNotFound(err) { - setupLog.Info("Policy not found during stale count cleanup, skipping", "policy", policyName) - return nil - } - return err + if !enactmentstatus.IsProgressing(&enactment.Status.Conditions) { + continue } - - if policy.Status.UnavailableNodeCountMap == nil { - return nil // Nothing to clean up + policyName := enactment.Labels[nmstateapi.EnactmentPolicyLabel] + if policyName == "" { + continue } - - if policy.Status.UnavailableNodeCountMap[generationKey] > 0 { - policy.Status.UnavailableNodeCountMap[generationKey]-- - setupLog.Info("Decremented stale unavailable count", - "policy", policyName, - "generation", generationKey, - "newCount", policy.Status.UnavailableNodeCountMap[generationKey]) - return cli.Status().Update(ctx, policy) + enactmentKey := types.NamespacedName{Name: enactment.Name} + policyKey := types.NamespacedName{Name: policyName} + generationKey := strconv.FormatInt(enactment.Status.PolicyGeneration, 10) + + if enactmentstatus.IsFinalizing(&enactment.Status.Conditions) { + // Apply already committed before the crash: finalize without + // re-applying. Record success (moving it out of the live-holder + // set) then release the slot with the idempotent audit. + setupLog.Info("finalizing post-apply enactment interrupted by restart", + "enactment", enactment.Name, "policy", policyName) + ec := enactmentconditions.New(apiClient, enactmentKey) + if err := ec.NotifySuccess(ctx); err != nil { + setupLog.Error(err, "failed recording success for finalizing enactment", "enactment", enactment.Name) + continue // still Finalizing (a live holder); audit-on-block will heal + } + if _, err := node.AuditUnavailableSlots(ctx, apiClient, apiClient, + policyKey, node.StaleEnactmentThreshold()); err != nil { + setupLog.Error(err, "failed auditing unavailable slots", "policy", policyName) + } + continue } - return nil - }) -} + setupLog.Info("marking interrupted enactment and auditing policy slots", + "enactment", enactment.Name, "policy", policyName, "generation", generationKey) -// resetStaleRetryCount resets the RetryCount for a specific enactment and generation -// during startup clean to prevent stale retry counts from previous interrupted reconciles. -func resetStaleRetryCount(ctx context.Context, cli client.Client, enactmentName, generationKey string) error { - return retry.RetryOnConflict(retry.DefaultRetry, func() error { - enactment := &nmstatev1beta1.NodeNetworkConfigurationEnactment{} - if err := cli.Get(ctx, types.NamespacedName{Name: enactmentName}, enactment); err != nil { - return err + if err := enactmentconditions.MarkInterrupted(ctx, apiClient, enactmentKey, generationKey); err != nil { + setupLog.Error(err, "failed marking enactment interrupted", "enactment", enactment.Name) + continue // audit would still count it live; skip to avoid double-freeing later } - if enactment.Status.RetryCount == nil || enactment.Status.RetryCount[generationKey] == 0 { - return nil + if _, err := node.AuditUnavailableSlots(ctx, apiClient, apiClient, + policyKey, node.StaleEnactmentThreshold()); err != nil { + setupLog.Error(err, "failed auditing unavailable slots", "policy", policyName) } + } - enactment.Status.RetryCount[generationKey] = 0 - setupLog.Info("Reset stale retry count", - "enactment", enactmentName, - "generation", generationKey) - return cli.Status().Update(ctx, enactment) - }) + setupLog.Info("Finished reclaiming interrupted slots", "node", nodeName) + return nil } // Handler runs as a daemonset and we want that each handler pod will cache/reconcile only resources that belong the node it runs on. diff --git a/controllers/handler/nodenetworkconfigurationpolicy_controller.go b/controllers/handler/nodenetworkconfigurationpolicy_controller.go index 1f94e6c997..09af209e10 100644 --- a/controllers/handler/nodenetworkconfigurationpolicy_controller.go +++ b/controllers/handler/nodenetworkconfigurationpolicy_controller.go @@ -97,7 +97,8 @@ var ( return false }, } - nmstatectlShowFn = nmstatectl.Show + nmstatectlShowFn = nmstatectl.Show + applyDesiredStateFn = nmstate.ApplyDesiredState ) // NodeNetworkConfigurationPolicyReconciler reconciles a NodeNetworkConfigurationPolicy object @@ -228,8 +229,37 @@ func (r *NodeNetworkConfigurationPolicyReconciler) Reconcile(ctx context.Context return ctrl.Result{}, nil } - if r.shouldIncrementUnavailableNodeCount(previousConditions) { - err = r.incrementUnavailableNodeCount(ctx, instance, generationKey) + // Already fully reconciled for this generation: a spurious re-reconcile + // (informer re-list, node label change) must not re-claim a slot and + // re-apply an already committed configuration. Release the slot with the + // idempotent audit in case an earlier finalize recorded success but could + // not release it, then finish. + if enactmentstatus.IsAvailable(&enactmentInstance.Status.Conditions) { + log.Info("enactment already Available for current generation, ensuring slot released") + if err := r.releaseUnavailableSlotByAudit(ctx, request.NamespacedName); err != nil { + log.Info("Available enactment: unavailable-node slot release will be retried", + "error", err, "requeueAfter", "10s") + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil + } + return ctrl.Result{}, nil + } + + // Post-apply finalization phase: the desired state was applied in an + // earlier reconcile but the slot release or success write did not complete. + // Only finalize (record success, release the slot); do NOT re-apply the + // already committed configuration. + if enactmentstatus.IsFinalizing(&enactmentInstance.Status.Conditions) { + log.Info("enactment already applied (finalizing); recording success and releasing slot without re-applying") + return r.finalizeInterruptedApply(ctx, request.NamespacedName, enactmentConditions), nil + } + + alreadyHoldsSlot := enactmentstatus.IsProgressing(&enactmentInstance.Status.Conditions) + if alreadyHoldsSlot { + log.Info("enactment already Progressing for current generation; slot held by an interrupted reconcile, skipping claim") + } + didClaim := false + if !alreadyHoldsSlot && r.shouldIncrementUnavailableNodeCount(previousConditions) { + err = r.claimUnavailableSlot(ctx, instance, request.NamespacedName, generationKey) if err != nil { if apierrors.IsConflict(err) || errors.Is(err, node.MaxUnavailableLimitReachedError{}) { enactmentConditions.NotifyPending(ctx) @@ -249,18 +279,47 @@ func (r *NodeNetworkConfigurationPolicyReconciler) Reconcile(ctx context.Context enactmentConditions.NotifyAborted(ctx, fmt.Errorf("reconciliation of enactment %q has aborted", enactmentInstance.Name)) return ctrl.Result{}, nil } + // The maxUnavailable cap refused the claim. Requeue via the + // controller's per-item exponential backoff (InitialBackoff -> + // MaximumTimeBackoff) rather than a fixed multi-minute delay: + // the slot a peer holds is freed by a status-only NNCP update, + // which does not trigger this controller's generation-scoped + // watch, so the blocked node must poll. Backing off at seconds + // (not ~90-120s) keeps rollouts moving on multi-wave clusters + // while the audit-on-block still bounds ghost-slot recovery. return ctrl.Result{Requeue: true}, nil } return ctrl.Result{}, err } + didClaim = true + } + + if err := enactmentConditions.NotifyProgressing(ctx); err != nil { + // Without a persisted Progressing marker the audit on other nodes + // cannot see this node as a live slot holder. Do not apply: release + // the slot claimed in this reconcile (if any) and retry shortly. + // + // Use a bounded release that stays within node.AuditGraceWindow so it + // cannot race the audit: a longer retry could land after the grace + // expired and another node had already audited this markerless claim + // away and taken the slot, double-freeing it. If the bounded release + // fails, the set-to-truth audit reclaims the slot (this enactment is + // not Progressing, so it never counts as a live holder). + if didClaim { + if releaseErr := tryDecrementingUnavailableNodeCount( + ctx, r.Client, r.APIClient, request.NamespacedName, generationKey, compensatingReleaseBackoff, + ); releaseErr != nil { + log.Error(releaseErr, + "failed releasing just-claimed slot after Progressing write failure; audit will reclaim it") + } + } + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } - - enactmentConditions.NotifyProgressing(ctx) if policyconditions.IsUnknown(&instance.Status.Conditions) { policyconditions.Update(ctx, r.Client, r.APIClient, request.NamespacedName) } - nmstateOutput, err := nmstate.ApplyDesiredState(ctx, r.APIClient, enactmentInstance.Status.DesiredState) + nmstateOutput, err := applyDesiredStateFn(ctx, r.APIClient, enactmentInstance.Status.DesiredState) if err != nil { errmsg := fmt.Errorf("error reconciling NodeNetworkConfigurationPolicy on node %s at desired state apply: %q,\n %v", nodeName, nmstateOutput, err) @@ -294,9 +353,42 @@ func (r *NodeNetworkConfigurationPolicyReconciler) Reconcile(ctx context.Context } log.Info("nmstate", "output", nmstateOutput) - enactmentConditions.NotifySuccess(ctx) + // Enter the finalization phase before touching the counter. The enactment + // stays Progressing (a live slot holder) but is marked Finalizing, so if + // the release or success write below fails, the requeue finalizes at the + // top of Reconcile instead of re-applying the already committed + // configuration. The marker write itself gets the authoritative retry + // budget because it runs right after the node reconfigured its own + // networking, when the API server may be briefly unreachable; only if it + // cannot be persisted at all does the reconcile fall back to a (safe, + // idempotent) re-apply on the next pass. + if err := retry.OnError(slotReleaseBackoff, func(error) bool { return true }, func() error { + return enactmentConditions.NotifyFinalizing(ctx) + }); err != nil { + r.Log.Info("Failed to record finalizing phase, will retry", + "error", err, "requeueAfter", "10s") + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil + } + + // Fast-path release: a blind decrement is safe here because the enactment + // is still Progressing (Finalizing reports Progressing=True), so a + // concurrent audit on another node counts it as a live holder and will not + // free its slot. Success is recorded only after the slot is released, so + // the poisonous Available+held-slot state stays unreachable. if err := r.decrementUnavailableNodeCount(ctx, instance, generationKey); err != nil { - r.Log.Info("Failed to update NNCP status, will retry", "error", err, "requeueAfter", "10s") + r.Log.Info("Failed to release unavailable-node slot, will retry", + "error", err, "requeueAfter", "10s") + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil + } + if err := enactmentConditions.NotifySuccess(ctx); err != nil { + // The slot is released, but success was not persisted. Do not swallow + // this: the enactment would stay Finalizing and, because this + // controller watches neither NNCE updates nor status-only NNCP + // updates, nothing would re-trigger reconciliation and the policy + // would stay Progressing forever. Requeue so a later reconcile + // finalizes (records success) without re-applying. + r.Log.Info("Failed to record enactment success, will retry", + "error", err, "requeueAfter", "10s") return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } r.forceNNSRefresh(ctx, nodeName) @@ -304,6 +396,54 @@ func (r *NodeNetworkConfigurationPolicyReconciler) Reconcile(ctx context.Context return ctrl.Result{}, nil } +// finalizeInterruptedApply completes a reconcile whose apply already committed +// (the enactment is in the Finalizing phase) but whose slot release or success +// write did not finish. It never re-applies the desired state. +// +// Unlike the happy path, the release here must be idempotent: a previous +// reconcile may already have decremented the slot before failing to record +// success, and a second blind decrement could drop another node's slot and +// break maxUnavailable. Success is therefore recorded first (moving this +// enactment out of the live-holder set), and the slot is then released with the +// set-to-truth audit, which repairs the counter to the number of live holders +// and can never free a slot another node still holds. +func (r *NodeNetworkConfigurationPolicyReconciler) finalizeInterruptedApply( + ctx context.Context, + policyKey types.NamespacedName, + enactmentConditions enactmentconditions.EnactmentConditions, +) ctrl.Result { + if err := enactmentConditions.NotifySuccess(ctx); err != nil { + r.Log.Info("Failed to record enactment success, will retry", + "error", err, "requeueAfter", "10s") + return ctrl.Result{RequeueAfter: 10 * time.Second} + } + if err := r.releaseUnavailableSlotByAudit(ctx, policyKey); err != nil { + r.Log.Info("Failed to release unavailable-node slot, will retry", + "error", err, "requeueAfter", "10s") + return ctrl.Result{RequeueAfter: 10 * time.Second} + } + r.forceNNSRefresh(ctx, nodeName) + return ctrl.Result{} +} + +// releaseUnavailableSlotByAudit releases a slot idempotently by repairing the +// policy counter down to the number of live Progressing holders. The caller +// must have already moved this enactment out of the live-holder set (it is +// Available), so the audit drops this node's slot. Being set-to-truth it is +// idempotent and never frees a slot another node still holds, so it is safe to +// run repeatedly and concurrently with audits on other nodes. It uses the +// larger slotReleaseBackoff because it runs right after the node reconfigured +// its own networking, when the API server may be briefly unreachable. +func (r *NodeNetworkConfigurationPolicyReconciler) releaseUnavailableSlotByAudit( + ctx context.Context, + policyKey types.NamespacedName, +) error { + return retry.OnError(slotReleaseBackoff, func(error) bool { return true }, func() error { + _, err := node.AuditUnavailableSlots(ctx, r.Client, r.APIClient, policyKey, node.StaleEnactmentThreshold()) + return err + }) +} + func (r *NodeNetworkConfigurationPolicyReconciler) incrementNNCERetryCount( ctx context.Context, instance *nmstatev1.NodeNetworkConfigurationPolicy, @@ -574,6 +714,33 @@ func (r *NodeNetworkConfigurationPolicyReconciler) shouldIncrementUnavailableNod return shouldIncrement } +// claimUnavailableSlot increments the policy's unavailable-node counter. If +// the counter is at the maxUnavailable cap, it audits the counter against +// live Progressing enactments (repairing ghost slots left by interrupted +// applies) and retries the increment once. +func (r *NodeNetworkConfigurationPolicyReconciler) claimUnavailableSlot( + ctx context.Context, + policy *nmstatev1.NodeNetworkConfigurationPolicy, + policyKey types.NamespacedName, + generationKey string, +) error { + err := r.incrementUnavailableNodeCount(ctx, policy, generationKey) + if err == nil || !errors.Is(err, node.MaxUnavailableLimitReachedError{}) { + return err + } + repaired, auditErr := node.AuditUnavailableSlots( + ctx, r.Client, r.APIClient, policyKey, node.StaleEnactmentThreshold()) + if auditErr != nil { + r.Log.Error(auditErr, "unavailable-slot audit failed", "policy", policyKey.Name) + return err + } + if !repaired { + return err + } + r.Log.Info("unavailable-slot audit repaired ghost slots, retrying claim", "policy", policyKey.Name) + return r.incrementUnavailableNodeCount(ctx, policy, generationKey) +} + func (r *NodeNetworkConfigurationPolicyReconciler) incrementUnavailableNodeCount( ctx context.Context, policy *nmstatev1.NodeNetworkConfigurationPolicy, @@ -598,19 +765,55 @@ func (r *NodeNetworkConfigurationPolicyReconciler) incrementUnavailableNodeCount return node.MaxUnavailableLimitReachedError{} } policy.Status.UnavailableNodeCountMap[generationKey] += 1 + now := metav1.Now() + policy.Status.LastUnavailableNodeCountUpdate = &now return r.Client.Status().Update(ctx, policy) }) } +// slotReleaseBackoff is the retry budget for the authoritative (non-cached) +// unavailable-slot release attempt. It runs right after the node's own +// networking was reconfigured, so it deserves a much larger budget than the +// cached fast-path. +// +// wait.Backoff.Steps counts attempts, not sleeps: ExponentialBackoff sleeps +// between attempts but not after the last one, so N steps yield N-1 sleeps. +// Seven steps therefore sleep 0.5+1+2+4+8+16 = ~31.5s cumulative across six +// waits. The enactment is still Progressing (a live slot holder) throughout +// this release, so a concurrent audit on another node counts it as live and +// will not free its slot, keeping the release safe even past the audit grace. +var slotReleaseBackoff = wait.Backoff{ + Duration: 500 * time.Millisecond, + Factor: 2.0, + Jitter: 0.1, + Steps: 7, // 6 sleeps: 0.5+1+2+4+8+16 = ~31.5s cumulative +} + +// compensatingReleaseBackoff bounds the best-effort release of a slot that was +// claimed in this reconcile but whose Progressing marker could not be +// persisted. Unlike the success-path release, this one MUST finish within +// node.AuditGraceWindow: the claim just stamped LastUnavailableNodeCountUpdate, +// so audits on other nodes defer for that window. Releasing inside it +// guarantees no other node can audit the (markerless) claim away and take the +// slot before this release lands, which would turn a late decrement into a +// double-free. If it still fails, the enactment is not Progressing, so the +// set-to-truth audit will reclaim the slot safely. +var compensatingReleaseBackoff = wait.Backoff{ + Duration: 500 * time.Millisecond, + Factor: 2.0, + Jitter: 0.1, + Steps: 5, // 4 sleeps: 0.5+1+2+4 = ~7.5s cumulative, well under AuditGraceWindow (30s) +} + func (r *NodeNetworkConfigurationPolicyReconciler) decrementUnavailableNodeCount( ctx context.Context, policy *nmstatev1.NodeNetworkConfigurationPolicy, generationKey string) error { policyKey := types.NamespacedName{Name: policy.GetName(), Namespace: policy.GetNamespace()} - err := tryDecrementingUnavailableNodeCount(ctx, r.Client, r.Client, policyKey, generationKey) + err := tryDecrementingUnavailableNodeCount(ctx, r.Client, r.Client, policyKey, generationKey, retry.DefaultRetry) if err != nil { - r.Log.Error(err, "error decrementing unavailableNodeCount with cached client, trying again with non-cached client.") - err = tryDecrementingUnavailableNodeCount(ctx, r.Client, r.APIClient, policyKey, generationKey) + r.Log.Error(err, "error decrementing unavailableNodeCount with cached client, retrying with non-cached client and larger budget.") + err = tryDecrementingUnavailableNodeCount(ctx, r.Client, r.APIClient, policyKey, generationKey, slotReleaseBackoff) if err != nil { r.Log.Error(err, "error decrementing unavailableNodeCount with non-cached client") return err @@ -624,9 +827,10 @@ func tryDecrementingUnavailableNodeCount( statusWriterClient client.StatusClient, readerClient client.Reader, policyKey types.NamespacedName, - generationKey string) error { + generationKey string, + backoff wait.Backoff) error { instance := &nmstatev1.NodeNetworkConfigurationPolicy{} - err := retry.OnError(retry.DefaultRetry, func(error) bool { return true }, func() error { + err := retry.OnError(backoff, func(error) bool { return true }, func() error { err := readerClient.Get(ctx, policyKey, instance) if err != nil { return err @@ -638,6 +842,8 @@ func tryDecrementingUnavailableNodeCount( return nil } instance.Status.UnavailableNodeCountMap[generationKey] -= 1 + now := metav1.Now() + instance.Status.LastUnavailableNodeCountUpdate = &now return statusWriterClient.Status().Update(ctx, instance) }) return err diff --git a/controllers/handler/nodenetworkconfigurationpolicy_controller_test.go b/controllers/handler/nodenetworkconfigurationpolicy_controller_test.go index 1d831518b6..c7c6dcc4e6 100644 --- a/controllers/handler/nodenetworkconfigurationpolicy_controller_test.go +++ b/controllers/handler/nodenetworkconfigurationpolicy_controller_test.go @@ -24,20 +24,468 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" 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/types" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes/scheme" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" "sigs.k8s.io/controller-runtime/pkg/event" "github.com/nmstate/kubernetes-nmstate/api/shared" nmstatev1 "github.com/nmstate/kubernetes-nmstate/api/v1" nmstatev1beta1 "github.com/nmstate/kubernetes-nmstate/api/v1beta1" + nmstate "github.com/nmstate/kubernetes-nmstate/pkg/client" + "github.com/nmstate/kubernetes-nmstate/pkg/enactmentstatus" "github.com/nmstate/kubernetes-nmstate/pkg/enactmentstatus/conditions" + "github.com/nmstate/kubernetes-nmstate/pkg/node" ) +// stubApplyOutput is the canned nmstate output returned by applyDesiredStateFn +// stubs in these specs. +const stubApplyOutput = "ok" + +var _ = Describe("success path slot release ordering", func() { + // buildSlotReleaseTestClient builds a reconciler + fake client where NNCP + // status writes that release the maxUnavailable slot (count 1 -> 0) fail + // while *failNNCPStatusWrites is true. + buildSlotReleaseTestClient := func(failNNCPStatusWrites *bool) ( + *NodeNetworkConfigurationPolicyReconciler, client.Client, types.NamespacedName, + ) { + originalNmstatectlShowFn := nmstatectlShowFn + nmstatectlShowFn = func() (string, error) { return "", nil } + DeferCleanup(func() { nmstatectlShowFn = originalNmstatectlShowFn }) + reconciler := &NodeNetworkConfigurationPolicyReconciler{ + RetriesUntilFail: 5, + MaximumTimeBackoff: 30 * time.Second, + InitialBackoff: 1 * time.Second, + } + s := scheme.Scheme + s.AddKnownTypes(nmstatev1beta1.GroupVersion, + &nmstatev1beta1.NodeNetworkState{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactment{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactmentList{}) + s.AddKnownTypes(nmstatev1.GroupVersion, + &nmstatev1.NodeNetworkConfigurationPolicy{}) + + node := corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + nns := nmstatev1beta1.NodeNetworkState{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + nncp := nmstatev1.NodeNetworkConfigurationPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Status: shared.NodeNetworkConfigurationPolicyStatus{ + UnavailableNodeCountMap: map[string]int{}, + }, + } + nnce := nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{ + Name: shared.EnactmentKey(nodeName, nncp.Name).Name, + Labels: map[string]string{shared.EnactmentPolicyLabel: nncp.Name}, + }, + } + + sawSlotClaimed := false + clb := fake.ClientBuilder{} + clb.WithScheme(s) + clb.WithRuntimeObjects(&nncp, &nnce, &nns, &node) + clb.WithStatusSubresource(&nncp, &nnce, &nns) + clb.WithInterceptorFuncs(interceptor.Funcs{ + SubResourceUpdate: func( + ctx context.Context, cl client.Client, subResourceName string, + obj client.Object, opts ...client.SubResourceUpdateOption, + ) error { + if updated, isPolicy := obj.(*nmstatev1.NodeNetworkConfigurationPolicy); isPolicy { + if updated.Status.UnavailableNodeCountMap["0"] >= 1 { + // The increment (slot claim): let it through, remember it. + sawSlotClaimed = true + } else if *failNNCPStatusWrites && sawSlotClaimed { + // The decrement (slot release, 1 -> 0): fail it. + return apierrors.NewInternalError(context.DeadlineExceeded) + } + } + return cl.SubResource(subResourceName).Update(ctx, obj, opts...) + }, + }) + cl := clb.Build() + reconciler.Client = cl + reconciler.APIClient = cl + reconciler.Log = ctrl.Log.WithName("test") + return reconciler, cl, types.NamespacedName{Name: nncp.Name} + } + + It("keeps the enactment Progressing when the slot release fails", func() { + applyDesiredStateFn = func(context.Context, client.Client, shared.State) (string, error) { return stubApplyOutput, nil } + defer func() { applyDesiredStateFn = nmstate.ApplyDesiredState }() + originalSlotReleaseBackoff := slotReleaseBackoff + slotReleaseBackoff = wait.Backoff{Duration: 1 * time.Millisecond, Steps: 1} + defer func() { slotReleaseBackoff = originalSlotReleaseBackoff }() + + failNNCPStatusWrites := true + reconciler, cl, policyKey := buildSlotReleaseTestClient(&failNNCPStatusWrites) + + res, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + Expect(res.RequeueAfter).To(Equal(10 * time.Second)) + + // The slot is still held and the enactment must NOT claim success. + nnceKey := shared.EnactmentKey(nodeName, policyKey.Name) + updatedNNCE := &nmstatev1beta1.NodeNetworkConfigurationEnactment{} + Expect(cl.Get(context.TODO(), nnceKey, updatedNNCE)).To(Succeed()) + Expect(enactmentstatus.IsAvailable(&updatedNNCE.Status.Conditions)).To(BeFalse(), + "enactment must not be Available while the slot is still held") + Expect(enactmentstatus.IsProgressing(&updatedNNCE.Status.Conditions)).To(BeTrue()) + }) + + // Note: buildSlotReleaseTestClient's interceptor couples sawSlotClaimed with + // *failNNCPStatusWrites so only the release (decrement) write fails, never the claim. + It("converges once NNCP status writes heal", func() { + applyDesiredStateFn = func(context.Context, client.Client, shared.State) (string, error) { return stubApplyOutput, nil } + defer func() { applyDesiredStateFn = nmstate.ApplyDesiredState }() + originalSlotReleaseBackoff := slotReleaseBackoff + slotReleaseBackoff = wait.Backoff{Duration: 1 * time.Millisecond, Steps: 1} + defer func() { slotReleaseBackoff = originalSlotReleaseBackoff }() + + failNNCPStatusWrites := true + reconciler, cl, policyKey := buildSlotReleaseTestClient(&failNNCPStatusWrites) + + _, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + + // The enactment is now parked in the Finalizing phase; its slot is + // released on the retry by the set-to-truth audit. In production the + // failed decrement above burned the ~31.5s slotReleaseBackoff, so by + // the retry the claim's LastUnavailableNodeCountUpdate is well past the + // audit grace window; simulate that elapsed time (the spec shrinks the + // backoff for speed) so the audit does not defer. + parked := &nmstatev1beta1.NodeNetworkConfigurationEnactment{} + Expect(cl.Get(context.TODO(), shared.EnactmentKey(nodeName, policyKey.Name), parked)).To(Succeed()) + Expect(enactmentstatus.IsFinalizing(&parked.Status.Conditions)).To(BeTrue(), + "the first reconcile must park the enactment in the Finalizing phase") + blocked := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, blocked)).To(Succeed()) + aged := metav1.NewTime(time.Now().Add(-2 * node.AuditGraceWindow)) + blocked.Status.LastUnavailableNodeCountUpdate = &aged + Expect(cl.Status().Update(context.TODO(), blocked)).To(Succeed()) + + // Heal: allow NNCP status writes again, reconcile converges. + failNNCPStatusWrites = false + _, err = reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + + nnceKey := shared.EnactmentKey(nodeName, policyKey.Name) + updatedNNCE := &nmstatev1beta1.NodeNetworkConfigurationEnactment{} + Expect(cl.Get(context.TODO(), nnceKey, updatedNNCE)).To(Succeed()) + Expect(enactmentstatus.IsAvailable(&updatedNNCE.Status.Conditions)).To(BeTrue()) + + updatedNNCP := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, updatedNNCP)).To(Succeed()) + Expect(updatedNNCP.Status.UnavailableNodeCountMap["0"]).To(Equal(0)) + }) +}) + +var _ = Describe("Progressing write failure after slot claim", func() { + It("releases the just-claimed slot and requeues after 10s without applying", func() { + originalNmstatectlShowFn := nmstatectlShowFn + nmstatectlShowFn = func() (string, error) { return "", nil } + defer func() { nmstatectlShowFn = originalNmstatectlShowFn }() + applyCalled := false + applyDesiredStateFn = func(context.Context, client.Client, shared.State) (string, error) { + applyCalled = true + return stubApplyOutput, nil + } + defer func() { applyDesiredStateFn = nmstate.ApplyDesiredState }() + + reconciler := &NodeNetworkConfigurationPolicyReconciler{ + RetriesUntilFail: 5, + MaximumTimeBackoff: 30 * time.Second, + InitialBackoff: 1 * time.Second, + } + s := scheme.Scheme + s.AddKnownTypes(nmstatev1beta1.GroupVersion, + &nmstatev1beta1.NodeNetworkState{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactment{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactmentList{}) + s.AddKnownTypes(nmstatev1.GroupVersion, + &nmstatev1.NodeNetworkConfigurationPolicy{}) + + node := corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + nns := nmstatev1beta1.NodeNetworkState{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + nncp := nmstatev1.NodeNetworkConfigurationPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Status: shared.NodeNetworkConfigurationPolicyStatus{ + UnavailableNodeCountMap: map[string]int{}, + }, + } + nnce := nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{ + Name: shared.EnactmentKey(nodeName, nncp.Name).Name, + Labels: map[string]string{shared.EnactmentPolicyLabel: nncp.Name}, + }, + } + + sawSlotClaimed := false + clb := fake.ClientBuilder{} + clb.WithScheme(s) + clb.WithRuntimeObjects(&nncp, &nnce, &nns, &node) + clb.WithStatusSubresource(&nncp, &nnce, &nns) + clb.WithInterceptorFuncs(interceptor.Funcs{ + SubResourceUpdate: func( + ctx context.Context, cl client.Client, subResourceName string, + obj client.Object, opts ...client.SubResourceUpdateOption, + ) error { + if updated, isPolicy := obj.(*nmstatev1.NodeNetworkConfigurationPolicy); isPolicy { + if updated.Status.UnavailableNodeCountMap["0"] >= 1 { + // The increment (slot claim): let it through, remember it. + sawSlotClaimed = true + } + } + if updatedNNCE, isNNCE := obj.(*nmstatev1beta1.NodeNetworkConfigurationEnactment); isNNCE { + if sawSlotClaimed && enactmentstatus.IsProgressing(&updatedNNCE.Status.Conditions) { + // The Progressing marker write after a successful claim: fail it. + return apierrors.NewInternalError(context.DeadlineExceeded) + } + } + return cl.SubResource(subResourceName).Update(ctx, obj, opts...) + }, + }) + cl := clb.Build() + reconciler.Client = cl + reconciler.APIClient = cl + reconciler.Log = ctrl.Log.WithName("test") + policyKey := types.NamespacedName{Name: nncp.Name} + + res, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + Expect(sawSlotClaimed).To(BeTrue(), "the slot claim must have happened before the Progressing write") + Expect(res.RequeueAfter).To(Equal(10*time.Second), + "a failed Progressing write after a claim must requeue after 10s") + Expect(applyCalled).To(BeFalse(), + "apply must not run when the Progressing marker could not be persisted") + + updatedNNCP := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, updatedNNCP)).To(Succeed()) + Expect(updatedNNCP.Status.UnavailableNodeCountMap["0"]).To(Equal(0), + "the just-claimed slot must be released when the Progressing write fails") + }) +}) + +var _ = Describe("success write failure after apply (finalization phase)", func() { + // buildFinalizeTestClient builds a reconciler + fake client where NNCE + // status writes that record success (Available=True) fail while + // *failSuccessWrites is true. Progressing/Finalizing NNCE writes and all + // NNCP writes are allowed. + buildFinalizeTestClient := func(failSuccessWrites *bool, applyCalled *bool) ( + *NodeNetworkConfigurationPolicyReconciler, client.Client, types.NamespacedName, + ) { + originalNmstatectlShowFn := nmstatectlShowFn + nmstatectlShowFn = func() (string, error) { return "", nil } + DeferCleanup(func() { nmstatectlShowFn = originalNmstatectlShowFn }) + originalApply := applyDesiredStateFn + applyDesiredStateFn = func(context.Context, client.Client, shared.State) (string, error) { + *applyCalled = true + return stubApplyOutput, nil + } + DeferCleanup(func() { applyDesiredStateFn = originalApply }) + + reconciler := &NodeNetworkConfigurationPolicyReconciler{ + RetriesUntilFail: 5, + MaximumTimeBackoff: 30 * time.Second, + InitialBackoff: 1 * time.Second, + } + s := scheme.Scheme + s.AddKnownTypes(nmstatev1beta1.GroupVersion, + &nmstatev1beta1.NodeNetworkState{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactment{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactmentList{}) + s.AddKnownTypes(nmstatev1.GroupVersion, + &nmstatev1.NodeNetworkConfigurationPolicy{}) + + node := corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + nns := nmstatev1beta1.NodeNetworkState{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + nncp := nmstatev1.NodeNetworkConfigurationPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Status: shared.NodeNetworkConfigurationPolicyStatus{ + UnavailableNodeCountMap: map[string]int{}, + }, + } + nnce := nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{ + Name: shared.EnactmentKey(nodeName, nncp.Name).Name, + Labels: map[string]string{shared.EnactmentPolicyLabel: nncp.Name}, + }, + } + + clb := fake.ClientBuilder{} + clb.WithScheme(s) + clb.WithRuntimeObjects(&nncp, &nnce, &nns, &node) + clb.WithStatusSubresource(&nncp, &nnce, &nns) + clb.WithInterceptorFuncs(interceptor.Funcs{ + SubResourceUpdate: func( + ctx context.Context, cl client.Client, subResourceName string, + obj client.Object, opts ...client.SubResourceUpdateOption, + ) error { + if nnceObj, isNNCE := obj.(*nmstatev1beta1.NodeNetworkConfigurationEnactment); isNNCE { + if *failSuccessWrites && enactmentstatus.IsAvailable(&nnceObj.Status.Conditions) { + return apierrors.NewInternalError(context.DeadlineExceeded) + } + } + return cl.SubResource(subResourceName).Update(ctx, obj, opts...) + }, + }) + cl := clb.Build() + reconciler.Client = cl + reconciler.APIClient = cl + reconciler.Log = ctrl.Log.WithName("test") + return reconciler, cl, types.NamespacedName{Name: nncp.Name} + } + + It("requeues on success-write failure, then finalizes without re-applying", func() { + failSuccessWrites := true + applyCalled := false + reconciler, cl, policyKey := buildFinalizeTestClient(&failSuccessWrites, &applyCalled) + + // Reconcile 1: applies, releases the slot, but cannot record success. + res, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + Expect(applyCalled).To(BeTrue(), "the first reconcile must apply the desired state") + Expect(res.RequeueAfter).To(Equal(10*time.Second), + "a failed success write must requeue rather than be swallowed") + + nnceKey := shared.EnactmentKey(nodeName, policyKey.Name) + nnce := &nmstatev1beta1.NodeNetworkConfigurationEnactment{} + Expect(cl.Get(context.TODO(), nnceKey, nnce)).To(Succeed()) + Expect(enactmentstatus.IsAvailable(&nnce.Status.Conditions)).To(BeFalse()) + Expect(enactmentstatus.IsFinalizing(&nnce.Status.Conditions)).To(BeTrue(), + "the enactment must be parked in the finalizing phase") + + // Heal the success write and reconcile again. The retry must NOT + // re-apply the already committed configuration; it only finalizes. + failSuccessWrites = false + applyCalled = false + res, err = reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + Expect(res).To(Equal(ctrl.Result{})) + Expect(applyCalled).To(BeFalse(), + "the finalization retry must not re-apply the desired state") + + Expect(cl.Get(context.TODO(), nnceKey, nnce)).To(Succeed()) + Expect(enactmentstatus.IsAvailable(&nnce.Status.Conditions)).To(BeTrue()) + + nncp := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, nncp)).To(Succeed()) + Expect(nncp.Status.UnavailableNodeCountMap["0"]).To(Equal(0), + "the slot must be released once finalization completes") + }) + + It("does not double-free another node's slot when success write fails after decrement", func() { + // Regression for the finalization double-decrement: this node applies, + // decrements its own slot, but fails to record success; a second node + // still legitimately holds a slot. The finalization retry must not + // blindly decrement again and release the other node's slot. + originalNmstatectlShowFn := nmstatectlShowFn + nmstatectlShowFn = func() (string, error) { return "", nil } + defer func() { nmstatectlShowFn = originalNmstatectlShowFn }() + applyCalled := false + applyDesiredStateFn = func(context.Context, client.Client, shared.State) (string, error) { + applyCalled = true + return stubApplyOutput, nil + } + defer func() { applyDesiredStateFn = nmstate.ApplyDesiredState }() + + reconciler := &NodeNetworkConfigurationPolicyReconciler{ + RetriesUntilFail: 5, MaximumTimeBackoff: 30 * time.Second, InitialBackoff: 1 * time.Second, + } + s := scheme.Scheme + s.AddKnownTypes(nmstatev1beta1.GroupVersion, + &nmstatev1beta1.NodeNetworkState{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactment{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactmentList{}) + s.AddKnownTypes(nmstatev1.GroupVersion, &nmstatev1.NodeNetworkConfigurationPolicy{}) + + maxUnavailable := intstr.FromInt(2) + node1 := corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + nns := nmstatev1beta1.NodeNetworkState{ObjectMeta: metav1.ObjectMeta{Name: nodeName}} + oldStamp := metav1.NewTime(time.Now().Add(-2 * node.AuditGraceWindow)) + nncp := nmstatev1.NodeNetworkConfigurationPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Spec: shared.NodeNetworkConfigurationPolicySpec{MaxUnavailable: &maxUnavailable}, + Status: shared.NodeNetworkConfigurationPolicyStatus{ + UnavailableNodeCountMap: map[string]int{"0": 1}, // node02 already holds a slot + LastUnavailableNodeCountUpdate: &oldStamp, + }, + } + nnce := nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{ + Name: shared.EnactmentKey(nodeName, nncp.Name).Name, + Labels: map[string]string{shared.EnactmentPolicyLabel: nncp.Name}, + }, + } + // node02 is a live Progressing holder of the current generation. + otherNNCE := nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{ + Name: shared.EnactmentKey("node02", nncp.Name).Name, + Labels: map[string]string{shared.EnactmentPolicyLabel: nncp.Name}, + }, + Status: shared.NodeNetworkConfigurationEnactmentStatus{PolicyGeneration: nncp.Generation}, + } + conditions.SetProgressing(&otherNNCE.Status.Conditions, "applying") + + failSuccessWrites := true + clb := fake.ClientBuilder{} + clb.WithScheme(s) + clb.WithRuntimeObjects(&nncp, &nnce, &otherNNCE, &nns, &node1) + clb.WithStatusSubresource(&nncp, &nnce, &otherNNCE, &nns) + clb.WithInterceptorFuncs(interceptor.Funcs{ + SubResourceUpdate: func( + ctx context.Context, cl client.Client, subResourceName string, + obj client.Object, opts ...client.SubResourceUpdateOption, + ) error { + if nnceObj, isNNCE := obj.(*nmstatev1beta1.NodeNetworkConfigurationEnactment); isNNCE { + if failSuccessWrites && nnceObj.Name == nnce.Name && + enactmentstatus.IsAvailable(&nnceObj.Status.Conditions) { + return apierrors.NewInternalError(context.DeadlineExceeded) + } + } + return cl.SubResource(subResourceName).Update(ctx, obj, opts...) + }, + }) + cl := clb.Build() + reconciler.Client = cl + reconciler.APIClient = cl + reconciler.Log = ctrl.Log.WithName("test") + policyKey := types.NamespacedName{Name: nncp.Name} + + // Reconcile 1: claim (1->2), apply, decrement (2->1), success write fails. + res, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + Expect(applyCalled).To(BeTrue()) + Expect(res.RequeueAfter).To(Equal(10 * time.Second)) + afterFirst := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, afterFirst)).To(Succeed()) + Expect(afterFirst.Status.UnavailableNodeCountMap["0"]).To(Equal(1), + "this node's slot is released; node02's slot remains") + + // Reconcile 2: heal, finalize. Must NOT decrement node02's slot away. + failSuccessWrites = false + applyCalled = false + _, err = reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: policyKey}) + Expect(err).To(BeNil()) + Expect(applyCalled).To(BeFalse(), "finalization retry must not re-apply") + + finalNNCP := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, finalNNCP)).To(Succeed()) + Expect(finalNNCP.Status.UnavailableNodeCountMap["0"]).To(Equal(1), + "node02's slot must be preserved; a blind re-decrement would break maxUnavailable") + updatedNNCE := &nmstatev1beta1.NodeNetworkConfigurationEnactment{} + Expect(cl.Get(context.TODO(), shared.EnactmentKey(nodeName, nncp.Name), updatedNNCE)).To(Succeed()) + Expect(enactmentstatus.IsAvailable(&updatedNNCE.Status.Conditions)).To(BeTrue()) + }) +}) + var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() { type predicateCase struct { GenerationOld int64 @@ -89,12 +537,22 @@ var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() type incrementUnavailableNodeCountCase struct { currentUnavailableNodeCount int - expectedReconcileResult ctrl.Result + lastCountUpdateAge time.Duration // 0 = nil timestamp + otherNodeLiveHolder bool // add a fresh Progressing enactment owned by another node previousEnactmentConditions func(*shared.ConditionList, string) + expectBlocked bool + expectedReconcileResult ctrl.Result // only checked when !expectBlocked } DescribeTable("when claimNodeRunningUpdate is called and", func(c incrementUnavailableNodeCountCase) { + originalNmstatectlShowFn := nmstatectlShowFn nmstatectlShowFn = func() (string, error) { return "", nil } + defer func() { nmstatectlShowFn = originalNmstatectlShowFn }() + // "Proceeds" entries must complete the claim + apply path + // deterministically, so stub the apply to succeed; blocked entries + // never reach the apply, the stub is irrelevant there. + applyDesiredStateFn = func(context.Context, client.Client, shared.State) (string, error) { return stubApplyOutput, nil } + defer func() { applyDesiredStateFn = nmstate.ApplyDesiredState }() reconciler := NodeNetworkConfigurationPolicyReconciler{ RetriesUntilFail: 5, MaximumTimeBackoff: 30 * time.Second, @@ -127,12 +585,18 @@ var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() Name: "test", }, Status: shared.NodeNetworkConfigurationPolicyStatus{ - UnavailableNodeCountMap: map[string]int{}, + UnavailableNodeCountMap: map[string]int{ + "0": c.currentUnavailableNodeCount, // policy generation is 0 + }, }, } + if c.lastCountUpdateAge > 0 { + nncp.Status.LastUnavailableNodeCountUpdate = &metav1.Time{Time: time.Now().Add(-c.lastCountUpdateAge)} + } nnce := nmstatev1beta1.NodeNetworkConfigurationEnactment{ ObjectMeta: metav1.ObjectMeta{ - Name: shared.EnactmentKey(nodeName, nncp.Name).Name, + Name: shared.EnactmentKey(nodeName, nncp.Name).Name, + Labels: map[string]string{shared.EnactmentPolicyLabel: nncp.Name}, }, Status: shared.NodeNetworkConfigurationEnactmentStatus{}, } @@ -141,6 +605,21 @@ var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() c.previousEnactmentConditions(&nnce.Status.Conditions, "") objs := []runtime.Object{&nncp, &nnce, &nns, &node} + if c.otherNodeLiveHolder { + otherNNCE := nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{ + Name: shared.EnactmentKey("node02", nncp.Name).Name, + Labels: map[string]string{shared.EnactmentPolicyLabel: nncp.Name}, + }, + Status: shared.NodeNetworkConfigurationEnactmentStatus{ + PolicyGeneration: nncp.Generation, + }, + } + // SetProgressing writes the full condition set (shouldAbortReconcile + // requires it) with a fresh heartbeat, making node02 a live holder. + conditions.SetProgressing(&otherNNCE.Status.Conditions, "applying") + objs = append(objs, &otherNNCE) + } // Create a fake client to mock API calls. clb := fake.ClientBuilder{} @@ -160,44 +639,65 @@ var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() }) Expect(err).To(BeNil()) - Expect(res).To(Equal(c.expectedReconcileResult)) + if c.expectBlocked { + // A refused claim requeues via the controller's per-item + // exponential backoff (Requeue: true), not a fixed delay, so + // the freed slot is retried within seconds on multi-wave + // clusters instead of after ~90-120s. + Expect(res).To(Equal(ctrl.Result{Requeue: true})) + } else { + Expect(res).To(Equal(c.expectedReconcileResult)) + } }, - Entry("No node applying policy with empty enactment, should succeed incrementing UnavailableNodeCount", + // "proceeds" entries expect ctrl.Result{}: with applyDesiredStateFn + // stubbed to succeed, a successful claim runs apply + slot release + + // NotifySuccess and the reconcile completes without requeueing. + Entry("count 0, empty enactment -> claims slot and proceeds", incrementUnavailableNodeCountCase{ currentUnavailableNodeCount: 0, previousEnactmentConditions: func(*shared.ConditionList, string) {}, - expectedReconcileResult: ctrl.Result{Requeue: true}, + expectBlocked: false, + expectedReconcileResult: ctrl.Result{}, }), - Entry("No node applying policy with progressing enactment, should succeed incrementing UnavailableNodeCount", + Entry("count at cap with fresh live holder on another node -> blocked, requeued via backoff", incrementUnavailableNodeCountCase{ - currentUnavailableNodeCount: 0, - previousEnactmentConditions: conditions.SetProgressing, - expectedReconcileResult: ctrl.Result{Requeue: true}, + currentUnavailableNodeCount: 1, + lastCountUpdateAge: 5 * time.Minute, + otherNodeLiveHolder: true, + previousEnactmentConditions: func(*shared.ConditionList, string) {}, + expectBlocked: true, }), - Entry("No node applying policy with Pending enactment, should succeed incrementing UnavailableNodeCount", + Entry("count at cap, no live holder, stale timestamp -> audit repairs and proceeds", incrementUnavailableNodeCountCase{ - currentUnavailableNodeCount: 0, - previousEnactmentConditions: conditions.SetPending, - expectedReconcileResult: ctrl.Result{Requeue: true}, + currentUnavailableNodeCount: 1, + lastCountUpdateAge: 5 * time.Minute, + previousEnactmentConditions: func(*shared.ConditionList, string) {}, + expectBlocked: false, + expectedReconcileResult: ctrl.Result{}, }), - Entry("One node applying policy with empty enactment, should conflict incrementing UnavailableNodeCount", + Entry("count at cap, no live holder, fresh timestamp -> grace defers, blocked", incrementUnavailableNodeCountCase{ currentUnavailableNodeCount: 1, + lastCountUpdateAge: 5 * time.Second, previousEnactmentConditions: func(*shared.ConditionList, string) {}, - expectedReconcileResult: ctrl.Result{Requeue: true}, + expectBlocked: true, }), - Entry("One node applying policy with Progressing enactment, should conflict incrementing UnavailableNodeCount", + Entry("own enactment Progressing (interrupted apply) -> already holds slot, proceeds without increment", incrementUnavailableNodeCountCase{ currentUnavailableNodeCount: 1, + lastCountUpdateAge: 5 * time.Second, // grace would block; guard must bypass previousEnactmentConditions: conditions.SetProgressing, - expectedReconcileResult: ctrl.Result{Requeue: true}, + expectBlocked: false, + expectedReconcileResult: ctrl.Result{}, }), - Entry("One node applying policy with Pending enactment, should conflict incrementing UnavailableNodeCount", + Entry("own enactment Pending at cap with live holder -> stays blocked", incrementUnavailableNodeCountCase{ currentUnavailableNodeCount: 1, + lastCountUpdateAge: 5 * time.Minute, + otherNodeLiveHolder: true, previousEnactmentConditions: conditions.SetPending, - expectedReconcileResult: ctrl.Result{Requeue: true}, + expectBlocked: true, }), ) @@ -356,10 +856,31 @@ var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() Expect(err).To(BeNil()) Expect(updatedNNCP.Status.UnavailableNodeCountMap["gen-1"]).To(Equal(1)) }) + + It("stamps LastUnavailableNodeCountUpdate on decrement", func() { + clb := fake.ClientBuilder{} + clb.WithScheme(s) + clb.WithRuntimeObjects(nncp) + clb.WithStatusSubresource(nncp) + cl := clb.Build() + reconciler.Client = cl + reconciler.APIClient = cl + + Expect(reconciler.decrementUnavailableNodeCount(context.TODO(), nncp, "gen-1")).To(Succeed()) + + updated := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), types.NamespacedName{Name: "test-policy"}, updated)).To(Succeed()) + Expect(updated.Status.LastUnavailableNodeCountUpdate).ToNot(BeNil()) + Expect(updated.Status.LastUnavailableNodeCountUpdate.Time).To(BeTemporally("~", time.Now(), 10*time.Second)) + }) }) Context("when status update fails with both cached and non-cached clients", func() { It("should return error", func() { + originalSlotReleaseBackoff := slotReleaseBackoff + slotReleaseBackoff = wait.Backoff{Duration: 1 * time.Millisecond, Steps: 1} + defer func() { slotReleaseBackoff = originalSlotReleaseBackoff }() + // Create a client that will fail status updates clb := fake.ClientBuilder{} clb.WithScheme(s) @@ -426,6 +947,56 @@ var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() }) }) + Describe("incrementUnavailableNodeCount", func() { + var ( + reconciler *NodeNetworkConfigurationPolicyReconciler + nncp *nmstatev1.NodeNetworkConfigurationPolicy + s *runtime.Scheme + ) + + BeforeEach(func() { + reconciler = &NodeNetworkConfigurationPolicyReconciler{ + RetriesUntilFail: 5, + MaximumTimeBackoff: 30 * time.Second, + InitialBackoff: 1 * time.Second, + } + s = scheme.Scheme + s.AddKnownTypes(nmstatev1.GroupVersion, + &nmstatev1.NodeNetworkConfigurationPolicy{}, + &nmstatev1.NodeNetworkConfigurationPolicyList{}, + ) + + nncp = &nmstatev1.NodeNetworkConfigurationPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-policy", + }, + Status: shared.NodeNetworkConfigurationPolicyStatus{ + UnavailableNodeCountMap: map[string]int{ + "gen-1": 0, + }, + }, + } + + reconciler.Log = ctrl.Log.WithName("test") + }) + + It("stamps LastUnavailableNodeCountUpdate on increment", func() { + clb := fake.ClientBuilder{} + clb.WithScheme(s) + clb.WithRuntimeObjects(nncp) + clb.WithStatusSubresource(nncp) + cl := clb.Build() + reconciler.Client = cl + reconciler.APIClient = cl + + Expect(reconciler.incrementUnavailableNodeCount(context.TODO(), nncp, "gen-1")).To(Succeed()) + + updated := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), types.NamespacedName{Name: "test-policy"}, updated)).To(Succeed()) + Expect(updated.Status.LastUnavailableNodeCountUpdate).ToNot(BeNil()) + }) + }) + Describe("fillInEnactmentStatus", func() { var ( reconciler *NodeNetworkConfigurationPolicyReconciler @@ -433,7 +1004,9 @@ var _ = Describe("NodeNetworkConfigurationPolicy controller predicates", func() ) BeforeEach(func() { + originalNmstatectlShowFn := nmstatectlShowFn nmstatectlShowFn = func() (string, error) { return "", nil } + DeferCleanup(func() { nmstatectlShowFn = originalNmstatectlShowFn }) reconciler = &NodeNetworkConfigurationPolicyReconciler{ RetriesUntilFail: 5, MaximumTimeBackoff: 30 * time.Second, diff --git a/pkg/enactmentstatus/conditions/conditions.go b/pkg/enactmentstatus/conditions/conditions.go index ff5753ffe5..89ed5f8b87 100644 --- a/pkg/enactmentstatus/conditions/conditions.go +++ b/pkg/enactmentstatus/conditions/conditions.go @@ -57,12 +57,27 @@ func (ec *EnactmentConditions) NotifyGenerateFailure(ctx context.Context, err er } } -func (ec *EnactmentConditions) NotifyProgressing(ctx context.Context) { +func (ec *EnactmentConditions) NotifyProgressing(ctx context.Context) error { ec.logger.Info("NotifyProgressing") err := ec.updateEnactmentConditions(ctx, SetProgressing, "Applying desired state") if err != nil { ec.logger.Error(err, "Error notifying state Progressing") } + return err +} + +// NotifyFinalizing records that the desired state has been applied and only the +// post-apply finalization (releasing the maxUnavailable slot and recording +// success) remains. The enactment stays Progressing (a live slot holder) but +// carries the ConfigurationFinalizing reason so a retry can skip the already +// committed apply. See SetFinalizing. +func (ec *EnactmentConditions) NotifyFinalizing(ctx context.Context) error { + ec.logger.Info("NotifyFinalizing") + err := ec.updateEnactmentConditions(ctx, SetFinalizing, "Desired state applied, finalizing") + if err != nil { + ec.logger.Error(err, "Error notifying state Finalizing") + } + return err } func (ec *EnactmentConditions) NotifyFailedToConfigure(ctx context.Context, failedErr error) { @@ -89,12 +104,13 @@ func (ec *EnactmentConditions) NotifyAborted(ctx context.Context, failedErr erro } } -func (ec *EnactmentConditions) NotifySuccess(ctx context.Context) { +func (ec *EnactmentConditions) NotifySuccess(ctx context.Context) error { ec.logger.Info("NotifySuccess") err := ec.updateEnactmentConditions(ctx, SetSuccess, "successfully reconciled") if err != nil { ec.logger.Error(err, "Error notifying state Success") } + return err } func (ec *EnactmentConditions) NotifyPending(ctx context.Context) { @@ -261,67 +277,124 @@ func SetSuccess(conditions *nmstate.ConditionList, message string) { } func SetProgressing(conditions *nmstate.ConditionList, message string) { + setProgressingWithReason( + conditions, + nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationProgressing, + message, + ) +} + +// SetFinalizing keeps the enactment Progressing (so it still counts as a live +// maxUnavailable slot holder) but stamps the ConfigurationFinalizing reason, +// marking that the desired state was already applied and only slot release and +// success recording remain. +func SetFinalizing(conditions *nmstate.ConditionList, message string) { + setProgressingWithReason( + conditions, + nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationFinalizing, + message, + ) +} + +func setProgressingWithReason(conditions *nmstate.ConditionList, reason nmstate.ConditionReason, message string) { conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionProgressing, corev1.ConditionTrue, - nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationProgressing, + reason, message, ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionFailing, corev1.ConditionUnknown, - nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationProgressing, + reason, "", ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionAvailable, corev1.ConditionUnknown, - nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationProgressing, + reason, "", ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionPending, corev1.ConditionFalse, - nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationProgressing, + reason, "", ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionAborted, corev1.ConditionFalse, - nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationProgressing, + reason, "", ) } +const interruptedByRestartMessage = "interrupted by handler restart; waiting to be reapplied" + +// MarkInterrupted transitions an enactment that was Progressing when the +// handler died to Pending, so slot audits across the cluster no longer see +// it as a live maxUnavailable slot holder, and resets its retry count for +// the given generation so the re-apply is not skipped. +// +// The Pending transition carries the ConfigurationInterrupted reason (not +// MaxUnavailableLimitReached) so consumers can tell a restart-interrupted +// enactment apart from one throttled by the maxUnavailable cap. +func MarkInterrupted(ctx context.Context, cli client.Client, enactmentKey types.NamespacedName, generationKey string) error { + return enactmentstatus.Update(ctx, cli, enactmentKey, + func(status *nmstate.NodeNetworkConfigurationEnactmentStatus) { + SetPendingWithReason( + &status.Conditions, + nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationInterrupted, + interruptedByRestartMessage, + ) + if status.RetryCount != nil { + status.RetryCount[generationKey] = 0 + } + }) +} + +// SetPending marks the enactment Pending because the maxUnavailable cap +// refused its slot claim. func SetPending(conditions *nmstate.ConditionList, message string) { + SetPendingWithReason( + conditions, + nmstate.NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached, + message, + ) +} + +// SetPendingWithReason marks the enactment Pending with an explicit reason so +// callers can distinguish why the apply is waiting (e.g. throttled by the +// maxUnavailable cap versus interrupted by a handler restart). +func SetPendingWithReason(conditions *nmstate.ConditionList, reason nmstate.ConditionReason, message string) { conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionPending, corev1.ConditionTrue, - nmstate.NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached, + reason, message, ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionAborted, corev1.ConditionFalse, - nmstate.NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached, + reason, "", ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionProgressing, corev1.ConditionFalse, - nmstate.NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached, + reason, message, ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionFailing, corev1.ConditionFalse, - nmstate.NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached, + reason, "", ) conditions.Set( nmstate.NodeNetworkConfigurationEnactmentConditionAvailable, corev1.ConditionFalse, - nmstate.NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached, + reason, "", ) } diff --git a/pkg/enactmentstatus/conditions/conditions_test.go b/pkg/enactmentstatus/conditions/conditions_test.go new file mode 100644 index 0000000000..7433c6b4c9 --- /dev/null +++ b/pkg/enactmentstatus/conditions/conditions_test.go @@ -0,0 +1,72 @@ +/* +Copyright The Kubernetes NMState Authors. + + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package conditions + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + shared "github.com/nmstate/kubernetes-nmstate/api/shared" + nmstatev1beta1 "github.com/nmstate/kubernetes-nmstate/api/v1beta1" +) + +var _ = Describe("MarkInterrupted", func() { + It("sets Pending, clears Progressing, and zeroes the retry count", func() { + s := scheme.Scheme + s.AddKnownTypes(nmstatev1beta1.GroupVersion, + &nmstatev1beta1.NodeNetworkConfigurationEnactment{}) + nnce := nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{Name: "node01.test-policy"}, + Status: shared.NodeNetworkConfigurationEnactmentStatus{ + PolicyGeneration: 3, + RetryCount: map[string]int{"3": 2}, + }, + } + SetProgressing(&nnce.Status.Conditions, "applying") + + clb := fake.ClientBuilder{} + clb.WithScheme(s) + clb.WithRuntimeObjects(&nnce) + clb.WithStatusSubresource(&nnce) + cl := clb.Build() + + Expect(MarkInterrupted(context.TODO(), cl, + types.NamespacedName{Name: "node01.test-policy"}, "3")).To(Succeed()) + + updated := &nmstatev1beta1.NodeNetworkConfigurationEnactment{} + Expect(cl.Get(context.TODO(), types.NamespacedName{Name: "node01.test-policy"}, updated)).To(Succeed()) + pendingCondition := updated.Status.Conditions.Find(shared.NodeNetworkConfigurationEnactmentConditionPending) + Expect(pendingCondition).ToNot(BeNil()) + Expect(pendingCondition.Status).To(Equal(corev1.ConditionTrue)) + Expect(pendingCondition.Message).To(ContainSubstring("interrupted by handler restart")) + Expect(pendingCondition.Reason).To(Equal(shared.NodeNetworkConfigurationEnactmentConditionConfigurationInterrupted), + "a restart-interrupted enactment must not reuse the MaxUnavailableLimitReached reason") + progressingCondition := updated.Status.Conditions.Find(shared.NodeNetworkConfigurationEnactmentConditionProgressing) + Expect(progressingCondition.Status).To(Equal(corev1.ConditionFalse)) + Expect(progressingCondition.Reason).To(Equal(shared.NodeNetworkConfigurationEnactmentConditionConfigurationInterrupted)) + Expect(updated.Status.RetryCount["3"]).To(Equal(0)) + }) +}) diff --git a/pkg/enactmentstatus/status.go b/pkg/enactmentstatus/status.go index e79c915074..78c4dfe7b7 100644 --- a/pkg/enactmentstatus/status.go +++ b/pkg/enactmentstatus/status.go @@ -72,6 +72,18 @@ func IsProgressing(conditions *nmstate.ConditionList) bool { return false } +// IsFinalizing reports whether the enactment is in the post-apply finalization +// phase: still Progressing (a live slot holder) but with the +// ConfigurationFinalizing reason, meaning the desired state was already applied +// and only slot release and success recording remain. Such an enactment must +// not be re-applied. +func IsFinalizing(conditions *nmstate.ConditionList) bool { + progressingCondition := conditions.Find(nmstate.NodeNetworkConfigurationEnactmentConditionProgressing) + return progressingCondition != nil && + progressingCondition.Status == corev1.ConditionTrue && + progressingCondition.Reason == nmstate.NodeNetworkConfigurationEnactmentConditionConfigurationFinalizing +} + func IsAvailable(conditions *nmstate.ConditionList) bool { availableCondition := conditions.Find(nmstate.NodeNetworkConfigurationEnactmentConditionAvailable) return availableCondition != nil && availableCondition.Status == corev1.ConditionTrue diff --git a/pkg/node/audit.go b/pkg/node/audit.go new file mode 100644 index 0000000000..bbb1c01067 --- /dev/null +++ b/pkg/node/audit.go @@ -0,0 +1,175 @@ +/* +Copyright The Kubernetes NMState Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package node + +import ( + "context" + "strconv" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/nmstate/kubernetes-nmstate/api/shared" + nmstatev1 "github.com/nmstate/kubernetes-nmstate/api/v1" + nmstatev1beta1 "github.com/nmstate/kubernetes-nmstate/api/v1beta1" + nmstateclient "github.com/nmstate/kubernetes-nmstate/pkg/client" + "github.com/nmstate/kubernetes-nmstate/pkg/environment" + "github.com/nmstate/kubernetes-nmstate/pkg/probe" +) + +// worstCaseApplyCycle bounds how long ApplyDesiredState can legitimately run +// on the success path. The Progressing heartbeat is stamped once at +// NotifyProgressing and is not refreshed again until the apply returns, so a +// live holder's heartbeat can be this old while it is still applying: +// +// probe.Select (sequential pre-apply gw+dns probes) <= ProbesTotalTimeout +// nmstatectl.Set <= DesiredStateConfigurationTimeout +// probe.Run (sequential post-apply probes) <= ProbesTotalTimeout +// +// Derived from the source-of-truth timeouts so it tracks any change to them. +const worstCaseApplyCycle = nmstateclient.DesiredStateConfigurationTimeout + 2*probe.ProbesTotalTimeout + +const ( + // DefaultStaleEnactmentThreshold is how old a Progressing enactment's + // heartbeat must be before the audit considers its holder dead. It must + // exceed the worst-case apply cycle (worstCaseApplyCycle) with margin; + // otherwise the audit could classify a node that is still legitimately + // applying as stale, free its slot, and let more than maxUnavailable + // nodes reconfigure concurrently. The extra margin absorbs the gap + // between stamping the heartbeat and entering the apply, plus clock skew. + DefaultStaleEnactmentThreshold = worstCaseApplyCycle + 5*time.Minute + + // StaleEnactmentThresholdEnvVar overrides DefaultStaleEnactmentThreshold + // (time.ParseDuration format, e.g. "20m"). + StaleEnactmentThresholdEnvVar = "NMSTATE_ENACTMENT_STALE_THRESHOLD" + + // AuditGraceWindow: if LastUnavailableNodeCountUpdate is younger than + // this, the audit defers. A legitimate incrementer's NotifyProgressing + // write may still be in flight; ghost slots are by definition old. + AuditGraceWindow = 30 * time.Second +) + +// StaleEnactmentThreshold returns the configured staleness threshold. An +// override is honored only if it still exceeds the worst-case apply cycle; +// a smaller value would let the audit free a slot held by a node that is +// still legitimately applying and break maxUnavailable, so it is rejected in +// favor of the safe default. +func StaleEnactmentThreshold() time.Duration { + raw := environment.GetEnvVar(StaleEnactmentThresholdEnvVar, "") + if raw == "" { + return DefaultStaleEnactmentThreshold + } + parsed, err := time.ParseDuration(raw) + if err != nil || parsed <= 0 { + return DefaultStaleEnactmentThreshold + } + if parsed < worstCaseApplyCycle { + return DefaultStaleEnactmentThreshold + } + return parsed +} + +// AuditUnavailableSlots recomputes UnavailableNodeCountMap[currentGeneration] +// from live enactments and repairs it downward if it exceeds the number of +// live holders. A live holder is an enactment of the policy's current +// generation with Progressing=True and a heartbeat younger than +// staleThreshold. Returns true if a repair was written. +// +// The repair is set-to-truth (never a blind decrement), so concurrent audits +// from multiple nodes are idempotent. The count is never raised. +func AuditUnavailableSlots( + ctx context.Context, + statusWriter client.Client, + apiReader client.Reader, + policyKey types.NamespacedName, + staleThreshold time.Duration, +) (bool, error) { + repaired := false + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + repaired = false + policy := &nmstatev1.NodeNetworkConfigurationPolicy{} + if err := apiReader.Get(ctx, policyKey, policy); err != nil { + return err + } + + if policy.Status.LastUnavailableNodeCountUpdate != nil && + time.Since(policy.Status.LastUnavailableNodeCountUpdate.Time) < AuditGraceWindow { + return nil + } + + generationKey := strconv.FormatInt(policy.Generation, 10) + stored := 0 + if policy.Status.UnavailableNodeCountMap != nil { + stored = policy.Status.UnavailableNodeCountMap[generationKey] + } + if stored == 0 { + return nil + } + + live, err := countLiveHolders(ctx, apiReader, policy, staleThreshold) + if err != nil { + return err + } + if live >= stored { + return nil + } + + policy.Status.UnavailableNodeCountMap[generationKey] = live + now := metav1.Now() + policy.Status.LastUnavailableNodeCountUpdate = &now + if err := statusWriter.Status().Update(ctx, policy); err != nil { + return err + } + repaired = true + return nil + }) + return repaired, err +} + +func countLiveHolders( + ctx context.Context, + apiReader client.Reader, + policy *nmstatev1.NodeNetworkConfigurationPolicy, + staleThreshold time.Duration, +) (int, error) { + enactments := nmstatev1beta1.NodeNetworkConfigurationEnactmentList{} + policyLabelFilter := client.MatchingLabels{shared.EnactmentPolicyLabel: policy.Name} + if err := apiReader.List(ctx, &enactments, policyLabelFilter); err != nil { + return 0, err + } + live := 0 + for i := range enactments.Items { + enactment := &enactments.Items[i] + if enactment.Status.PolicyGeneration != policy.Generation { + continue + } + progressing := enactment.Status.Conditions.Find( + shared.NodeNetworkConfigurationEnactmentConditionProgressing) + if progressing == nil || progressing.Status != corev1.ConditionTrue { + continue + } + if time.Since(progressing.LastHeartbeatTime.Time) >= staleThreshold { + continue + } + live++ + } + return live, nil +} diff --git a/pkg/node/audit_test.go b/pkg/node/audit_test.go new file mode 100644 index 0000000000..9df683dcd6 --- /dev/null +++ b/pkg/node/audit_test.go @@ -0,0 +1,229 @@ +/* +Copyright The Kubernetes NMState Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package node + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/nmstate/kubernetes-nmstate/api/shared" + nmstatev1 "github.com/nmstate/kubernetes-nmstate/api/v1" + nmstatev1beta1 "github.com/nmstate/kubernetes-nmstate/api/v1beta1" +) + +func auditPolicy(generation int64, count int, lastUpdate *metav1.Time) *nmstatev1.NodeNetworkConfigurationPolicy { + return &nmstatev1.NodeNetworkConfigurationPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "test-policy", Generation: generation}, + Status: shared.NodeNetworkConfigurationPolicyStatus{ + UnavailableNodeCountMap: map[string]int{"2": count}, + LastUnavailableNodeCountUpdate: lastUpdate, + }, + } +} + +func auditEnactment( + name string, + policyGeneration int64, + progressing corev1.ConditionStatus, + heartbeatAge time.Duration, +) *nmstatev1beta1.NodeNetworkConfigurationEnactment { + e := &nmstatev1beta1.NodeNetworkConfigurationEnactment{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{shared.EnactmentPolicyLabel: "test-policy"}, + }, + Status: shared.NodeNetworkConfigurationEnactmentStatus{ + PolicyGeneration: policyGeneration, + Conditions: shared.ConditionList{ + shared.Condition{ + Type: shared.NodeNetworkConfigurationEnactmentConditionProgressing, + Status: progressing, + LastHeartbeatTime: metav1.Time{Time: time.Now().Add(-heartbeatAge)}, + }, + }, + }, + } + return e +} + +var _ = Describe("AuditUnavailableSlots", func() { + var ( + policyKey = types.NamespacedName{Name: "test-policy"} + oldUpdate = metav1.Time{Time: time.Now().Add(-5 * time.Minute)} + ) + + buildClient := func() *fake.ClientBuilder { + s := scheme.Scheme + s.AddKnownTypes(nmstatev1.GroupVersion, + &nmstatev1.NodeNetworkConfigurationPolicy{}) + s.AddKnownTypes(nmstatev1beta1.GroupVersion, + &nmstatev1beta1.NodeNetworkConfigurationEnactment{}, + &nmstatev1beta1.NodeNetworkConfigurationEnactmentList{}) + clb := fake.ClientBuilder{} + clb.WithScheme(s) + return &clb + } + + It("repairs a ghost slot (count>0, no live holders)", func() { + policy := auditPolicy(2, 1, &oldUpdate) + clb := buildClient() + clb.WithRuntimeObjects(policy) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeTrue()) + + updated := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, updated)).To(Succeed()) + Expect(updated.Status.UnavailableNodeCountMap["2"]).To(Equal(0)) + Expect(updated.Status.LastUnavailableNodeCountUpdate).ToNot(BeNil()) + }) + + It("does not repair when a fresh Progressing holder exists", func() { + policy := auditPolicy(2, 1, &oldUpdate) + holder := auditEnactment("node01.test-policy", 2, corev1.ConditionTrue, time.Minute) + clb := buildClient() + clb.WithRuntimeObjects(policy, holder) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeFalse()) + + updated := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, updated)).To(Succeed()) + Expect(updated.Status.UnavailableNodeCountMap["2"]).To(Equal(1)) + }) + + It("repairs when the only Progressing holder is stale", func() { + policy := auditPolicy(2, 1, &oldUpdate) + dead := auditEnactment("node01.test-policy", 2, corev1.ConditionTrue, DefaultStaleEnactmentThreshold+time.Minute) + clb := buildClient() + clb.WithRuntimeObjects(policy, dead) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeTrue()) + + updated := &nmstatev1.NodeNetworkConfigurationPolicy{} + Expect(cl.Get(context.TODO(), policyKey, updated)).To(Succeed()) + Expect(updated.Status.UnavailableNodeCountMap["2"]).To(Equal(0)) + }) + + It("defers inside the grace window", func() { + recent := metav1.Time{Time: time.Now().Add(-5 * time.Second)} + policy := auditPolicy(2, 1, &recent) + clb := buildClient() + clb.WithRuntimeObjects(policy) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeFalse()) + }) + + It("audits when LastUnavailableNodeCountUpdate is nil", func() { + policy := auditPolicy(2, 1, nil) + clb := buildClient() + clb.WithRuntimeObjects(policy) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeTrue()) + }) + + It("ignores enactments from other generations", func() { + policy := auditPolicy(2, 1, &oldUpdate) + oldGen := auditEnactment("node01.test-policy", 1, corev1.ConditionTrue, time.Minute) + clb := buildClient() + clb.WithRuntimeObjects(policy, oldGen) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeTrue(), "old-generation Progressing must not count as live holder") + }) + + It("is idempotent under repeated invocation", func() { + policy := auditPolicy(2, 2, &oldUpdate) + clb := buildClient() + clb.WithRuntimeObjects(policy) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeTrue()) + + // Second run: count already 0 and timestamp is fresh -> grace defers. + repaired, err = AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeFalse()) + }) + + It("returns zero-count no-op without status write", func() { + policy := auditPolicy(2, 0, &oldUpdate) + clb := buildClient() + clb.WithRuntimeObjects(policy) + clb.WithStatusSubresource(policy) + cl := clb.Build() + + repaired, err := AuditUnavailableSlots(context.TODO(), cl, cl, policyKey, DefaultStaleEnactmentThreshold) + Expect(err).ToNot(HaveOccurred()) + Expect(repaired).To(BeFalse()) + }) +}) + +var _ = Describe("StaleEnactmentThreshold", func() { + It("defaults to the derived threshold", func() { + Expect(StaleEnactmentThreshold()).To(Equal(DefaultStaleEnactmentThreshold)) + }) + It("exceeds the worst-case apply cycle so a live applier is never freed", func() { + Expect(DefaultStaleEnactmentThreshold).To(BeNumerically(">", worstCaseApplyCycle)) + }) + It("honors an env var that still exceeds the worst-case apply cycle", func() { + override := worstCaseApplyCycle + 10*time.Minute + GinkgoT().Setenv(StaleEnactmentThresholdEnvVar, override.String()) + Expect(StaleEnactmentThreshold()).To(Equal(override)) + }) + It("rejects an env var below the worst-case apply cycle, using the default", func() { + GinkgoT().Setenv(StaleEnactmentThresholdEnvVar, "5m") + Expect(StaleEnactmentThreshold()).To(Equal(DefaultStaleEnactmentThreshold)) + }) + It("falls back to default on unparsable value", func() { + GinkgoT().Setenv(StaleEnactmentThresholdEnvVar, "bogus") + Expect(StaleEnactmentThreshold()).To(Equal(DefaultStaleEnactmentThreshold)) + }) +}) diff --git a/test/e2e/handler/nncp_slot_recovery_test.go b/test/e2e/handler/nncp_slot_recovery_test.go new file mode 100644 index 0000000000..b19960844a --- /dev/null +++ b/test/e2e/handler/nncp_slot_recovery_test.go @@ -0,0 +1,106 @@ +/* +Copyright The Kubernetes NMState Authors. + + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package handler + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + corev1 "k8s.io/api/core/v1" + + "sigs.k8s.io/controller-runtime/pkg/client" + + nmstate "github.com/nmstate/kubernetes-nmstate/api/shared" + policyconditions "github.com/nmstate/kubernetes-nmstate/test/e2e/policy" + testenv "github.com/nmstate/kubernetes-nmstate/test/env" +) + +// deleteHandlerPodOnNode deletes the nmstate-handler pod running on the given +// node, simulating a handler death mid-apply. The DaemonSet recreates it. +func deleteHandlerPodOnNode(node string) { + Byf("Deleting nmstate-handler pod on node %s", node) + podList := corev1.PodList{} + filterHandlers := client.MatchingLabels{"component": "kubernetes-nmstate-handler"} + err := testenv.Client.List(context.TODO(), &podList, filterHandlers, client.InNamespace(testenv.OperatorNamespace)) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + deleted := false + for i := range podList.Items { + pod := &podList.Items[i] + if pod.Spec.NodeName == node { + ExpectWithOffset(1, testenv.Client.Delete(context.TODO(), pod)).To(Succeed()) + deleted = true + } + } + ExpectWithOffset(1, deleted).To(BeTrue(), "no nmstate-handler pod found on node %s", node) +} + +// Regression test for OCPBUGS-74261: a handler killed mid-apply must not +// leave the policy permanently blocked on MaxUnavailableLimitReached. The +// replacement handler pod reclaims the ghost unavailable slot at startup and +// the audit recomputes the unavailable node count, so the policy converges. +var _ = Describe("NNCP unavailable-slot recovery after handler death", func() { + Context("when the nmstate-handler pod is deleted while a policy is applying", func() { + BeforeEach(func() { + By("Create a policy that touches all test nodes") + updateDesiredState(linuxBrUp(bridge1)) + + By("Waiting for a node whose enactment is Progressing") + var progressingNode string + Eventually(func() string { + for _, node := range nodes { + enactment := policyconditions.EnactmentConditionsStatus(node, TestPolicy) + condProgressing := enactment.Find(nmstate.NodeNetworkConfigurationEnactmentConditionProgressing) + if condProgressing != nil && condProgressing.Status == corev1.ConditionTrue { + progressingNode = node + return progressingNode + } + } + return "" + }, 15*time.Second, 500*time.Millisecond).ShouldNot(BeEmpty(), + "no node reached Progressing for policy %s", TestPolicy) + + By("Killing the handler pod on the progressing node while the policy progresses") + deleteHandlerPodOnNode(progressingNode) + }) + AfterEach(func() { + // Policy deletion and node reset must run even when the + // absent-wait fails (e.g. the policy is stuck), otherwise the + // policy leaks into subsequent specs. + defer func() { + By("Remove the policy") + deletePolicy(TestPolicy) + By("Reset desired state at all nodes") + resetDesiredStateForNodes() + }() + By("Remove the bridge") + updateDesiredStateAndWait(linuxBrAbsent(bridge1)) + }) + It("should eventually reach Available without recreating the policy", func() { + policyconditions.WaitForAvailablePolicy(TestPolicy) + + By("Verifying no ghost unavailable slots remain") + nncp := nodeNetworkConfigurationPolicy(TestPolicy) + for generation, count := range nncp.Status.UnavailableNodeCountMap { + Expect(count).To(Equal(0), "generation %s should have no unavailable slots", generation) + } + }) + }) +}) diff --git a/vendor/github.com/nmstate/kubernetes-nmstate/api/shared/nodenetworkconfigurationenactment_types.go b/vendor/github.com/nmstate/kubernetes-nmstate/api/shared/nodenetworkconfigurationenactment_types.go index bed2524b3f..f281667902 100644 --- a/vendor/github.com/nmstate/kubernetes-nmstate/api/shared/nodenetworkconfigurationenactment_types.go +++ b/vendor/github.com/nmstate/kubernetes-nmstate/api/shared/nodenetworkconfigurationenactment_types.go @@ -83,6 +83,13 @@ const ( NodeNetworkConfigurationEnactmentConditionMaxUnavailableLimitReached ConditionReason = "MaxUnavailableLimitReached" NodeNetworkConfigurationEnactmentConditionConfigurationProgressing ConditionReason = "ConfigurationProgressing" NodeNetworkConfigurationEnactmentConditionConfigurationAborted ConditionReason = "ConfigurationAborted" + NodeNetworkConfigurationEnactmentConditionConfigurationInterrupted ConditionReason = "ConfigurationInterrupted" + // NodeNetworkConfigurationEnactmentConditionConfigurationFinalizing marks the + // post-apply phase: the desired state was applied and only releasing the + // maxUnavailable slot and recording success remain. A reconcile that finds + // the enactment in this phase must not re-apply the (already committed) + // configuration. + NodeNetworkConfigurationEnactmentConditionConfigurationFinalizing ConditionReason = "ConfigurationFinalizing" ) func EnactmentKey(node, policy string) types.NamespacedName {