nf-k8s: Handle node disruption as an infrastructure failure in Pod and Job mode

plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy

@@ -395,20 +395,18 @@ class K8sClient {
             try {
                 return podState(podName)
             }
-            /* pod might be deleted by control plane just after findPodNameForJob() call
-             * so try fallback to jobState
-             */
+            // pod gone (evicted or cleaned up by the control plane): fall back to the Job
+            // status, passing the exception so a node termination can be re-surfaced
             catch (NodeTerminationException err) {
-                log.warn1("Job $jobName's Pod not found, probably cleaned by controlplane")
-                return jobStateFallback0(jobName)
+                return jobStateFallback0(jobName, err)
             }
         }
         else {
             return jobStateFallback0(jobName)
         }
     }
 
-    protected Map jobStateFallback0(String jobName) {
+    protected Map jobStateFallback0(String jobName, NodeTerminationException original=null) {
         final K8sResponseJson jobResp = jobStatus(jobName)
         final jobStatus = jobResp.status as Map
         if( jobStatus?.succeeded == 1 && jobStatus.conditions instanceof List ) {
@@ -431,6 +429,11 @@ class K8sClient {
         }
 
         if( jobStatus?.failed && (int)(jobStatus.failed) > 0 ) {
+            // re-surface a node termination as an infrastructure failure rather than the Job's
+            // generic `backoffLimit` failure: the disruption signal lived on the (now gone) pod
+            // and cannot be recovered from the Job status alone
+            if( original != null )
+                throw original
             String message = 'unknown'
             if( jobStatus.conditions instanceof List ) {
                 final allConditions = jobStatus.conditions as List<Map>
@@ -474,6 +477,15 @@ class K8sClient {
         final status = resp.status as Map
         final containerStatuses = status?.containerStatuses as List<Map>
 
+        // best-effort detection of an involuntary node disruption (e.g. Spot/preemptible node
+        // preemption or graceful node shutdown), signalled by the `DisruptionTarget` pod condition.
+        // When present -- and the container has not already exited successfully -- surface it as a
+        // node termination so that Nextflow retries the task. This complements the `Shutdown` and
+        // pod-not-found (404) detections, which do not fire reliably for graceful preemptions.
+        // See https://kubernetes.io/docs/concepts/workloads/pods/pod-condition/
+        if( isNodeDisruption(status) && !isSuccessfullyTerminated(containerStatuses) )
+            throw new NodeTerminationException("K8s pod '$podName' was terminated due to a node disruption event")
+
         if( containerStatuses?.size()>0 ) {
             final container = containerStatuses.get(0)
             // note: when the pod is created by a Job submission
@@ -523,6 +535,37 @@ class K8sClient {
         throw new K8sResponseException("K8s undetermined status conditions for pod $podName", resp)
     }
 
+    /**
+     * Determine whether the pod status carries a `DisruptionTarget` condition set to `True`,
+     * which K8s adds when a pod is about to be deleted due to an involuntary disruption such
+     * as node preemption or graceful node shutdown.
+     *
+     * @param status The pod `status` object
+     * @return {@code true} when a `DisruptionTarget` condition with status `True` is present
+     */
+    protected static boolean isNodeDisruption(Map status) {
+        final conditions = status?.conditions
+        if( !(conditions instanceof List) )
+            return false
+        for( Object it : (List) conditions ) {
+            if( it instanceof Map && it.type == 'DisruptionTarget' && it.status == 'True' )
+                return true
+        }
+        return false
+    }
+
+    /**
+     * @param containerStatuses The pod `containerStatuses` list
+     * @return {@code true} when the first container has already terminated with exit code {@code 0}
+     */
+    protected static boolean isSuccessfullyTerminated(List<Map> containerStatuses) {
+        if( !containerStatuses )
+            return false
+        final state = containerStatuses.get(0)?.state as Map
+        final terminated = state?.terminated as Map
+        return terminated != null && (terminated.exitCode as Integer) == 0
+    }
+
     protected void checkInvalidWaitingState( Map waiting, K8sResponseJson resp ) {
         if( waiting.reason == 'ErrImagePull' || waiting.reason == 'ImagePullBackOff') {
             def message = "K8s pod image cannot be pulled"

plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy

@@ -24,6 +24,7 @@ import nextflow.exception.K8sOutOfMemoryException
 import javax.net.ssl.HttpsURLConnection
 
 import nextflow.exception.NodeTerminationException
+import nextflow.exception.ProcessFailedException
 import spock.lang.Specification
 /**
  *
@@ -894,6 +895,107 @@ class K8sClientTest extends Specification {
 
     }
 
+    def 'should throw node termination on DisruptionTarget condition' () {
+        given:
+        def JSON = '''
+             {
+              "kind": "Pod",
+              "apiVersion": "v1",
+              "metadata": {
+                  "name": "nf-disrupt",
+                  "namespace": "default"
+              },
+              "status": {
+                  "phase": "Running",
+                  "conditions": [
+                      { "type": "PodScheduled", "status": "True" },
+                      { "type": "DisruptionTarget", "status": "True", "reason": "TerminationByKubelet",
+                        "message": "Pod was terminated in response to imminent node shutdown" }
+                  ],
+                  "containerStatuses": [
+                      {
+                          "name": "nf-disrupt",
+                          "state": { "terminated": { "exitCode": 143, "reason": "Error" } }
+                      }
+                  ]
+              }
+            }
+'''
+        def client = Spy(K8sClient)
+        final POD_NAME = 'nf-disrupt'
+
+        when:
+        client.podState(POD_NAME)
+        then:
+        1 * client.podStatus(POD_NAME) >> new K8sResponseJson(JSON)
+        and:
+        def e = thrown(NodeTerminationException)
+        e.message == "K8s pod 'nf-disrupt' was terminated due to a node disruption event"
+    }
+
+    def 'should not throw node termination when DisruptionTarget pod terminated successfully' () {
+        given:
+        def JSON = '''
+             {
+              "kind": "Pod",
+              "apiVersion": "v1",
+              "metadata": {
+                  "name": "nf-disrupt",
+                  "namespace": "default"
+              },
+              "status": {
+                  "phase": "Succeeded",
+                  "conditions": [
+                      { "type": "DisruptionTarget", "status": "True", "reason": "TerminationByKubelet" }
+                  ],
+                  "containerStatuses": [
+                      {
+                          "name": "nf-disrupt",
+                          "state": { "terminated": { "exitCode": 0, "reason": "Completed" } }
+                      }
+                  ]
+              }
+            }
+'''
+        def client = Spy(K8sClient)
+        final POD_NAME = 'nf-disrupt'
+
+        when:
+        def result = client.podState(POD_NAME)
+        then:
+        1 * client.podStatus(POD_NAME) >> new K8sResponseJson(JSON)
+        and:
+        result == [terminated: [exitCode: 0, reason: 'Completed']]
+    }
+
+    def 'should detect DisruptionTarget node disruption condition' () {
+        expect:
+        K8sClient.isNodeDisruption(status) == expected
+        where:
+        status                                                                                          | expected
+        [conditions: [[type: 'DisruptionTarget', status: 'True', reason: 'TerminationByKubelet']]]      | true
+        [conditions: [[type: 'Ready', status: 'True'], [type: 'DisruptionTarget', status: 'True']]]     | true
+        [conditions: [[type: 'DisruptionTarget', status: 'False']]]                                     | false
+        [conditions: [[type: 'PodScheduled', status: 'True']]]                                          | false
+        [conditions: []]                                                                                | false
+        [phase: 'Running']                                                                              | false
+        [:]                                                                                             | false
+        null                                                                                            | false
+    }
+
+    def 'should detect a successful container termination' () {
+        expect:
+        K8sClient.isSuccessfullyTerminated(containerStatuses) == expected
+        where:
+        containerStatuses                                       | expected
+        [[state: [terminated: [exitCode: 0]]]]                  | true
+        [[state: [terminated: [exitCode: 143]]]]                | false
+        [[state: [running: [:]]]]                               | false
+        [[state: [:]]]                                          | false
+        []                                                      | false
+        null                                                    | false
+    }
+
     def 'client should fail when config fail' () {
         given:
         def JSON = '''
@@ -1105,6 +1207,107 @@ class K8sClientTest extends Specification {
         result.terminated.exitcode == null
     }
 
+    def 'should re-throw node termination when job failed and original exception is present' () {
+        given:
+        def JOB_STATUS_JSON = '''
+        {
+            "apiVersion": "batch/v1",
+            "kind": "Job",
+            "metadata": {
+                "name": "test-job"
+            },
+            "status": {
+                "failed": 1,
+                "conditions": [
+                    {
+                        "type": "Failed",
+                        "status": "True",
+                        "reason": "BackoffLimitExceeded",
+                        "message": "Job has reached the specified backoff limit"
+                    }
+                ]
+            }
+        }
+        '''
+        def client = Spy(K8sClient)
+        final JOB_NAME = 'test-job'
+        final original = new NodeTerminationException('Pod terminated by node disruption')
+
+        when:
+        client.jobStateFallback0(JOB_NAME, original)
+
+        then:
+        1 * client.jobStatus(JOB_NAME) >> new K8sResponseJson(JOB_STATUS_JSON)
+
+        and:
+        def e = thrown(NodeTerminationException)
+        e.is(original)
+    }
+
+    def 'should throw process failed exception when job failed and no original exception is present' () {
+        given:
+        def JOB_STATUS_JSON = '''
+        {
+            "apiVersion": "batch/v1",
+            "kind": "Job",
+            "metadata": {
+                "name": "test-job"
+            },
+            "status": {
+                "failed": 1,
+                "conditions": [
+                    {
+                        "type": "Failed",
+                        "status": "True",
+                        "reason": "BackoffLimitExceeded",
+                        "message": "Job has reached the specified backoff limit"
+                    }
+                ]
+            }
+        }
+        '''
+        def client = Spy(K8sClient)
+        final JOB_NAME = 'test-job'
+
+        when:
+        client.jobStateFallback0(JOB_NAME)
+
+        then:
+        1 * client.jobStatus(JOB_NAME) >> new K8sResponseJson(JOB_STATUS_JSON)
+
+        and:
+        def e = thrown(ProcessFailedException)
+        e.message == "K8s Job test-job execution failed: Job has reached the specified backoff limit"
+    }
+
+    def 'should return empty map when job has no pods scheduled yet' () {
+        given:
+        def JOB_STATUS_JSON = '''
+        {
+            "apiVersion": "batch/v1",
+            "kind": "Job",
+            "metadata": {
+                "name": "test-job"
+            },
+            "status": {
+                "active": 0
+            }
+        }
+        '''
+        def client = Spy(K8sClient)
+        final JOB_NAME = 'test-job'
+
+        when:
+        def result = client.jobStateFallback0(JOB_NAME)
+
+        then:
+        1 * client.jobStatus(JOB_NAME) >> new K8sResponseJson(JOB_STATUS_JSON)
+
+        and:
+        result != null
+        result.isEmpty()
+    }
+
     def 'should re-read token from disk and retry on 401 when tokenPath is set' () {
 
         given: