From 4e672121b451883f07613af029f81307692ebd0e Mon Sep 17 00:00:00 2001 From: Jeremy Poulin Date: Tue, 4 Aug 2026 15:34:22 -0400 Subject: [PATCH] tnf: continuously run update-setup for node recovery Ensure update-setup job controller runs continuously in post-transition runtime mode to handle node replacement and recovery scenarios. Previously only ran during initial bootstrap transition. When a control plane node is replaced or fails, update-setup now: - Runs continuously to detect missing nodes from Pacemaker cluster - Automatically adds missing nodes back to cluster configuration via pcs cluster node add when detected - Restarts when nodes transition to Ready state (via custom event handler) to ensure new nodes are incorporated into Pacemaker cluster This allows TNF clusters to self-heal from single-node degraded state back to full 2-node operation without manual intervention. Change lifecycle manager to use WithBareInformers with 60-second periodic sync to avoid unnecessary work on unrelated cluster events while maintaining responsiveness for node recovery via custom Ready transition handler. Improve reliability by retrying lifecycle manager creation on transient failures instead of crashing operator. Split bootstrap/runtime logic into separate code paths. Preserve MaxRetriesExceeded latch by passing operator status directly instead of re-fetching. Co-Authored-By: Claude Sonnet 4.5 --- docs/tnf/job-controllers.md | 10 +- docs/tnf/lifecycle-manager.md | 5 +- pkg/tnf/operator/job_controllers.go | 232 +++++++++++---------- pkg/tnf/operator/job_controllers_test.go | 54 +++-- pkg/tnf/operator/lifecycle_manager.go | 93 ++------- pkg/tnf/operator/starter.go | 42 ++-- pkg/tnf/operator/status_collector_test.go | 11 +- pkg/tnf/pkg/jobs/jobcontroller.go | 59 +++++- pkg/tnf/pkg/jobs/jobcontroller_test.go | 243 +++++++++++++++++++++- pkg/tnf/pkg/jobs/lifecycle.go | 109 +++------- pkg/tnf/pkg/jobs/lifecycle_test.go | 23 +- pkg/tnf/pkg/jobs/utils.go | 2 +- pkg/tnf/pkg/pacemaker/healthcheck.go | 37 ++-- pkg/tnf/pkg/pcs/auth.go | 7 + pkg/tnf/pkg/tools/jobs.go | 2 +- pkg/tnf/pkg/tools/nodes.go | 5 + pkg/tnf/update-setup/runner.go | 140 ++++++++++--- 17 files changed, 721 insertions(+), 353 deletions(-) diff --git a/docs/tnf/job-controllers.md b/docs/tnf/job-controllers.md index 87c3e8f5f..9ae25adc3 100644 --- a/docs/tnf/job-controllers.md +++ b/docs/tnf/job-controllers.md @@ -143,7 +143,7 @@ RunNodeJobController(jobType, node, retries, ...) │ ├─ Check node readiness via checkNodesReadinessAndSetCondition │ │ ├─ Not ready < 10min → Return (skip job, retry on next sync) │ │ ├─ Not ready ≥ 10min → Return error (triggers Degraded via WithSyncDegradedOnError) - │ │ └─ Ready → Clear Degraded condition (if was blocked), continue + │ │ └─ Ready → Continue (Degraded clears naturally when job succeeds/completes) │ │ │ └─ Create job spec: │ - Fetch fresh node from informer (handles node replacement) @@ -208,11 +208,11 @@ RunClusterJobController(jobType, schedulableNodesFunc, affectedNodesFunc, jobCon │ │ │ │ │ ├─ Check affectedNodesFunc() → any nodes not ready? │ │ │ ├─ YES → Return error after 10min (triggers Degraded via WithSyncDegradedOnError) - │ │ │ └─ NO → Clear Degraded condition (if was blocked), continue + │ │ │ └─ NO → Continue (Degraded clears naturally when job succeeds/completes) │ │ │ │ │ ├─ Check schedulableNodesFunc() → any nodes available? │ │ │ ├─ NO → Return error after 10min (triggers Degraded via WithSyncDegradedOnError) - │ │ │ └─ YES → Clear Degraded condition (if was blocked), continue + │ │ │ └─ YES → Continue (Degraded clears naturally when job succeeds/completes) │ │ │ │ │ ├─ Get or initialize retry state (AttemptNumber, NodeIndex, config) │ │ │ @@ -224,12 +224,12 @@ RunClusterJobController(jobType, schedulableNodesFunc, affectedNodesFunc, jobCon │ │ │ │ │ ├─ Get current job from cluster │ │ │ │ - │ │ │ ├─ Complete? → Clear degraded, preserve state, return + │ │ │ ├─ Complete? → Return success (Degraded clears naturally via syncManaged flow) │ │ │ │ │ │ │ └─ Failed? │ │ │ ├─ Move to next node (NodeIndex++) │ │ │ ├─ All nodes tried? → Increment attempt (AttemptNumber++) - │ │ │ ├─ Max attempts exhausted? → Set Degraded, reset to attempt 1 + │ │ │ ├─ Max attempts exhausted? → Return error (triggers Degraded via WithSyncDegradedOnError), reset to attempt 1 │ │ │ └─ Update retry state (ApplyJob will detect retry field drift and delete/recreate) │ │ │ │ │ └─ ApplyJob detects drift (NodeName, node-index, or attempt labels changed) → Deletes and recreates job diff --git a/docs/tnf/lifecycle-manager.md b/docs/tnf/lifecycle-manager.md index ab9b5f23b..1152bad41 100644 --- a/docs/tnf/lifecycle-manager.md +++ b/docs/tnf/lifecycle-manager.md @@ -97,8 +97,11 @@ startJobControllers() └─ YES (Runtime Mode): │ ├─ Ensure auth/after-setup controllers running (per-node) + ├─ Ensure setup job controller running (maintains conditions) ├─ Ensure update-setup controller running (if 2 nodes) - └─ Ensure fencing controller running (cluster-wide) + ├─ Ensure fencing controller running (cluster-wide) + ├─ Start status collector CronJob + └─ Start Pacemaker health check controller ``` See [Job Controllers](job-controllers.md) for details on job execution patterns and retry logic. diff --git a/pkg/tnf/operator/job_controllers.go b/pkg/tnf/operator/job_controllers.go index 44f7615e6..cfc61560e 100644 --- a/pkg/tnf/operator/job_controllers.go +++ b/pkg/tnf/operator/job_controllers.go @@ -13,6 +13,7 @@ import ( "github.com/openshift/library-go/pkg/controller/controllercmd" "github.com/openshift/library-go/pkg/operator/v1helpers" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" @@ -34,8 +35,11 @@ const ( ) var ( - // startTnfJobcontrollersFunc is a variable to allow mocking in tests - startTnfJobcontrollersFunc = startTnfJobcontrollers + // startBootstrapJobControllersFunc is a variable to allow mocking in tests + startBootstrapJobControllersFunc = startBootstrapJobControllers + + // startRuntimeJobControllersFunc is a variable to allow mocking in tests + startRuntimeJobControllersFunc = startRuntimeJobControllers // retryBackoffConfig allows customizing retry behavior for tests retryBackoffConfig = wait.Backoff{ @@ -107,7 +111,7 @@ func (c *pacemakerLifecycleManager) startJobControllers(ctx context.Context) err // Jobs report TNFDegraded if affected nodes not ready or no schedulable nodes (after 10 min timeout). klog.V(4).Infof("Transition complete - ensuring job controllers running for %d control plane nodes", len(controlPlaneNodes)) - err = c.startJobControllersWithLock(ctx, controlPlaneNodes) + err = c.startRuntimeJobControllersWithLock(ctx, controlPlaneNodes) if err != nil { return err } @@ -122,7 +126,7 @@ func (c *pacemakerLifecycleManager) startJobControllers(ctx context.Context) err func (c *pacemakerLifecycleManager) retryInitialTransitionOrDegrade(ctx context.Context, nodes []*corev1.Node) error { var setupErr error err := wait.ExponentialBackoffWithContext(ctx, retryBackoffConfig, func(ctx context.Context) (bool, error) { - setupErr = c.startJobControllersWithLock(ctx, nodes) + setupErr = c.startBootstrapJobControllersWithLock(ctx, nodes) if setupErr != nil { klog.Warningf("failed to setup TNF job controllers, will retry: %v", setupErr) return false, nil @@ -162,106 +166,146 @@ func (c *pacemakerLifecycleManager) retryInitialTransitionOrDegrade(ctx context. return nil } -// startJobControllersWithLock serializes job controller startup to prevent concurrent: +// startBootstrapJobControllersWithLock serializes bootstrap job controller startup to prevent concurrent: // - etcd bootstrap / stable revision waits // - duplicate job controller creation // - races in wait logic -func (c *pacemakerLifecycleManager) startJobControllersWithLock(ctx context.Context, nodes []*corev1.Node) error { +func (c *pacemakerLifecycleManager) startBootstrapJobControllersWithLock(ctx context.Context, nodes []*corev1.Node) error { + c.startJobControllersMu.Lock() + defer c.startJobControllersMu.Unlock() + + return startBootstrapJobControllersFunc(ctx, nodes, c.controllerContext, c.operatorClient, c.kubeClient, c.kubeInformersForNamespaces, c.etcdInformer, c) +} + +// startRuntimeJobControllersWithLock serializes runtime job controller startup to prevent concurrent: +// - duplicate job controller creation +func (c *pacemakerLifecycleManager) startRuntimeJobControllersWithLock(ctx context.Context, nodes []*corev1.Node) error { c.startJobControllersMu.Lock() defer c.startJobControllersMu.Unlock() - return startTnfJobcontrollersFunc(ctx, nodes, c.controllerContext, c.operatorClient, c.kubeClient, c.kubeInformersForNamespaces, c.etcdInformer, c) + return startRuntimeJobControllersFunc(ctx, nodes, c.controllerContext, c.operatorClient, c.kubeClient, c.kubeInformersForNamespaces, c.etcdInformer, c) } -// startTnfJobcontrollers creates TNF job controllers for the given nodes. -// During bootstrap: waits for etcd bootstrap completion and stable revision before creating controllers. -// Post-transition: skips bootstrap flow and ensures controllers are running (idempotent restart). +// startCommonJobControllers creates the job controllers that run in both bootstrap and runtime modes. // Creates auth jobs (per-node), setup job (cluster-wide), fencing job (cluster-wide), and after-setup jobs (per-node). -// Setup job is one-time execution but controller runs in both modes to maintain conditions. -func startTnfJobcontrollers( +// Clears legacy condition names from upgrades. +func startCommonJobControllers( ctx context.Context, controlPlaneNodeList []*corev1.Node, controllerContext *controllercmd.ControllerContext, operatorClient v1helpers.StaticPodOperatorClient, kubeClient kubernetes.Interface, kubeInformersForNamespaces v1helpers.KubeInformersForNamespaces, - etcdInformer operatorv1informers.EtcdInformer, - lifecycleManager *pacemakerLifecycleManager) error { + lifecycleManager *pacemakerLifecycleManager, +) { + // Node job controllers (per-node) + for _, node := range controlPlaneNodeList { + jobs.RunNodeJobController(ctx, tools.JobTypeAuth, node, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager.controlPlaneNodeInformer, jobs.DefaultConditions) + jobs.RunNodeJobController(ctx, tools.JobTypeAfterSetup, node, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager.controlPlaneNodeInformer, jobs.DefaultConditions) + } - // Check if transition already complete (operator restart scenario) - // If so, skip bootstrap flow and just ensure controllers are running - transitionComplete, err := ceohelpers.HasExternalEtcdCompletedTransition(ctx, operatorClient) - if err != nil { - klog.Warningf("Failed to check transition status: %v - proceeding with bootstrap flow", err) + // schedulableNodesFunc: returns ready nodes where job can run (K8s ∩ Pacemaker intersection) + // During bootstrap: PacemakerCluster CR doesn't exist yet, so getActivePacemakerNodes falls back to all ready nodes. + // Returns error only when informer is unsynced or no ready nodes exist. + schedulableNodesFunc := func() ([]*corev1.Node, error) { + return lifecycleManager.getActivePacemakerNodes() } - if transitionComplete { - klog.V(4).Infof("Transition already complete - skipping bootstrap flow, ensuring controllers are running") + // affectedNodesFunc for setup: all control plane nodes (ready or not) + // Job waits for these nodes to become ready before proceeding + // Query dynamically from informer to avoid stale node list + setupAffectedNodesFunc := func() ([]*corev1.Node, error) { + return tools.ListNodesFromInformer(lifecycleManager.controlPlaneNodeInformer) + } - // Just start the controllers without going through bootstrap/setup again - // This prevents recreating setup job and racing with reconciliation - for _, node := range controlPlaneNodeList { - jobs.RunNodeJobController(ctx, tools.JobTypeAuth, node, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager.controlPlaneNodeInformer, jobs.DefaultConditions) - jobs.RunNodeJobController(ctx, tools.JobTypeAfterSetup, node, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager.controlPlaneNodeInformer, jobs.DefaultConditions) + // affectedNodesFunc for fencing: all control plane nodes with fencing secrets (ready or not) + // Job waits for these nodes to become ready before proceeding + // Nodes without secrets won't block the job; when secret is added, drift detection triggers restart + // Query dynamically from informer to avoid stale node list + fencingAffectedNodesFunc := func() ([]*corev1.Node, error) { + nodes, err := tools.ListNodesFromInformer(lifecycleManager.controlPlaneNodeInformer) + if err != nil { + return nil, err } + return getNodesWithFencingSecrets(nodes, kubeInformersForNamespaces) + } + + // Setup job controller: maintains conditions for completed setup job + // Even though setup is one-time only, the controller must run to keep conditions current + // (fencing/auth/after-setup jobs wait for setup job completion status) + jobs.RunClusterJobController(ctx, tools.JobTypeSetup, schedulableNodesFunc, setupAffectedNodesFunc, nil, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.AllConditions) + + // Fencing job with drift detection: captures node UIDs + fencing secret UIDs + fencingJobConfigFunc := createFencingJobConfigFunc(lifecycleManager, kubeInformersForNamespaces) + jobs.RunClusterJobController(ctx, tools.JobTypeFencing, schedulableNodesFunc, fencingAffectedNodesFunc, fencingJobConfigFunc, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.DefaultConditions) + + // Clear legacy condition names from upgrades (controllers recreate with new names) + clearLegacyConditions(ctx, operatorClient) +} +// startRuntimeJobControllers creates TNF job controllers after transition is complete. +// Ensures controllers are running (idempotent restart safe). +// Also starts update-setup job (if 2 nodes), status collector CronJob, and health check controller. +func startRuntimeJobControllers( + ctx context.Context, + controlPlaneNodeList []*corev1.Node, + controllerContext *controllercmd.ControllerContext, + operatorClient v1helpers.StaticPodOperatorClient, + kubeClient kubernetes.Interface, + kubeInformersForNamespaces v1helpers.KubeInformersForNamespaces, + etcdInformer operatorv1informers.EtcdInformer, + lifecycleManager *pacemakerLifecycleManager) error { + + klog.V(4).Infof("Transition complete - starting runtime job controllers") + + // Start common job controllers (auth, after-setup, setup, fencing) + startCommonJobControllers(ctx, controlPlaneNodeList, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager) + + // Update-setup job: ensures pacemaker cluster configuration is current + // Runs post-transition only (not needed during bootstrap) + // Only runs when exactly 2 control plane nodes exist (pacemaker limitation) + if len(controlPlaneNodeList) == 2 { // schedulableNodesFunc: returns ready nodes where job can run (K8s ∩ Pacemaker intersection) schedulableNodesFunc := func() ([]*corev1.Node, error) { return lifecycleManager.getActivePacemakerNodes() } - // affectedNodesFunc for update-setup: all control plane nodes (ready or not) - // Job waits for these nodes to become ready before proceeding - // Query dynamically from informer to avoid stale node list on node replacement + // affectedNodesFunc for update-setup: K8s ∩ Pacemaker active nodes (ready or not) + // Only blocks on nodes that are actually in the Pacemaker cluster configuration. + // This prevents deadlock when a node is removed from Pacemaker but can't become Ready + // without update-setup re-adding it (kubelet is a Pacemaker resource). updateSetupAffectedNodesFunc := func() ([]*corev1.Node, error) { - return tools.ListNodesFromInformer(lifecycleManager.controlPlaneNodeInformer) - } - - // affectedNodesFunc for fencing: all control plane nodes with fencing secrets (ready or not) - // Job waits for these nodes to become ready before proceeding - // Nodes without secrets won't block the job; when secret is added, drift detection triggers restart - // Query dynamically from informer to avoid stale node list on node replacement - fencingAffectedNodesFunc := func() ([]*corev1.Node, error) { - nodes, err := tools.ListNodesFromInformer(lifecycleManager.controlPlaneNodeInformer) - if err != nil { - return nil, err - } - return getNodesWithFencingSecrets(nodes, kubeInformersForNamespaces) - } - - // Setup job controller: maintains conditions for completed setup job - // Even though setup is one-time only, the controller must run to keep conditions current - // (fencing/auth/after-setup jobs wait for setup job completion status) - // Query dynamically from informer to avoid stale node list on node replacement - setupAffectedNodesFunc := func() ([]*corev1.Node, error) { - return tools.ListNodesFromInformer(lifecycleManager.controlPlaneNodeInformer) - } - jobs.RunClusterJobController(ctx, tools.JobTypeSetup, schedulableNodesFunc, setupAffectedNodesFunc, nil, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.AllConditions) - - // Update-setup job: ensures pacemaker cluster configuration is current - // Runs post-transition only (not needed during bootstrap) - // Only runs when exactly 2 control plane nodes exist (pacemaker limitation) - if len(controlPlaneNodeList) == 2 { - jobs.RunClusterJobController(ctx, tools.JobTypeUpdateSetup, schedulableNodesFunc, updateSetupAffectedNodesFunc, nil, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.DefaultConditions) - } else { - klog.V(4).Infof("Skipping update-setup job controller: requires exactly 2 control plane nodes, have %d", len(controlPlaneNodeList)) + return lifecycleManager.getActivePacemakerNodes() } - fencingJobConfigFunc := createFencingJobConfigFunc(lifecycleManager, kubeInformersForNamespaces) - jobs.RunClusterJobController(ctx, tools.JobTypeFencing, schedulableNodesFunc, fencingAffectedNodesFunc, fencingJobConfigFunc, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.DefaultConditions) + jobs.RunClusterJobController(ctx, tools.JobTypeUpdateSetup, schedulableNodesFunc, updateSetupAffectedNodesFunc, nil, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.DefaultConditions) + } else { + klog.V(4).Infof("Skipping update-setup job controller: requires exactly 2 control plane nodes, have %d", len(controlPlaneNodeList)) + } - // Start status collector (only after transition is complete, when Pacemaker exists) - lifecycleManager.runPacemakerStatusCollectorCronJob(ctx) + // Start status collector (only after transition is complete, when Pacemaker exists) + lifecycleManager.runPacemakerStatusCollectorCronJob(ctx) - // Start health check controller (only after transition is complete, when Pacemaker exists) - lifecycleManager.runPacemakerHealthCheckController(ctx) + // Start health check controller (only after transition is complete, when Pacemaker exists) + lifecycleManager.runPacemakerHealthCheckController(ctx) - // Clear legacy condition names from upgrades (controllers recreate with new names) - clearLegacyConditions(ctx, operatorClient) + klog.V(4).Infof("Runtime controllers running") + return nil +} - klog.V(4).Infof("Controllers running (post-transition)") - return nil - } +// startBootstrapJobControllers creates TNF job controllers during initial bootstrap. +// Waits for etcd bootstrap completion and stable revision before creating controllers. +// Creates auth jobs (per-node), setup job (cluster-wide), fencing job (cluster-wide), and after-setup jobs (per-node). +// Does not start status collector or health check controller (Pacemaker doesn't exist yet). +func startBootstrapJobControllers( + ctx context.Context, + controlPlaneNodeList []*corev1.Node, + controllerContext *controllercmd.ControllerContext, + operatorClient v1helpers.StaticPodOperatorClient, + kubeClient kubernetes.Interface, + kubeInformersForNamespaces v1helpers.KubeInformersForNamespaces, + etcdInformer operatorv1informers.EtcdInformer, + lifecycleManager *pacemakerLifecycleManager) error { klog.Infof("Running TNF setup procedure. Waiting for etcd bootstrap to complete") @@ -285,46 +329,9 @@ func startTnfJobcontrollers( klog.Infof("all nodes at latest revision, creating TNF job controllers") - // the order of job creation does not matter, the jobs wait on each other as needed - for _, node := range controlPlaneNodeList { - jobs.RunNodeJobController(ctx, tools.JobTypeAuth, node, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager.controlPlaneNodeInformer, jobs.DefaultConditions) - jobs.RunNodeJobController(ctx, tools.JobTypeAfterSetup, node, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager.controlPlaneNodeInformer, jobs.DefaultConditions) - } - - // schedulableNodesFunc: returns ready nodes where job can run (K8s ∩ Pacemaker intersection) - // During bootstrap: PacemakerCluster CR doesn't exist yet, so getActivePacemakerNodes falls back to controlPlaneNodeList - schedulableNodesFunc := func() ([]*corev1.Node, error) { - return lifecycleManager.getActivePacemakerNodes() - } - - // affectedNodesFunc for setup: all control plane nodes (ready or not) - // Job waits for these nodes to become ready before proceeding - // Query dynamically from informer to avoid stale node list - setupAffectedNodesFunc := func() ([]*corev1.Node, error) { - return tools.ListNodesFromInformer(lifecycleManager.controlPlaneNodeInformer) - } - - // affectedNodesFunc for fencing: all control plane nodes with fencing secrets (ready or not) - // Job waits for these nodes to become ready before proceeding - // Nodes without secrets won't block the job; when secret is added, drift detection triggers restart - // Query dynamically from informer to avoid stale node list - fencingAffectedNodesFunc := func() ([]*corev1.Node, error) { - nodes, err := tools.ListNodesFromInformer(lifecycleManager.controlPlaneNodeInformer) - if err != nil { - return nil, err - } - return getNodesWithFencingSecrets(nodes, kubeInformersForNamespaces) - } - - // Cluster-wide jobs: setup and fencing can run on any node - jobs.RunClusterJobController(ctx, tools.JobTypeSetup, schedulableNodesFunc, setupAffectedNodesFunc, nil, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.AllConditions) - - // Fencing job with drift detection: captures node UIDs + fencing secret UIDs - fencingJobConfigFunc := createFencingJobConfigFunc(lifecycleManager, kubeInformersForNamespaces) - jobs.RunClusterJobController(ctx, tools.JobTypeFencing, schedulableNodesFunc, fencingAffectedNodesFunc, fencingJobConfigFunc, 3, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, jobs.DefaultConditions) - - // Clear legacy condition names from upgrades (controllers recreate with new names) - clearLegacyConditions(ctx, operatorClient) + // Start common job controllers (auth, after-setup, setup, fencing) + // The order of job creation does not matter, the jobs wait on each other as needed + startCommonJobControllers(ctx, controlPlaneNodeList, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, lifecycleManager) return nil } @@ -387,7 +394,10 @@ func getNodesWithFencingSecrets(nodes []*corev1.Node, kubeInformersForNamespaces _, err := secretsLister.Secrets(operatorclient.TargetNamespace).Get(secretName) if err == nil { nodesWithSecrets[node.Name] = true + } else if !apierrors.IsNotFound(err) { + return nil, fmt.Errorf("failed to check for fencing secret %s: %w", secretName, err) } + // NotFound is treated as "node doesn't have secret" - skip silently // Note: We don't check for MAC-hashed secrets here (would require expensive matching). // Those nodes will be configured when the fencing job runs and succeeds. } diff --git a/pkg/tnf/operator/job_controllers_test.go b/pkg/tnf/operator/job_controllers_test.go index 0fcaf7f3a..5bc004988 100644 --- a/pkg/tnf/operator/job_controllers_test.go +++ b/pkg/tnf/operator/job_controllers_test.go @@ -17,7 +17,7 @@ Job Controller Startup: └── TestStartJobControllers - Entry point logic and path selection ├── Before transition (bootstrap path): │ ├── 2 ready nodes → bootstrap with retry - │ └── 3 nodes → skip (pacemaker only supports 2) + │ └── 3 nodes → error (pacemaker only supports 2) └── After transition (post-transition path): ├── 2 ready nodes → ensure running └── 1 ready node → ensure running (single-node case) @@ -147,10 +147,10 @@ func TestRetryInitialTransitionOrDegrade(t *testing.T) { etcdInformer: etcdInformers.Operator().V1().Etcds(), } - // Store original startTnfJobcontrollersFunc and replace with mock - originalStartFunc := startTnfJobcontrollersFunc - startTnfJobcontrollersFunc = tt.setupMockStartFunc() - defer func() { startTnfJobcontrollersFunc = originalStartFunc }() + // Store original startBootstrapJobControllersFunc and replace with mock + originalStartFunc := startBootstrapJobControllersFunc + startBootstrapJobControllersFunc = tt.setupMockStartFunc() + defer func() { startBootstrapJobControllersFunc = originalStartFunc }() // Store original backoff config and use faster settings for testing originalBackoff := retryBackoffConfig @@ -353,14 +353,27 @@ func TestStartJobControllers(t *testing.T) { etcdInformer: etcdInformers.Operator().V1().Etcds(), } - // Track if startTnfJobcontrollersFunc was called - startCalled := false - originalStartFunc := startTnfJobcontrollersFunc - startTnfJobcontrollersFunc = func(_ context.Context, _ []*corev1.Node, _ *controllercmd.ControllerContext, _ v1helpers.StaticPodOperatorClient, _ kubernetes.Interface, _ v1helpers.KubeInformersForNamespaces, _ operatorv1informers.EtcdInformer, _ *pacemakerLifecycleManager) error { - startCalled = true + // Track if job controller startup functions were called + bootstrapCalled := 0 + runtimeCalled := 0 + mockBootstrapFunc := func(_ context.Context, _ []*corev1.Node, _ *controllercmd.ControllerContext, _ v1helpers.StaticPodOperatorClient, _ kubernetes.Interface, _ v1helpers.KubeInformersForNamespaces, _ operatorv1informers.EtcdInformer, _ *pacemakerLifecycleManager) error { + bootstrapCalled++ return nil } - defer func() { startTnfJobcontrollersFunc = originalStartFunc }() + mockRuntimeFunc := func(_ context.Context, _ []*corev1.Node, _ *controllercmd.ControllerContext, _ v1helpers.StaticPodOperatorClient, _ kubernetes.Interface, _ v1helpers.KubeInformersForNamespaces, _ operatorv1informers.EtcdInformer, _ *pacemakerLifecycleManager) error { + runtimeCalled++ + return nil + } + + // Mock both bootstrap and runtime paths + originalBootstrapFunc := startBootstrapJobControllersFunc + originalRuntimeFunc := startRuntimeJobControllersFunc + startBootstrapJobControllersFunc = mockBootstrapFunc + startRuntimeJobControllersFunc = mockRuntimeFunc + defer func() { + startBootstrapJobControllersFunc = originalBootstrapFunc + startRuntimeJobControllersFunc = originalRuntimeFunc + }() // Use faster backoff for testing originalBackoff := retryBackoffConfig @@ -381,8 +394,23 @@ func TestStartJobControllers(t *testing.T) { } else { require.NoError(t, err) } - require.Equal(t, tt.expectStartCalled, startCalled, - "Expected startTnfJobcontrollersFunc called=%v, got=%v", tt.expectStartCalled, startCalled) + + // Check correct code path was taken + if tt.transitionComplete { + require.Equal(t, 0, bootstrapCalled, "Bootstrap should not be called when transition complete") + if tt.expectStartCalled { + require.Equal(t, 1, runtimeCalled, "Runtime should be called once when transition complete") + } else { + require.Equal(t, 0, runtimeCalled, "Runtime should not be called when start not expected") + } + } else { + require.Equal(t, 0, runtimeCalled, "Runtime should not be called when transition not complete") + if tt.expectStartCalled { + require.Equal(t, 1, bootstrapCalled, "Bootstrap should be called once when transition not complete") + } else { + require.Equal(t, 0, bootstrapCalled, "Bootstrap should not be called when start not expected") + } + } // If retry path expected, verify TNFJobControllersDegraded condition was set if tt.expectRetryPath && tt.expectStartCalled { diff --git a/pkg/tnf/operator/lifecycle_manager.go b/pkg/tnf/operator/lifecycle_manager.go index 14ed82286..f22111c26 100644 --- a/pkg/tnf/operator/lifecycle_manager.go +++ b/pkg/tnf/operator/lifecycle_manager.go @@ -12,9 +12,6 @@ import ( "github.com/openshift/library-go/pkg/operator/events" "github.com/openshift/library-go/pkg/operator/v1helpers" corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/cache" @@ -70,6 +67,7 @@ type pacemakerLifecycleManager struct { // Returns the controller, the PacemakerLifecycleManager instance, and the PacemakerCluster informer // (which must be started separately - see runPacemakerControllers in pkg/tnf/operator/starter.go). func newPacemakerLifecycleManager( + ctx context.Context, operatorClient v1helpers.StaticPodOperatorClient, kubeClient kubernetes.Interface, eventRecorder events.Recorder, @@ -79,56 +77,12 @@ func newPacemakerLifecycleManager( kubeInformersForNamespaces v1helpers.KubeInformersForNamespaces, etcdInformer operatorv1informers.EtcdInformer, ) (factory.Controller, *pacemakerLifecycleManager, cache.SharedIndexInformer, error) { - // Create REST client for PacemakerStatus CRs - restClient, err := pacemaker.CreatePacemakerRESTClient(restConfig) + // Create PacemakerCluster informer + informer, err := pacemaker.NewPacemakerClusterInformer(restConfig) if err != nil { - return nil, nil, nil, fmt.Errorf("failed to create REST client: %w", err) + return nil, nil, nil, err } - // Create scheme for the parameter codec - scheme := runtime.NewScheme() - if err := pacmkrv1.AddToScheme(scheme); err != nil { - return nil, nil, nil, fmt.Errorf("failed to add scheme for informer: %w", err) - } - - // Create informer for PacemakerCluster - klog.Infof("Creating PacemakerCluster informer for group %s, resource %s", pacmkrv1.SchemeGroupVersion.String(), pacemaker.PacemakerResourceName) - informer := cache.NewSharedIndexInformer( - &pacemaker.PacemakerListWatch{ListWatch: cache.ListWatch{ - ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { - klog.V(4).Infof("PacemakerCluster informer ListFunc called for resource %s", pacemaker.PacemakerResourceName) - sanitizedOptions := pacemaker.SanitizeListOptions(options) - result := &pacmkrv1.PacemakerClusterList{} - err := restClient.Get(). - Resource(pacemaker.PacemakerResourceName). - VersionedParams(&sanitizedOptions, runtime.NewParameterCodec(scheme)). - Do(context.Background()). - Into(result) - if err != nil { - klog.Errorf("Failed to list PacemakerCluster resources (%s): %v", pacemaker.PacemakerResourceName, err) - } else { - klog.V(4).Infof("Successfully listed PacemakerCluster resources, found %d items", len(result.Items)) - } - return result, err - }, - WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { - klog.V(4).Infof("PacemakerCluster informer WatchFunc called for resource %s", pacemaker.PacemakerResourceName) - sanitizedOptions := pacemaker.SanitizeListOptions(options) - watcher, err := restClient.Get(). - Resource(pacemaker.PacemakerResourceName). - VersionedParams(&sanitizedOptions, runtime.NewParameterCodec(scheme)). - Watch(context.Background()) - if err != nil { - klog.Errorf("Failed to watch PacemakerCluster resources (%s): %v", pacemaker.PacemakerResourceName, err) - } - return watcher, err - }, - }}, - &pacmkrv1.PacemakerCluster{}, - pacemaker.HealthCheckResyncInterval, - cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, - ) - c := &pacemakerLifecycleManager{ operatorClient: operatorClient, kubeClient: kubeClient, @@ -138,6 +92,7 @@ func newPacemakerLifecycleManager( controllerContext: controllerContext, kubeInformersForNamespaces: kubeInformersForNamespaces, etcdInformer: etcdInformer, + controllerCtx: ctx, } syncCtx := factory.NewSyncContext(controllerNamePacemakerLifecycle, eventRecorder.WithComponentSuffix("pacemaker-lifecycle-manager")) @@ -145,13 +100,14 @@ func newPacemakerLifecycleManager( klog.Infof("%s controller created, waiting for informers to sync before starting", controllerNamePacemakerLifecycle) klog.Infof("PacemakerLifecycleManager will watch: operatorClient and %s/%s resource", pacmkrv1.SchemeGroupVersion.String(), pacemaker.PacemakerResourceName) - // ResyncEvery ensures the sync function is called at regular intervals (1 minute) - // even if no informer events are detected. + // ResyncEvery ensures the sync function is called every minute with fresh cache data. + // We use WithBareInformers to keep caches synced without triggering sync on every event. + // Custom event handlers (registered below) handle specific cases like node Ready transitions. controller := factory.New(). WithSyncContext(syncCtx). ResyncEvery(time.Minute). WithSync(c.sync). - WithInformers( + WithBareInformers( operatorClient.Informer(), informer, controlPlaneNodeInformer, @@ -193,18 +149,12 @@ func (c *pacemakerLifecycleManager) registerNodeEventHandlers() error { if !oldReady && newReady { klog.Infof("node %s transitioned to ready state - restarting update-setup job", newNode.GetName()) go func() { - // Use controller context (cancelled on shutdown) instead of background context + // Use controller context (cancelled on shutdown) + // Context is initialized in constructor, so always available c.controllerCtxMu.Lock() ctx := c.controllerCtx c.controllerCtxMu.Unlock() - if ctx == nil { - // Controller hasn't started yet, event fired before first sync - // This shouldn't happen (informers sync before events fire), but be defensive - klog.V(4).Infof("Skipping node ready event handler - controller context not yet available") - return - } - // Restart update-setup job when nodes become ready (e.g., after replacement) // This ensures update-setup reruns if it completed before auth ran on new node if err := c.restartUpdateSetupJob(ctx); err != nil { @@ -228,13 +178,6 @@ func (c *pacemakerLifecycleManager) sync(ctx context.Context, syncCtx factory.Sy klog.V(4).Infof("PacemakerLifecycleManager sync started") defer klog.V(4).Infof("PacemakerLifecycleManager sync completed") - // Store controller context on first sync (for event handler goroutines) - c.controllerCtxMu.Lock() - if c.controllerCtx == nil { - c.controllerCtx = ctx - } - c.controllerCtxMu.Unlock() - // Start job controllers (runs in both bootstrap and runtime modes) if err := c.startJobControllers(ctx); err != nil { klog.Errorf("Failed to start job controllers: %v", err) @@ -255,7 +198,6 @@ func (c *pacemakerLifecycleManager) runPacemakerHealthCheckController(ctx contex klog.V(4).Infof("Health check controller already started, skipping duplicate start") return } - c.healthCheckStarted = true c.healthCheckMu.Unlock() healthCheckController, _, err := pacemaker.NewHealthCheckWithInformer( @@ -269,6 +211,11 @@ func (c *pacemakerLifecycleManager) runPacemakerHealthCheckController(ctx contex return } + // Only set flag after successful creation + c.healthCheckMu.Lock() + c.healthCheckStarted = true + c.healthCheckMu.Unlock() + go healthCheckController.Run(ctx, 1) klog.Infof("Health check controller started") } @@ -307,10 +254,12 @@ func (c *pacemakerLifecycleManager) restartUpdateSetupJob(ctx context.Context) e return c.getActivePacemakerNodes() } - // affectedNodesFunc: all control plane nodes (ready or not) - // Job waits for these nodes to become ready before proceeding + // affectedNodesFunc: K8s ∩ Pacemaker active nodes (ready or not) + // Only blocks on nodes that are actually in the Pacemaker cluster configuration. + // This prevents deadlock when a node is removed from Pacemaker but can't become Ready + // without update-setup re-adding it (kubelet is a Pacemaker resource). updateSetupAffectedNodesFunc := func() ([]*corev1.Node, error) { - return tools.ListNodesFromInformer(c.controlPlaneNodeInformer) + return c.getActivePacemakerNodes() } klog.Infof("Restarting update-setup job controller after node ready event") diff --git a/pkg/tnf/operator/starter.go b/pkg/tnf/operator/starter.go index 7929cf7f0..949eb8770 100644 --- a/pkg/tnf/operator/starter.go +++ b/pkg/tnf/operator/starter.go @@ -9,6 +9,7 @@ import ( configv1informers "github.com/openshift/client-go/config/informers/externalversions/config/v1" operatorv1informers "github.com/openshift/client-go/operator/informers/externalversions/operator/v1" "github.com/openshift/library-go/pkg/controller/controllercmd" + "github.com/openshift/library-go/pkg/controller/factory" "github.com/openshift/library-go/pkg/operator/resource/resourceapply" "github.com/openshift/library-go/pkg/operator/staticresourcecontroller" "github.com/openshift/library-go/pkg/operator/v1helpers" @@ -68,10 +69,8 @@ func HandleDualReplicaClusters( return false, err } // Start pacemaker controllers (lifecycle manager, status collector) - // PacemakerLifecycleManager handles ALL node lifecycle events: - // - UpdateFunc: Ready transitions for initial bootstrap - // - AddFunc/DeleteFunc: drift-driven reconciliation - // Secret handler registration happens inside runPacemakerControllers after lifecycleManager is created + // PacemakerLifecycleManager registers UpdateFunc handler for node Ready transitions + // to trigger update-setup job restart during initial bootstrap. runPacemakerControllers(ctx, controllerContext, operatorClient, kubeClient, kubeInformersForNamespaces, etcdInformer, controlPlaneNodeInformer, dynamicClient) return true, nil @@ -172,18 +171,31 @@ func runPacemakerControllers(ctx context.Context, controllerContext *controllerc klog.Infof("PacemakerCluster CRD is established") // Prerequisites met: create and start lifecycle manager controller. - lifecycleController, _, pacemakerInformer, err := newPacemakerLifecycleManager( - operatorClient, - kubeClient, - controllerContext.EventRecorder, - controllerContext.KubeConfig, - controlPlaneNodeInformer, - controllerContext, - kubeInformersForNamespaces, - etcdInformer, - ) + // Retry with backoff to handle transient errors (e.g., API server unavailable). + var lifecycleController factory.Controller + var pacemakerInformer cache.SharedIndexInformer + err = wait.PollUntilContextCancel(ctx, 30*time.Second, true, func(ctx context.Context) (bool, error) { + var createErr error + lifecycleController, _, pacemakerInformer, createErr = newPacemakerLifecycleManager( + ctx, + operatorClient, + kubeClient, + controllerContext.EventRecorder, + controllerContext.KubeConfig, + controlPlaneNodeInformer, + controllerContext, + kubeInformersForNamespaces, + etcdInformer, + ) + if createErr != nil { + klog.Errorf("Failed to create Pacemaker lifecycle manager, will retry: %v", createErr) + return false, nil + } + return true, nil + }) if err != nil { - klog.Fatalf("Failed to create Pacemaker lifecycle manager: %v", err) + klog.Errorf("Context cancelled while creating Pacemaker lifecycle manager: %v", err) + return } // Start the PacemakerCluster informer (controller waits for sync before processing events). diff --git a/pkg/tnf/operator/status_collector_test.go b/pkg/tnf/operator/status_collector_test.go index 668296095..edf0848a1 100644 --- a/pkg/tnf/operator/status_collector_test.go +++ b/pkg/tnf/operator/status_collector_test.go @@ -50,6 +50,7 @@ func TestCheckLastJobFailed(t *testing.T) { jobConditions []batchv1.JobCondition jobFailedCount int32 wantFailed bool + createJob bool // Explicitly control job creation // For multi-job tests multipleJobs []jobSetup }{ @@ -58,6 +59,7 @@ func TestCheckLastJobFailed(t *testing.T) { jobConditions: []batchv1.JobCondition{ {Type: batchv1.JobComplete, Status: corev1.ConditionTrue}, }, + createJob: true, wantFailed: false, }, { @@ -65,6 +67,7 @@ func TestCheckLastJobFailed(t *testing.T) { jobConditions: []batchv1.JobCondition{ {Type: batchv1.JobFailed, Status: corev1.ConditionTrue}, }, + createJob: true, wantFailed: true, }, { @@ -72,6 +75,7 @@ func TestCheckLastJobFailed(t *testing.T) { jobConditions: []batchv1.JobCondition{ {Type: batchv1.JobFailureTarget, Status: corev1.ConditionTrue}, }, + createJob: true, wantFailed: true, }, { @@ -81,16 +85,19 @@ func TestCheckLastJobFailed(t *testing.T) { {Type: batchv1.JobFailed, Status: corev1.ConditionTrue}, }, jobFailedCount: 2, // Had pod failures during retries + createJob: true, wantFailed: false, }, { name: "job still running - no conditions set", jobConditions: []batchv1.JobCondition{}, + createJob: true, wantFailed: false, }, { name: "no jobs exist", - jobConditions: nil, // Will result in empty list + jobConditions: nil, + createJob: false, wantFailed: false, }, { @@ -154,7 +161,7 @@ func TestCheckLastJobFailed(t *testing.T) { _, err := kubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Create(context.Background(), job, metav1.CreateOptions{}) require.NoError(t, err) } - } else if tt.jobConditions != nil { + } else if tt.createJob { job := &batchv1.Job{ ObjectMeta: metav1.ObjectMeta{ Name: "test-job-1", diff --git a/pkg/tnf/pkg/jobs/jobcontroller.go b/pkg/tnf/pkg/jobs/jobcontroller.go index cae9d7a92..18d1e5f4a 100644 --- a/pkg/tnf/pkg/jobs/jobcontroller.go +++ b/pkg/tnf/pkg/jobs/jobcontroller.go @@ -36,6 +36,11 @@ const ( // stuckJobRecoveryTimeout is how long a job must be in Failed state before auto-deletion. // Used to recover from controller parameter migrations or stuck jobs. stuckJobRecoveryTimeout = 10 * time.Minute + + // DegradedMessageMaxRetries is the error message used for MaxRetriesExceeded conditions. + // Used both as the full message when first set, and as a prefix when concatenating with additional errors. + // Detection relies on this message being present in the condition's Message field. + DegradedMessageMaxRetries = "Job failed after maximum retry attempts" ) // TODO This based on DeploymentController in openshift/library-go @@ -232,13 +237,13 @@ func (c *JobController) syncManaged(ctx context.Context, opSpec *opv1.OperatorSp required, err := c.getJob(opSpec) if err != nil { - return err + return preserveMaxRetriesOnReturn(c.instanceName, err, "hook error", opStatus) } // If hook returned nil job, skip applying (nodes not ready, etc.) if required == nil { klog.V(4).Infof("Skipping job application: hook requested skip") - return nil + return preserveMaxRetriesOnReturn(c.instanceName, nil, "hook skipped job application", opStatus) } job, _, err := ApplyJob( @@ -249,7 +254,7 @@ func (c *JobController) syncManaged(ctx context.Context, opSpec *opv1.OperatorSp ExpectedJobGeneration(required, opStatus.Generations), ) if err != nil { - return err + return preserveMaxRetriesOnReturn(c.instanceName, err, "ApplyJob error", opStatus) } // Auto-recovery: delete jobs stuck in Failed state for > 10 minutes @@ -277,7 +282,7 @@ func (c *JobController) syncManaged(ctx context.Context, opSpec *opv1.OperatorSp klog.Infof("Deleted stuck failed job %s, will recreate on next sync", job.Name) } // Return early - next sync will recreate the job - return nil + return preserveMaxRetriesOnReturn(c.instanceName, nil, "deleted stuck job", opStatus) } } @@ -347,14 +352,22 @@ func (c *JobController) syncManaged(ctx context.Context, opSpec *opv1.OperatorSp status, ) if err != nil { - return err + return preserveMaxRetriesOnReturn(c.instanceName, err, "ApplyOperatorStatus error", opStatus) } // return an error for reporting degraded status! // setting a condition manually, similar to available and progressing, doesn't work if IsFailed(*job) { - return fmt.Errorf("Job failed") + return preserveMaxRetriesOnReturn(c.instanceName, fmt.Errorf("Job failed"), "", opStatus) } + + // Preserve MaxRetriesExceeded if job hasn't completed + // (only cleared on actual job success below) + if !IsComplete(*job) { + return preserveMaxRetriesOnReturn(c.instanceName, nil, "waiting for success", opStatus) + } + + // Job complete - actual success, don't preserve degraded conditions return nil } @@ -389,3 +402,37 @@ func (c *JobController) getJob(opSpec *opv1.OperatorSpec) (*batchv1.Job, error) } return required, nil } + +// isJobMaxRetriesExceeded checks if the job has the MaxRetriesExceeded degraded condition set. +// Checks both the Reason field (exact match for MaxRetriesExceeded) and Message field (contains prefix). +// This handles cases where other errors (e.g., SyncError) override the Reason but preserve MaxRetriesExceeded in the message. +func isJobMaxRetriesExceeded(jobName string, opStatus *opv1.OperatorStatus) bool { + degradedCondition := v1helpers.FindOperatorCondition(opStatus.Conditions, + tools.ToPascalCase(jobName)+opv1.OperatorStatusTypeDegraded) + + if degradedCondition == nil || degradedCondition.Status != opv1.ConditionTrue { + return false + } + + // Check if Message contains the MaxRetriesExceeded message + return strings.Contains(degradedCondition.Message, DegradedMessageMaxRetries) +} + +// preserveMaxRetriesOnReturn preserves MaxRetriesExceeded condition when returning from syncManaged. +// If MaxRetriesExceeded is not set, returns the original error/nil unchanged. +// If MaxRetriesExceeded is set: +// - If err != nil: wraps error to preserve MaxRetriesExceeded message +// - If err == nil: returns error with reason to prevent clearing condition +func preserveMaxRetriesOnReturn(jobName string, err error, reason string, opStatus *opv1.OperatorStatus) error { + if !isJobMaxRetriesExceeded(jobName, opStatus) { + return err // Return original error or nil + } + + // MaxRetriesExceeded is set, need to preserve it + if err != nil { + // Wrap the error to preserve MaxRetriesExceeded message + return fmt.Errorf("%s; %w", DegradedMessageMaxRetries, err) + } + // err is nil but MaxRetriesExceeded set, return error to prevent clearing + return fmt.Errorf("%s; %s", DegradedMessageMaxRetries, reason) +} diff --git a/pkg/tnf/pkg/jobs/jobcontroller_test.go b/pkg/tnf/pkg/jobs/jobcontroller_test.go index 688ca2134..0f5093905 100644 --- a/pkg/tnf/pkg/jobs/jobcontroller_test.go +++ b/pkg/tnf/pkg/jobs/jobcontroller_test.go @@ -1,7 +1,49 @@ package jobs +/* +TEST COVERAGE SUMMARY - jobcontroller_test.go +============================================== + +This file tests TNF job controller sync logic and MaxRetriesExceeded condition handling. + +WHAT'S TESTED +------------- + +Job Controller Sync: +├── TestSync - Job controller sync loop and job lifecycle +│ ├── Job not created without sync +│ ├── Initial sync will create job +│ ├── Job completed -> clears degraded condition +│ ├── Job running -> no-op +│ ├── Job failed -> updates retry state +│ └── Job failed stuck -> auto-deleted after 10 minutes +├── TestSyncWithAvailableCondition - Available condition management +│ ├── Job completed with available condition -> Available=True +│ ├── Job running with available condition -> Available=False +│ └── Job failed with available condition -> Available=False +└── TestJobModificationRecreation - Job drift detection and recreation + └── Job spec modified -> detected and recreated + +MaxRetriesExceeded Condition Handling: +├── TestIsJobMaxRetriesExceeded - Detection via Message field +│ ├── No degraded condition -> false +│ ├── Degraded false -> false +│ ├── Degraded true with exact MaxRetries message -> true +│ ├── Degraded true with concatenated MaxRetries message -> true +│ ├── Degraded true with SyncError but different message -> false +│ └── Degraded true with different reason and no MaxRetries -> false +└── TestPreserveMaxRetriesOnReturn - Preservation for both nil and error returns + ├── err=nil, MaxRetriesExceeded not set -> returns nil + ├── err=nil, degraded false -> returns nil + ├── err=nil, MaxRetriesExceeded set -> returns wrapped error with reason + ├── err!=nil, MaxRetriesExceeded not set -> returns original error + ├── err!=nil, MaxRetriesExceeded set -> returns wrapped error + └── err=nil, different degraded reason -> returns nil +*/ + import ( "context" + "errors" "slices" "sort" "strings" @@ -23,7 +65,7 @@ import ( batchv1 "k8s.io/api/batch/v1" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/equality" - "k8s.io/apimachinery/pkg/api/errors" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" coreinformers "k8s.io/client-go/informers" @@ -245,7 +287,7 @@ func TestSync(t *testing.T) { if err == nil { t.Errorf("Expected Job to be deleted, found generation %d", actualJob.Generation) } - if !errors.IsNotFound(err) { + if !apierrors.IsNotFound(err) { t.Errorf("Expected error to be NotFound, got %s", err) } } @@ -454,7 +496,7 @@ func TestJobModificationRecreation(t *testing.T) { if err == nil { t.Errorf("Job should have been deleted after first sync") } - if !errors.IsNotFound(err) { + if !apierrors.IsNotFound(err) { t.Errorf("Expected NotFound error, got: %v", err) } @@ -832,3 +874,198 @@ func sanitizeObjectMeta(meta *metav1.ObjectMeta) { meta.Finalizers = nil } } + +func TestIsJobMaxRetriesExceeded(t *testing.T) { + tests := []struct { + name string + degradedCondition *opv1.OperatorCondition + expectedResult bool + }{ + { + name: "no degraded condition", + degradedCondition: nil, + expectedResult: false, + }, + { + name: "degraded false", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionFalse, + }, + expectedResult: false, + }, + { + name: "degraded true with exact MaxRetries message", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionTrue, + Reason: "SyncError", + Message: DegradedMessageMaxRetries, + }, + expectedResult: true, + }, + { + name: "degraded true with concatenated MaxRetries message", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionTrue, + Reason: "SyncError", + Message: DegradedMessageMaxRetries + "; job spec was modified, old job is deleted", + }, + expectedResult: true, + }, + { + name: "degraded true with SyncError reason and message does not contain MaxRetries", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionTrue, + Reason: "SyncError", + Message: "some other error", + }, + expectedResult: false, + }, + { + name: "degraded true with different reason and no MaxRetries in message", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionTrue, + Reason: "SomeOtherReason", + Message: "different error", + }, + expectedResult: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Create fake operator client with the condition + var conditions []opv1.OperatorCondition + if tt.degradedCondition != nil { + conditions = append(conditions, *tt.degradedCondition) + } + + opStatus := &opv1.StaticPodOperatorStatus{ + OperatorStatus: opv1.OperatorStatus{ + Conditions: conditions, + }, + } + + result := isJobMaxRetriesExceeded(controllerName, &opStatus.OperatorStatus) + if result != tt.expectedResult { + t.Errorf("isJobMaxRetriesExceeded() = %v, want %v", result, tt.expectedResult) + } + }) + } +} + +func TestPreserveMaxRetriesOnReturn(t *testing.T) { + // Sentinel error for testing error wrapping and identity preservation + sentinelErr := errors.New("ApplyJob failed") + + tests := []struct { + name string + degradedCondition *opv1.OperatorCondition + inputErr error + reason string + expectError bool + expectedErrContains string + }{ + { + name: "err=nil, MaxRetriesExceeded not set - returns nil", + degradedCondition: nil, + inputErr: nil, + reason: "waiting for success", + expectError: false, + }, + { + name: "err=nil, degraded false - returns nil", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionFalse, + }, + inputErr: nil, + reason: "waiting for success", + expectError: false, + }, + { + name: "err=nil, MaxRetriesExceeded set - returns wrapped error with reason", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionTrue, + Reason: "SyncError", + Message: DegradedMessageMaxRetries, + }, + inputErr: nil, + reason: "waiting for success", + expectError: true, + expectedErrContains: DegradedMessageMaxRetries + "; waiting for success", + }, + { + name: "err!=nil, MaxRetriesExceeded not set - returns original error", + degradedCondition: nil, + inputErr: sentinelErr, + reason: "", + expectError: true, + expectedErrContains: "ApplyJob failed", + }, + { + name: "err!=nil, MaxRetriesExceeded set - returns wrapped error", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionTrue, + Reason: "SyncError", + Message: DegradedMessageMaxRetries, + }, + inputErr: sentinelErr, + reason: "", + expectError: true, + expectedErrContains: DegradedMessageMaxRetries + "; ApplyJob failed", + }, + { + name: "err=nil, different degraded reason - returns nil", + degradedCondition: &opv1.OperatorCondition{ + Type: conditionDegraded, + Status: opv1.ConditionTrue, + Reason: "SomeOtherReason", + Message: "some other error", + }, + inputErr: nil, + reason: "waiting for success", + expectError: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Create fake operator status with the condition + var conditions []opv1.OperatorCondition + if tt.degradedCondition != nil { + conditions = append(conditions, *tt.degradedCondition) + } + + opStatus := &opv1.OperatorStatus{ + Conditions: conditions, + } + + err := preserveMaxRetriesOnReturn(controllerName, tt.inputErr, tt.reason, opStatus) + + if tt.expectError { + if err == nil { + t.Errorf("preserveMaxRetriesOnReturn() expected error, got nil") + } else { + if !strings.Contains(err.Error(), tt.expectedErrContains) { + t.Errorf("preserveMaxRetriesOnReturn() error = %v, want to contain %v", err.Error(), tt.expectedErrContains) + } + // Verify error wrapping preserves identity when inputErr is non-nil + if tt.inputErr != nil && !errors.Is(err, tt.inputErr) { + t.Errorf("preserveMaxRetriesOnReturn() error does not wrap inputErr: errors.Is(err, inputErr) = false") + } + } + } else { + if err != nil { + t.Errorf("preserveMaxRetriesOnReturn() expected nil, got error: %v", err) + } + } + }) + } +} diff --git a/pkg/tnf/pkg/jobs/lifecycle.go b/pkg/tnf/pkg/jobs/lifecycle.go index 526b9b308..095f499a4 100644 --- a/pkg/tnf/pkg/jobs/lifecycle.go +++ b/pkg/tnf/pkg/jobs/lifecycle.go @@ -2,6 +2,7 @@ package jobs import ( "context" + "errors" "fmt" "os" "sync" @@ -82,29 +83,22 @@ const ( // blockedConditionTimeout is how long to wait before returning error when jobs are blocked. // Error triggers Degraded condition via WithSyncDegradedOnError. blockedConditionTimeout = 10 * time.Minute - - // Degraded condition reasons - degradedReasonAsExpected = "AsExpected" - degradedReasonMaxRetriesExceeded = "MaxRetriesExceeded" ) // manageTimedBlockedCondition manages the Degraded condition for blocked jobs. -// Returns error after timeout to trigger WithSyncDegradedOnError, manually clears when unblocked. +// Returns error after timeout to trigger WithSyncDegradedOnError. When unblocked, returns success +// and lets natural syncManaged flow determine degraded state (job complete/failed/running). // Returns (ready bool, error): // - (true, nil): unblocked, ready to proceed // - (false, nil): blocked but < timeout, wait without reporting // - (false, error): blocked >= timeout, error triggers Degraded func manageTimedBlockedCondition( - ctx context.Context, jobName string, isBlocked bool, blockedSinceMap map[string]time.Time, mapMutex *sync.Mutex, errorMessage string, - operatorClient v1helpers.StaticPodOperatorClient, ) (bool, error) { - conditionName := tools.ToPascalCase(jobName) + "Degraded" - if isBlocked { // Track blocked time mapMutex.Lock() @@ -124,61 +118,44 @@ func manageTimedBlockedCondition( return false, nil // Blocked but not long enough yet } - // Clear blocked tracking and Degraded condition + // Clear blocked tracking (condition cleared naturally via syncManaged flow) mapMutex.Lock() - _, wasBlocked := blockedSinceMap[jobName] - if wasBlocked { - delete(blockedSinceMap, jobName) - } + delete(blockedSinceMap, jobName) mapMutex.Unlock() - // Manually clear Degraded condition when unblocked (empty message to avoid cluttering ClusterOperator rollup) - if wasBlocked { - _, _, updateErr := v1helpers.UpdateStatus(ctx, operatorClient, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{ - Type: conditionName, - Status: operatorv1.ConditionFalse, - Reason: degradedReasonAsExpected, - Message: "", - })) - if updateErr != nil { - klog.Errorf("Failed to clear %s condition: %v", conditionName, updateErr) - } - } - return true, nil } -// manageBlockedCondition manages Degraded for jobs blocked by nodes not ready. -// Returns error after 10 min timeout, manually clears when ready. -func manageBlockedCondition(ctx context.Context, jobName string, notReadyNodes []string, operatorClient v1helpers.StaticPodOperatorClient) (bool, error) { +// manageBlockedCondition tracks blocked state for jobs waiting on nodes to become ready. +// Returns (true, nil) if nodes are ready, (false, nil) if blocked but not timed out, +// or (false, error) after 10 min timeout (triggers WithSyncDegradedOnError). +func manageBlockedCondition(jobName string, notReadyNodes []string) (bool, error) { return manageTimedBlockedCondition( - ctx, jobName, len(notReadyNodes) > 0, jobBlockedSince, &jobBlockedMutex, fmt.Sprintf("Affected nodes not ready: %v", notReadyNodes), - operatorClient, ) } -// manageNoSchedulableNodesBlockedCondition manages Degraded when no schedulable nodes available. -// Returns error after 10 min timeout, manually clears when nodes available. -func manageNoSchedulableNodesBlockedCondition(ctx context.Context, jobName string, hasSchedulableNodes bool, operatorClient v1helpers.StaticPodOperatorClient) (bool, error) { +// manageNoSchedulableNodesBlockedCondition tracks blocked state when no schedulable nodes are available. +// Returns (true, nil) if schedulable nodes exist, (false, nil) if blocked but not timed out, +// or (false, error) after 10 min timeout (triggers WithSyncDegradedOnError). +func manageNoSchedulableNodesBlockedCondition(jobName string, hasSchedulableNodes bool) (bool, error) { return manageTimedBlockedCondition( - ctx, jobName, !hasSchedulableNodes, jobNoSchedulableNodesSince, &jobNoSchedulableNodesMutex, "No schedulable nodes available", - operatorClient, ) } -// checkNodesReadinessAndSetCondition checks node readiness and manages Degraded condition. -// Returns error if nodes not ready > 10 min (triggers WithSyncDegradedOnError). -func checkNodesReadinessAndSetCondition(ctx context.Context, nodes []*corev1.Node, jobName string, operatorClient v1helpers.StaticPodOperatorClient) (bool, error) { +// checkNodesReadinessAndSetCondition checks if all nodes are ready and tracks blocked state with timeout. +// Returns (true, nil) if all nodes ready, (false, nil) if some not ready but not timed out, +// or (false, error) if nodes not ready > 10 min (triggers WithSyncDegradedOnError). +func checkNodesReadinessAndSetCondition(nodes []*corev1.Node, jobName string) (bool, error) { // Collect all not-ready nodes var notReadyNodes []string @@ -188,8 +165,8 @@ func checkNodesReadinessAndSetCondition(ctx context.Context, nodes []*corev1.Nod } } - // Manage degraded condition, propagate error if blocked >= timeout - ready, err := manageBlockedCondition(ctx, jobName, notReadyNodes, operatorClient) + // Track blocked state with timeout, propagate error if blocked >= timeout + ready, err := manageBlockedCondition(jobName, notReadyNodes) if err != nil { return false, err // Propagate error to trigger WithSyncDegradedOnError } @@ -213,7 +190,7 @@ func syncMultiNodeJobState(ctx context.Context, jobName string, schedulableNodes } // Check readiness and manage blocked condition - ready, err := checkNodesReadinessAndSetCondition(ctx, affectedNodes, jobName, operatorClient) + ready, err := checkNodesReadinessAndSetCondition(affectedNodes, jobName) if err != nil { return err } @@ -236,7 +213,7 @@ func syncMultiNodeJobState(ctx context.Context, jobName string, schedulableNodes return fmt.Errorf("failed to get schedulable nodes: %w", err) } if len(schedulableNodes) == 0 { - _, err := manageNoSchedulableNodesBlockedCondition(ctx, jobName, false, operatorClient) + _, err := manageNoSchedulableNodesBlockedCondition(jobName, false) return err // Propagate error to trigger WithSyncDegradedOnError if blocked >= timeout } @@ -266,7 +243,7 @@ func syncMultiNodeJobState(ctx context.Context, jobName string, schedulableNodes retryStateMutex.Unlock() // Clear blocked condition now that schedulable nodes are available - if _, err := manageNoSchedulableNodesBlockedCondition(ctx, jobName, true, operatorClient); err != nil { + if _, err := manageNoSchedulableNodesBlockedCondition(jobName, true); err != nil { klog.Errorf("Failed to clear blocked condition for %s: %v", jobName, err) } @@ -285,12 +262,12 @@ func syncMultiNodeJobState(ctx context.Context, jobName string, schedulableNodes return fmt.Errorf("failed to get schedulable nodes: %w", err) } if len(schedulableNodes) == 0 { - _, err = manageNoSchedulableNodesBlockedCondition(ctx, jobName, false, operatorClient) + _, err = manageNoSchedulableNodesBlockedCondition(jobName, false) return err // Propagate error to trigger WithSyncDegradedOnError if blocked >= timeout } // Clear blocked condition if schedulable nodes are now available - if _, err := manageNoSchedulableNodesBlockedCondition(ctx, jobName, true, operatorClient); err != nil { + if _, err := manageNoSchedulableNodesBlockedCondition(jobName, true); err != nil { klog.Errorf("Failed to clear blocked condition for %s: %v", jobName, err) } @@ -353,20 +330,8 @@ func syncMultiNodeJobState(ctx context.Context, jobName string, schedulableNodes // Job exists - check if it's done if IsComplete(*existingJob) { - // Success - clear degraded condition + // Success - condition cleared naturally via syncManaged flow klog.V(4).Infof("Job %s completed successfully", jobName) - - // Clear degraded condition if it was set - _, _, err := v1helpers.UpdateStatus(ctx, operatorClient, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{ - Type: tools.ToPascalCase(jobName) + operatorv1.OperatorStatusTypeDegraded, - Status: operatorv1.ConditionFalse, - Reason: "AsExpected", - Message: fmt.Sprintf("Job %s completed successfully", jobName), - })) - if err != nil { - klog.Errorf("Failed to clear degraded condition for %s: %v", jobName, err) - } - return nil } @@ -383,27 +348,16 @@ func syncMultiNodeJobState(ctx context.Context, jobName string, schedulableNodes // Check if we've exhausted all nodes in this attempt exhaustedNodes := nextNodeIndex >= len(schedulableNodes) + maxRetriesExceeded := false if exhaustedNodes { nextNodeIndex = 0 exhaustedAttempts := currentAttemptNumber >= state.MaxRetryAttempts if exhaustedAttempts { - // Exceeded max attempts - set degraded condition and reset to attempt 1 + // Exceeded max attempts - reset to attempt 1 and continue trying klog.Warningf("Job %s exhausted all %d attempts (tried %d nodes each), marking degraded", jobName, state.MaxRetryAttempts, len(schedulableNodes)) - - // Set degraded condition to indicate job has failed after all retries - _, _, err := v1helpers.UpdateStatus(ctx, operatorClient, v1helpers.UpdateConditionFn(operatorv1.OperatorCondition{ - Type: tools.ToPascalCase(jobName) + operatorv1.OperatorStatusTypeDegraded, - Status: operatorv1.ConditionTrue, - Reason: degradedReasonMaxRetriesExceeded, - Message: fmt.Sprintf("Job failed after %d attempts across all nodes", state.MaxRetryAttempts), - })) - if err != nil { - klog.Errorf("Failed to set degraded condition for %s: %v", jobName, err) - } - - // Reset to attempt 1 and continue trying (degraded condition remains set until success) nextAttemptNumber = 1 + maxRetriesExceeded = true } else { // Start new attempt nextAttemptNumber++ @@ -416,6 +370,11 @@ func syncMultiNodeJobState(ctx context.Context, jobName string, schedulableNodes klog.V(4).Infof("Job %s failed - updating retry state to node index %d", jobName, nextNodeIndex) state.NodeIndex = nextNodeIndex state.AttemptNumber = nextAttemptNumber + + // If we just exceeded max retries, return error to set degraded condition + if maxRetriesExceeded { + return errors.New(DegradedMessageMaxRetries) + } } // Job is running - nothing to do @@ -517,7 +476,7 @@ func RunNodeJobController(ctx context.Context, jobType tools.JobType, node *core } // Check node readiness before configuring job - ready, err := checkNodesReadinessAndSetCondition(ctx, []*corev1.Node{freshNode}, job.Name, operatorClient) + ready, err := checkNodesReadinessAndSetCondition([]*corev1.Node{freshNode}, job.Name) if err != nil { return false, err } diff --git a/pkg/tnf/pkg/jobs/lifecycle_test.go b/pkg/tnf/pkg/jobs/lifecycle_test.go index b2ec07fb4..feaded94b 100644 --- a/pkg/tnf/pkg/jobs/lifecycle_test.go +++ b/pkg/tnf/pkg/jobs/lifecycle_test.go @@ -16,7 +16,7 @@ Job Controller Lifecycle: ├── TestSyncMultiNodeJobState_RetryProgression - Multi-node retry state machine │ ├── Job fails on node 0 -> advances to node 1 │ ├── All nodes fail in attempt 1 -> starts attempt 2 -│ ├── Max attempts exhausted -> degraded condition set, reset to attempt 1 +│ ├── Max attempts exhausted -> returns error with DegradedMessageMaxRetries, resets to attempt 1 │ └── Job succeeds -> degraded cleared ├── TestSyncMultiNodeJobState_DriftDetection - Infrastructure drift detection │ ├── schedulableNodesFunc changes (node added) -> reset state, delete job @@ -137,7 +137,8 @@ func TestRestartJobOrRunController(t *testing.T) { restartJobLocksMutex.Unlock() // Setup - ctx := context.Background() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() client := tt.setupClient() fakeOperatorClient := v1helpers.NewFakeStaticPodOperatorClient( @@ -216,7 +217,7 @@ func TestSyncMultiNodeJobState_RetryProgression(t *testing.T) { // This test verifies the multi-node retry state machine: // - Job fails on node 0 -> retries on node 1 // - Job fails on node 1 -> new attempt, back to node 0 - // - Max attempts exhausted -> degraded condition set, reset to attempt 1 + // - Max attempts exhausted -> returns error (not nil), resets to attempt 1 // - Job succeeds -> degraded cleared, state reset ctx := context.Background() @@ -328,14 +329,15 @@ func TestSyncMultiNodeJobState_RetryProgression(t *testing.T) { _, err = fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Create(ctx, failedJob.DeepCopy(), metav1.CreateOptions{}) require.NoError(t, err) err = syncMultiNodeJobState(ctx, jobName, targetNodesFunc, nil, nil, maxRetries, fakeKubeClient, fakeOperatorClient) - require.NoError(t, err) + require.Error(t, err, "syncMultiNodeJobState returns error when max retries exceeded") + require.Contains(t, err.Error(), DegradedMessageMaxRetries, "Error should contain MaxRetriesExceeded message") state = getState() require.Equal(t, 1, state.AttemptNumber, "Should reset to attempt 1 after exhausting max attempts") require.Equal(t, 0, state.NodeIndex, "Should reset to node index 0") - require.True(t, isDegraded(), "Should be degraded after exhausting max attempts") // Delete job to simulate ApplyJob detecting drift - fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Delete(ctx, jobName, metav1.DeleteOptions{}) + err = fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Delete(ctx, jobName, metav1.DeleteOptions{}) + require.NoError(t, err) // Step 5: Job succeeds -> should clear degraded and preserve state successJob := &batchv1.Job{ @@ -346,7 +348,8 @@ func TestSyncMultiNodeJobState_RetryProgression(t *testing.T) { }, }, } - fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Create(ctx, successJob, metav1.CreateOptions{}) + _, err = fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Create(ctx, successJob, metav1.CreateOptions{}) + require.NoError(t, err) err = syncMultiNodeJobState(ctx, jobName, targetNodesFunc, nil, nil, maxRetries, fakeKubeClient, fakeOperatorClient) require.NoError(t, err) @@ -474,7 +477,8 @@ func TestSyncMultiNodeJobState_DriftDetection(t *testing.T) { }, }, } - fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Create(ctx, failedJob, metav1.CreateOptions{}) + _, err = fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Create(ctx, failedJob, metav1.CreateOptions{}) + require.NoError(t, err) // Step 2: Trigger drift (either nodes or config change) if tt.testDriftType == "nodes" { @@ -490,7 +494,8 @@ func TestSyncMultiNodeJobState_DriftDetection(t *testing.T) { // For non-drift cases, simulate ApplyJob deleting the job due to NodeName change from retry progression // (syncMultiNodeJobState updated state, next sync ApplyJob would detect NodeName drift and delete) if !tt.expectStateReset && tt.expectJobDeleted { - fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Delete(ctx, jobName, metav1.DeleteOptions{}) + err = fakeKubeClient.BatchV1().Jobs(operatorclient.TargetNamespace).Delete(ctx, jobName, metav1.DeleteOptions{}) + require.NoError(t, err) } // Step 4: Verify state reset diff --git a/pkg/tnf/pkg/jobs/utils.go b/pkg/tnf/pkg/jobs/utils.go index 0cfd11968..b81afcba3 100644 --- a/pkg/tnf/pkg/jobs/utils.go +++ b/pkg/tnf/pkg/jobs/utils.go @@ -110,7 +110,7 @@ func IsStopped(job batchv1.Job) bool { // Check for FailureTarget condition (Kubernetes 1.31+) // FailureTarget means job is targeting failure but pods may still be terminating - if IsConditionTrue(job.Status.Conditions, batchv1.JobConditionType("FailureTarget")) { + if IsConditionTrue(job.Status.Conditions, batchv1.JobFailureTarget) { klog.V(2).Infof("Job %s considered stopped: FailureTarget condition is True", job.Name) return true } diff --git a/pkg/tnf/pkg/pacemaker/healthcheck.go b/pkg/tnf/pkg/pacemaker/healthcheck.go index 3ddf2cff3..2205d1344 100644 --- a/pkg/tnf/pkg/pacemaker/healthcheck.go +++ b/pkg/tnf/pkg/pacemaker/healthcheck.go @@ -132,25 +132,20 @@ type HealthCheck struct { crNodes *[]pacmkrv1.PacemakerClusterNodeStatus } -// NewHealthCheck creates a new HealthCheck for monitoring pacemaker status -// in clusters that use ExternalEtcd clusters. -// Returns the controller and the PacemakerCluster informer (which must be started separately). -func NewHealthCheck( - operatorClient v1helpers.StaticPodOperatorClient, - kubeClient kubernetes.Interface, - eventRecorder events.Recorder, - restConfig *rest.Config, -) (factory.Controller, cache.SharedIndexInformer, error) { - // Create REST client for PacemakerStatus CRs +// NewPacemakerClusterInformer creates a PacemakerCluster informer with standard configuration. +// Returns a SharedIndexInformer ready to be started and used by controllers. +// The caller is responsible for starting the informer (via informer.Run(stopCh)). +func NewPacemakerClusterInformer(restConfig *rest.Config) (cache.SharedIndexInformer, error) { + // Create REST client for PacemakerCluster CRs restClient, err := CreatePacemakerRESTClient(restConfig) if err != nil { - return nil, nil, fmt.Errorf("failed to create REST client: %w", err) + return nil, fmt.Errorf("failed to create REST client: %w", err) } // Create scheme for the parameter codec scheme := runtime.NewScheme() if err := pacmkrv1.AddToScheme(scheme); err != nil { - return nil, nil, fmt.Errorf("failed to add scheme for informer: %w", err) + return nil, fmt.Errorf("failed to add scheme for informer: %w", err) } // Create informer for PacemakerCluster @@ -191,6 +186,24 @@ func NewHealthCheck( cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, ) + return informer, nil +} + +// NewHealthCheck creates a new HealthCheck for monitoring pacemaker status +// in clusters that use ExternalEtcd clusters. +// Returns the controller and the PacemakerCluster informer (which must be started separately). +func NewHealthCheck( + operatorClient v1helpers.StaticPodOperatorClient, + kubeClient kubernetes.Interface, + eventRecorder events.Recorder, + restConfig *rest.Config, +) (factory.Controller, cache.SharedIndexInformer, error) { + // Create PacemakerCluster informer + informer, err := NewPacemakerClusterInformer(restConfig) + if err != nil { + return nil, nil, err + } + return NewHealthCheckWithInformer(operatorClient, kubeClient, eventRecorder, informer) } diff --git a/pkg/tnf/pkg/pcs/auth.go b/pkg/tnf/pkg/pcs/auth.go index 075860dcc..d1eb18455 100644 --- a/pkg/tnf/pkg/pcs/auth.go +++ b/pkg/tnf/pkg/pcs/auth.go @@ -46,9 +46,16 @@ func Authenticate(ctx context.Context, configClient *versioned.Clientset, cfg co return false, fmt.Errorf("failed to accept token: %w", err) } + // Check for partial second-node configuration (likely a config error) + if (cfg.NodeName2 != "" && cfg.NodeIP2 == "") || (cfg.NodeName2 == "" && cfg.NodeIP2 != "") { + return false, fmt.Errorf("partial second-node configuration detected: NodeName2=%q, NodeIP2=%q - both should be set for two-node auth or both empty for single-node", cfg.NodeName2, cfg.NodeIP2) + } + if cfg.NodeName2 != "" && cfg.NodeIP2 != "" { + klog.V(4).Infof("Running two-node pacemaker auth: %s (%s) and %s (%s)", cfg.NodeName1, cfg.NodeIP1, cfg.NodeName2, cfg.NodeIP2) command = fmt.Sprintf("/usr/sbin/pcs host auth %s addr=%s %s addr=%s --token %s --debug", cfg.NodeName1, cfg.NodeIP1, cfg.NodeName2, cfg.NodeIP2, TokenPath) } else { + klog.V(4).Infof("Running single-node pacemaker auth: %s (%s)", cfg.NodeName1, cfg.NodeIP1) command = fmt.Sprintf("/usr/sbin/pcs host auth %s addr=%s --token %s --debug", cfg.NodeName1, cfg.NodeIP1, TokenPath) } _, _, err = exec.Execute(ctx, command) diff --git a/pkg/tnf/pkg/tools/jobs.go b/pkg/tnf/pkg/tools/jobs.go index 59d1077bd..66e65c4aa 100644 --- a/pkg/tnf/pkg/tools/jobs.go +++ b/pkg/tnf/pkg/tools/jobs.go @@ -144,7 +144,7 @@ func sanitizeDNSLabel(name string) string { } // ToPascalCase converts kebab-case job names to PascalCase for condition names. -// Capitalizes the first letter of each part, keeping the rest lowercase. +// Capitalizes the first letter of each part, preserving remaining characters as-is. // Special cases "tnf" to be all uppercase "TNF" (acronym). // Examples: // - "tnf-setup-job" → "TNFSetupJob" diff --git a/pkg/tnf/pkg/tools/nodes.go b/pkg/tnf/pkg/tools/nodes.go index c44ce4e53..93dced8be 100644 --- a/pkg/tnf/pkg/tools/nodes.go +++ b/pkg/tnf/pkg/tools/nodes.go @@ -70,11 +70,16 @@ func StringSlicesEqual(a, b []string) bool { // ListNodesFromInformer returns all nodes from the informer. // Returns only nodes matching the informer's filter (e.g., controlPlaneNodeInformer). +// Returns error if informer hasn't synced yet (helps callers distinguish unsynced cache from empty node list). func ListNodesFromInformer(informer cache.SharedIndexInformer) ([]*corev1.Node, error) { if informer == nil { return nil, fmt.Errorf("informer is nil") } + if !informer.HasSynced() { + return nil, fmt.Errorf("informer has not synced yet") + } + lister := corev1listers.NewNodeLister(informer.GetIndexer()) return lister.List(labels.Everything()) } diff --git a/pkg/tnf/update-setup/runner.go b/pkg/tnf/update-setup/runner.go index ef8bcbfbd..ee1632d99 100644 --- a/pkg/tnf/update-setup/runner.go +++ b/pkg/tnf/update-setup/runner.go @@ -10,6 +10,7 @@ import ( operatorv1 "github.com/openshift/api/operator/v1" "github.com/openshift/library-go/pkg/operator/genericoperatorclient" + "k8s.io/apimachinery/pkg/util/validation" "k8s.io/apiserver/pkg/server" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" @@ -108,8 +109,55 @@ func RunTnfUpdateSetup() error { offlineNodeName := strings.TrimSpace(stdOut) if offlineNodeName == "" { - klog.Info("No offline node found, nothing to do") - return nil + klog.Info("No offline node found, checking if cluster has correct number of nodes configured") + + // Check if Pacemaker cluster has exactly 2 nodes configured + // This catches the case where a node was removed from Pacemaker but not re-added + // (e.g., update-setup job failed after node removal) + command = "/usr/sbin/pcs status xml" + stdOut, stdErr, err = exec.Execute(ctx, command) + if err != nil { + klog.Errorf("Failed to query cluster status: %s, stdout: %s, stderr: %s, err: %v", command, stdOut, stdErr, err) + return fmt.Errorf("failed to check cluster node count: %w", err) + } + + var result pacemaker.PacemakerResult + if parseErr := xml.Unmarshal([]byte(stdOut), &result); parseErr != nil { + klog.Errorf("Failed to parse pcs status xml: %v", parseErr) + return fmt.Errorf("failed to parse cluster status: %w", parseErr) + } + + // Count total nodes configured (online or offline) + totalNodes := len(result.Nodes.Node) + if totalNodes == 2 { + klog.Info("Cluster has 2 nodes configured, nothing to do") + return nil + } + + // Cluster missing a node - determine which one and add it back + klog.Warningf("Cluster has %d nodes configured (expected 2), determining missing node", totalNodes) + + // Build set of nodes in Pacemaker + pacemakerNodes := make(map[string]bool) + for _, node := range result.Nodes.Node { + pacemakerNodes[node.Name] = true + } + + // Determine which K8s node is missing from Pacemaker + var missingNodeName string + if !pacemakerNodes[cfg.NodeName1] { + missingNodeName = cfg.NodeName1 + } else if !pacemakerNodes[cfg.NodeName2] { + missingNodeName = cfg.NodeName2 + } else { + // Should never happen - totalNodes != 2 but both K8s nodes are in Pacemaker + return fmt.Errorf("cluster has %d nodes but both K8s nodes (%s, %s) are in Pacemaker - unexpected state", totalNodes, cfg.NodeName1, cfg.NodeName2) + } + + klog.Infof("Node %q is missing from Pacemaker cluster, adding it back", missingNodeName) + + // Add missing node back to cluster (skip remove step - already removed) + return addNodeBackToCluster(ctx, kubeClient, missingNodeName, currentNodeName, cfg) } klog.Infof("Current node: %q (IP: %s), Other node: %q (IP: %s), Offline node: %q", currentNodeName, currentNodeIP, otherNodeName, otherNodeIP, offlineNodeName) @@ -133,19 +181,52 @@ func RunTnfUpdateSetup() error { return err } - // update fence devices - // this is needed for being able to start resources on the new node! - // node order matters here: resources can't be restarted while fencing isn't configured on all nodes! - err = pcs.ConfigureFencing(ctx, kubeClient, []string{otherNodeName, currentNodeName}) + // Reconfigure cluster after node change (fencing, etcd, validation) + return reconfigureClusterAfterNodeChange(ctx, kubeClient, currentNodeName, cfg) +} + +// addNodeBackToCluster adds a missing node back to the Pacemaker cluster. +// This handles the recovery case where a node was removed from Pacemaker but not re-added +// (e.g., update-setup job failed after node removal). +func addNodeBackToCluster(ctx context.Context, kubeClient kubernetes.Interface, missingNodeName string, currentNodeName string, cfg config.ClusterConfig) error { + klog.Infof("Adding node %q back to Pacemaker cluster", missingNodeName) + + // Validate node name before constructing command (defense-in-depth against bad data in PacemakerCluster CR) + if errs := validation.IsDNS1123Label(missingNodeName); len(errs) > 0 { + return fmt.Errorf("invalid node name %q: %v", missingNodeName, errs) + } + + // Add missing node to cluster configuration + // Use --force to override warning about existing cluster config files on the node + // (the node was removed from cluster but retains config files) + command := fmt.Sprintf("/usr/sbin/pcs cluster node add %s --force", missingNodeName) + stdOut, stdErr, err := exec.Execute(ctx, command) + if err != nil { + klog.Errorf("Failed to add node to cluster: %s, stdout: %s, stderr: %s, err: %v", command, stdOut, stdErr, err) + return fmt.Errorf("failed to add missing node %s: %w", missingNodeName, err) + } + klog.Infof("Successfully executed: %s", command) + + // Reconfigure cluster after node change (fencing, etcd, validation) + return reconfigureClusterAfterNodeChange(ctx, kubeClient, currentNodeName, cfg) +} + +// reconfigureClusterAfterNodeChange handles the common post-node-change operations: +// configures fencing, updates etcd resource, removes unstarted members, starts cluster, validates. +// Called by both the offline-node path (after remove+add) and missing-node path (after add). +func reconfigureClusterAfterNodeChange(ctx context.Context, kubeClient kubernetes.Interface, currentNodeName string, cfg config.ClusterConfig) error { + // Update fence devices (both nodes in correct order) + // Node order matters: resources can't be restarted while fencing isn't configured on all nodes + err := pcs.ConfigureFencing(ctx, kubeClient, []string{cfg.NodeName1, cfg.NodeName2}) if err != nil { klog.Error(err, "Failed to configure fencing, skipping update of etcd! Restart update-setup job when fencing config is fixed!") return err } - commands = []string{ - // Force new cluster on next etcd restart on this node + commands := []string{ + // Force new cluster on next etcd restart on current node fmt.Sprintf("crm_attribute --lifetime reboot --node %s --name \"force_new_cluster\" --update %s", currentNodeName, currentNodeName), - // Update etcd resource + // Update etcd resource with correct node IP map fmt.Sprintf("/usr/sbin/pcs resource update etcd node_ip_map=\"%s:%s;%s:%s\" --wait=300", cfg.NodeName1, cfg.NodeIP1, cfg.NodeName2, cfg.NodeIP2), } err = runCommands(ctx, commands) @@ -153,30 +234,31 @@ func RunTnfUpdateSetup() error { return err } - // remove old node from etcd members - command = "podman exec etcd /usr/bin/etcdctl member list | grep unstarted | awk -F, '{ print $1 }'" - stdOut, stdErr, err = exec.Execute(ctx, command) + // Remove unstarted etcd member if present + command := "podman exec etcd /usr/bin/etcdctl member list | grep unstarted | awk -F, '{ print $1 }'" + stdOut, stdErr, err := exec.Execute(ctx, command) if err != nil { klog.Errorf("Failed to find unstarted etcd member: %s, stdout: %s, stderr: %s, err: %v", command, stdOut, stdErr, err) } else { unstartedMemberID := strings.TrimSpace(stdOut) - command = fmt.Sprintf("podman exec etcd /usr/bin/etcdctl member remove %s", unstartedMemberID) - stdOut, stdErr, err = exec.Execute(ctx, command) - if err != nil { - klog.Errorf("Failed to remove unstarted etcd member: %s, stdout: %s, stderr: %s, err: %v", command, stdOut, stdErr, err) - return err + if unstartedMemberID != "" { + command = fmt.Sprintf("podman exec etcd /usr/bin/etcdctl member remove %s", unstartedMemberID) + stdOut, stdErr, err = exec.Execute(ctx, command) + if err != nil { + klog.Errorf("Failed to remove unstarted etcd member: %s, stdout: %s, stderr: %s, err: %v", command, stdOut, stdErr, err) + return err + } + klog.Infof("Removed unstarted etcd member: %s", unstartedMemberID) } - klog.Infof("Removed unstarted etcd member: %s", unstartedMemberID) } - // wait a bit for things to settle - // without this the etcd start on the new node fails for some reason... + // Wait for cluster to settle time.Sleep(10 * time.Second) commands = []string{ - // Enable cluster on new node + // Enable cluster on all nodes "/usr/sbin/pcs cluster enable --all", - // Start cluster on new node + // Start cluster on all nodes "/usr/sbin/pcs cluster start --all", } err = runCommands(ctx, commands) @@ -188,12 +270,16 @@ func RunTnfUpdateSetup() error { klog.Info("Waiting for cluster to stabilize...") time.Sleep(10 * time.Second) - // Validate final cluster state: must have exactly 2 nodes - // This ensures we don't succeed if auth hasn't run on the new node yet - // or if node add/remove operations didn't complete correctly + // Validate final cluster state + return validateClusterState(ctx) +} + +// validateClusterState validates that the Pacemaker cluster has exactly 2 nodes online. +// Returns an error if validation fails. +func validateClusterState(ctx context.Context) error { klog.Info("Validating final cluster configuration...") - command = "/usr/sbin/pcs status xml" - stdOut, stdErr, err = exec.Execute(ctx, command) + command := "/usr/sbin/pcs status xml" + stdOut, stdErr, err := exec.Execute(ctx, command) if err != nil { klog.Errorf("Failed to query cluster status: %s, stdout: %s, stderr: %s, err: %v", command, stdOut, stdErr, err) return fmt.Errorf("failed to validate cluster state: %w", err)