diff --git a/plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy b/plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy index 779472eacf..923ebb562d 100644 --- a/plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy +++ b/plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy @@ -395,12 +395,10 @@ 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 { @@ -408,7 +406,7 @@ class K8sClient { } } - 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 @@ -474,6 +477,15 @@ class K8sClient { final status = resp.status as Map final containerStatuses = status?.containerStatuses as List + // 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 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" diff --git a/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy b/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy index ea0458785f..ed16b83a5e 100644 --- a/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy +++ b/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: