diff --git a/modules/nextflow/src/main/resources/nextflow/executor/command-run.txt b/modules/nextflow/src/main/resources/nextflow/executor/command-run.txt index 5aefcaa57b..a23cdb4073 100644 --- a/modules/nextflow/src/main/resources/nextflow/executor/command-run.txt +++ b/modules/nextflow/src/main/resources/nextflow/executor/command-run.txt @@ -125,10 +125,16 @@ on_exit() { local exit_status=${nxf_main_ret:=0} [[ ${exit_status} -eq 0 && ${nxf_unstage_ret:=0} -ne 0 ]] && exit_status=${nxf_unstage_ret:=0} [[ ${exit_status} -eq 0 && ${last_err} -ne 0 ]] && exit_status=${last_err} - printf -- $exit_status {{exit_file}} set +u - {{cleanup_cmd}} + ## Flush the pending writes *before* publishing the exit status: the `.exitcode` + ## file is the completion marker polled by the executors, therefore it must not + ## become visible to the driver before everything else the task has written into + ## the work directory. On a shared file system with delayed write-back the driver + ## may otherwise observe a task as completed while its outputs are still missing + ## or truncated. {{sync_cmd}} + printf -- $exit_status {{exit_file}} + {{cleanup_cmd}} exit $exit_status } diff --git a/modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy b/modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy index 75e1c1c99f..506101fb14 100644 --- a/modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy @@ -524,6 +524,23 @@ class BashWrapperBuilderTest extends Specification { SysEnv.pop() } + def 'should flush the file system before writing the exit file' () { + given: + SysEnv.push([NXF_ENABLE_FS_SYNC: 'true']) + + when: + def wrapper = newBashWrapperBuilder(workDir: Paths.get('/work/dir')).buildNew0() + then: + // the `.exitcode` file is the completion marker polled by the executors, therefore + // it must be written only after all other task writes have been flushed, otherwise + // a shared file system may expose the marker before the task outputs + wrapper.indexOf('sync || true') > 0 + wrapper.indexOf('sync || true') < wrapper.indexOf('printf -- $exit_status > /work/dir/.exitcode') + + cleanup: + SysEnv.pop() + } + def 'should unstage outputs' () { given: def folder = Paths.get('/work/dir') diff --git a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt b/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt index b766c16f00..fcef9adde4 100644 --- a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt +++ b/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt @@ -281,8 +281,8 @@ on_exit() { local exit_status=${nxf_main_ret:=0} [[ ${exit_status} -eq 0 && ${nxf_unstage_ret:=0} -ne 0 ]] && exit_status=${nxf_unstage_ret:=0} [[ ${exit_status} -eq 0 && ${last_err} -ne 0 ]] && exit_status=${last_err} - printf -- $exit_status > {{folder}}/.exitcode set +u + printf -- $exit_status > {{folder}}/.exitcode exit $exit_status } diff --git a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt b/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt index 28ff124443..bc56898c26 100644 --- a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt +++ b/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt @@ -89,8 +89,8 @@ on_exit() { local exit_status=${nxf_main_ret:=0} [[ ${exit_status} -eq 0 && ${nxf_unstage_ret:=0} -ne 0 ]] && exit_status=${nxf_unstage_ret:=0} [[ ${exit_status} -eq 0 && ${last_err} -ne 0 ]] && exit_status=${last_err} - printf -- $exit_status > {{folder}}/.exitcode set +u + printf -- $exit_status > {{folder}}/.exitcode exit $exit_status } diff --git a/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy b/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy index 2ded21331e..174ee9606f 100644 --- a/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy +++ b/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy @@ -472,10 +472,28 @@ class FileHelper { } /** - * Check if the specified path is a NFS mount + * The file system types that are known to be network shared file systems, and + * therefore may expose a delayed view of the writes made by another client. + * These are the values reported by {@code stat -f -c %T}. + */ + private static final Set SHARED_FS_TYPES = Set.of('nfs', 'lustre', 'ceph', 'beegfs', 'gpfs') + + /** + * Check if the given file system type, as reported by {@code stat -f -c %T}, + * denotes a network shared file system + * + * @param type The file system type name + * @return {@code true} when the type is a shared file system, {@code false} otherwise + */ + static boolean isSharedFsType(String type) { + return type != null && type in SHARED_FS_TYPES + } + + /** + * Check if the specified path is a shared file system mount * * @param path The path to verify - * @return The {@code true} when the path is a NFS mount {@code false} otherwise + * @return The {@code true} when the path is a shared file system mount {@code false} otherwise */ @Memoized static boolean isPathSharedFS(Path path) { @@ -483,8 +501,7 @@ class FileHelper { if( path.getFileSystem() != FileSystems.getDefault() ) return false - final type = getPathFsType(path) - final result = type == 'nfs' || type == 'lustre' + final result = isSharedFsType(getPathFsType(path)) log.debug "FS path type ($result): $path" return result } diff --git a/modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy b/modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy index c95513f1b8..7260473798 100644 --- a/modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy +++ b/modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy @@ -1048,6 +1048,25 @@ class FileHelperTest extends Specification { null | '1234://xyz.com/abc' } + @Unroll + def 'should detect shared file system type'() { + expect: + FileHelper.isSharedFsType(TYPE) == EXPECTED + + where: + TYPE | EXPECTED + 'nfs' | true + 'lustre' | true + 'ceph' | true + 'beegfs' | true + 'gpfs' | true + 'ext2/ext3' | false + 'xfs' | false + 'overlayfs' | false + 'tmpfs' | false + null | false + } + def 'should check symlink status'() { given: def folder = Files.createTempDirectory('test') diff --git a/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy b/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy index 270e20dad6..0a6b179e63 100644 --- a/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy +++ b/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy @@ -32,6 +32,7 @@ import nextflow.exception.NodeTerminationException import nextflow.k8s.client.PodUnschedulableException import nextflow.exception.ProcessSubmitException import nextflow.executor.BashWrapperBuilder +import nextflow.file.FileHelper import nextflow.fusion.FusionAwareTask import nextflow.k8s.client.K8sClient import nextflow.k8s.client.K8sResponseException @@ -78,12 +79,14 @@ class K8sTaskHandler extends TaskHandler implements FusionAwareTask { private Path errorFile - private Path exitFile + protected Path exitFile private Map state private long timestamp + private long terminatedTimestamp + private K8sExecutor executor private String runsOnNode = null @@ -439,6 +442,15 @@ class K8sTaskHandler extends TaskHandler implements FusionAwareTask { // https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#containerstateterminated-v1-core log.trace("[k8s] Container Terminated state ${state.terminated}") final k8sExitCode = (state.terminated as Map)?.exitCode as Integer + // the K8s API can report the pod as terminated before the writes made by the task + // are visible on the node running the driver. The `.exitcode` file is the last file + // written by the task wrapper, therefore waiting for it also guarantees the + // visibility of the task outputs, `.command.env` and `.command.out`. + // The wait is skipped when the API reports a failure, because a task killed by the + // system never writes the exit file and the API status is authoritative anyway + // See https://github.com/nextflow-io/nextflow/issues/7247 + if( (k8sExitCode == null || k8sExitCode == 0) && !isExitFileVisible() ) + return false task.exitStatus = k8sExitCode != null ? k8sExitCode : readExitFile() task.stdout = outputFile task.stderr = errorFile @@ -489,6 +501,55 @@ class K8sTaskHandler extends TaskHandler implements FusionAwareTask { } } + /** + * @return + * {@code true} when the task work directory is hosted by a shared file system + * that may expose a delayed view of the writes made by another node. + * Fusion mounts are excluded because task outputs are uploaded to the object + * storage before the task terminates. + */ + protected boolean isWorkDirSharedFS() { + return !fusionEnabled() && FileHelper.workDirIsSharedFS + } + + protected long getExitReadTimeoutMillis() { + return executor.config.getExitReadTimeout(executor.name).toMillis() + } + + /** + * Check that the task `.exitcode` file is visible on the node running the driver. + * + * The file is written by the task wrapper only after all other task writes have been + * flushed, therefore it acts as a barrier for the whole task work directory. On a + * shared file system with delayed write-back the pod can be reported as terminated + * while the file is still missing or empty; in that case the task completion is + * deferred to a subsequent polling cycle, up to `executor.exitReadTimeout`. + * + * @return + * {@code true} when the exit file is visible, or the wait timed out, + * {@code false} to keep waiting + */ + protected boolean isExitFileVisible() { + if( !isWorkDirSharedFS() ) + return true + + final attrs = FileHelper.readAttributes(exitFile) + if( attrs && attrs.size() > 0 ) + return true + + if( !terminatedTimestamp ) + terminatedTimestamp = System.currentTimeMillis() + + final delta = System.currentTimeMillis() - terminatedTimestamp + if( delta < getExitReadTimeoutMillis() ) { + log.trace "[K8s] Waiting for exit file to become visible for task: `$task.name` -- delta: $delta ms" + return false + } + + log.warn "[K8s] Exit file is still not available after $delta ms for task: `$task.name` -- work dir: ${task.workDir?.toUriString()}" + return true + } + /** * Terminates the current task execution */ diff --git a/plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy b/plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy index 1ecf193bec..bd80685345 100644 --- a/plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy +++ b/plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy @@ -489,6 +489,7 @@ class K8sTaskHandlerTest extends Specification { and: def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME, outputFile: OUT_FILE, errorFile: ERR_FILE)) handler.getClient() >> client + handler.isExitFileVisible() >> true when: def result = handler.checkIfCompleted() @@ -537,6 +538,109 @@ class K8sTaskHandlerTest extends Specification { } + def 'should not complete the task until the exit file is visible' () { + given: + def POD_NAME = 'pod-xyz' + def termState = [ reason: "Completed", + startedAt: "2018-01-13T10:09:36Z", + finishedAt: "2018-01-13T10:19:36Z", + exitCode: 0 ] + def task = new TaskRun() + def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME)) + + when: + def result = handler.checkIfCompleted() + then: + 1 * handler.getState() >> [terminated: termState] + 1 * handler.isExitFileVisible() >> false + 0 * handler.updateTimestamps(_) + 0 * handler.deleteJobIfSuccessful(_) + handler.status != TaskStatus.COMPLETED + result == false + } + + def 'should not wait for the exit file when the pod reports a failure' () { + given: + def POD_NAME = 'pod-xyz' + def client = Mock(K8sClient) + // a task killed by the system (e.g. OOMKilled) never writes the exit file, + // therefore waiting for it would only delay the error report + def termState = [ reason: "OOMKilled", + startedAt: "2018-01-13T10:09:36Z", + finishedAt: "2018-01-13T10:19:36Z", + exitCode: 137 ] + def task = new TaskRun() + def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME)) + handler.getClient() >> client + + when: + def result = handler.checkIfCompleted() + then: + 1 * handler.getState() >> [terminated: termState] + 0 * handler.isExitFileVisible() + 1 * handler.deleteJobIfSuccessful(task) >> null + 1 * handler.saveJobLogOnError(task) >> null + handler.task.exitStatus == 137 + handler.status == TaskStatus.COMPLETED + result == true + } + + def 'should wait for the exit file on a shared file system' () { + given: + def folder = Files.createTempDirectory('test') + def exitFile = folder.resolve('.exitcode') + def task = new TaskRun(name: 'foo', workDir: folder) + def handler = Spy(new K8sTaskHandler(task: task, exitFile: exitFile)) + handler.isWorkDirSharedFS() >> true + handler.getExitReadTimeoutMillis() >> 10_000 + + expect: 'the exit file does not exist yet' + !handler.isExitFileVisible() + + when: 'the exit file exists but is still empty' + Files.createFile(exitFile) + then: + !handler.isExitFileVisible() + + when: 'the exit file holds the exit status' + exitFile.text = '0' + then: + handler.isExitFileVisible() + + cleanup: + folder?.deleteDir() + } + + def 'should give up waiting for the exit file after the read timeout' () { + given: + def folder = Files.createTempDirectory('test') + def exitFile = folder.resolve('.exitcode') + def task = new TaskRun(name: 'foo', workDir: folder) + def handler = Spy(new K8sTaskHandler(task: task, exitFile: exitFile)) + handler.isWorkDirSharedFS() >> true + handler.getExitReadTimeoutMillis() >> 0 + + expect: + handler.isExitFileVisible() + + cleanup: + folder?.deleteDir() + } + + def 'should not wait for the exit file when the work dir is not a shared file system' () { + given: + def folder = Files.createTempDirectory('test') + def task = new TaskRun(name: 'foo', workDir: folder) + def handler = Spy(new K8sTaskHandler(task: task, exitFile: folder.resolve('.exitcode'))) + handler.isWorkDirSharedFS() >> false + + expect: + handler.isExitFileVisible() + + cleanup: + folder?.deleteDir() + } + def 'should use K8s exit code when available' () { given: def ERR_FILE = Paths.get('err.file') @@ -550,6 +654,7 @@ class K8sTaskHandlerTest extends Specification { def task = new TaskRun() def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME, outputFile: OUT_FILE, errorFile: ERR_FILE)) handler.getClient() >> client + handler.isExitFileVisible() >> true when: def result = handler.checkIfCompleted()