Add support for Wave container status API (#5384)

This commit improves the support for Wave:
- add support for container mirror
- add support for container scan mode and levels
- use container status API, in place of build status
- improve the the container provision await logic

Signed-off-by: Paolo Di Tommaso <[email protected]>

modules/nextflow/src/main/groovy/nextflow/container/resolver/ContainerResolver.groovy

@@ -44,6 +44,14 @@ interface ContainerResolver extends ExtensionPoint {
      * @return
      *      The resolved container image name
      */
-    abstract ContainerInfo resolveImage(TaskRun task, String imageName)
+    ContainerInfo resolveImage(TaskRun task, String imageName)
+
+    /**
+     * Check the availability of the specified container reference
+     *
+     * @param key The container reference
+     * @return {@code true} when the container is available for use, {@code false} otherwise
+     */
+    boolean isContainerReady(String key)
 
 }

modules/nextflow/src/main/groovy/nextflow/container/resolver/DefaultContainerResolver.groovy

@@ -55,4 +55,8 @@ class DefaultContainerResolver implements ContainerResolver {
         return new ContainerInfo(imageName, ret, hashKey)
     }
 
+    @Override
+    boolean isContainerReady(String key) {
+        return true
+    }
 }

modules/nextflow/src/main/groovy/nextflow/executor/Executor.groovy

@@ -93,7 +93,7 @@ abstract class Executor {
         /*
          * Add the task to the queue for processing
          * Note: queue is implemented as a fixed size blocking queue, when
-         * there's not space *put* operation will block until, some other tasks finish
+         * there's not space the *put* operation will block until some other tasks finish
          */
         monitor.schedule(handler)
     }

modules/nextflow/src/main/groovy/nextflow/processor/TaskHandler.groovy

@@ -242,6 +242,16 @@ abstract class TaskHandler {
         return !max ? true : task.processor.forksCount < max
     }
 
+    /**
+     * Determine if a task is ready for execution or it depends on resources
+     * e.g. container that needs to be provisionied
+     *
+     * @return {@code true} when the task is ready for execution, {@code false} otherwise
+     */
+    boolean isReady() {
+        task.isContainerReady()
+    }
+
     /**
      * Increment the number of current forked processes
      */

modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy

@@ -191,7 +191,7 @@ class TaskPollingMonitor implements TaskMonitor {
      *      by the polling monitor
      */
     protected boolean canSubmit(TaskHandler handler) {
-        (capacity>0 ? runningQueue.size() < capacity : true) && handler.canForkProcess()
+        (capacity>0 ? runningQueue.size() < capacity : true) && handler.canForkProcess() && handler.isReady()
     }
 
     /**

modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy

@@ -23,12 +23,14 @@ import java.util.concurrent.ConcurrentHashMap
 import java.util.function.Function
 
 import com.google.common.hash.HashCode
+import groovy.transform.Memoized
 import groovy.transform.PackageScope
 import groovy.util.logging.Slf4j
 import nextflow.Session
 import nextflow.conda.CondaCache
 import nextflow.container.ContainerConfig
 import nextflow.container.resolver.ContainerInfo
+import nextflow.container.resolver.ContainerResolver
 import nextflow.container.resolver.ContainerResolverProvider
 import nextflow.exception.ProcessException
 import nextflow.exception.ProcessTemplateException
@@ -326,6 +328,11 @@ class TaskRun implements Cloneable {
      */
     volatile ErrorStrategy errorAction
 
+    /**
+     * Unique key for the container used by this task
+     */
+    volatile String containerKey
+
     TaskConfig config
 
     TaskContext context
@@ -671,6 +678,11 @@ class TaskRun implements Cloneable {
             }})
     }
 
+    @Memoized
+    private ContainerResolver containerResolver() {
+        ContainerResolverProvider.load()
+    }
+
     private ContainerInfo containerInfo0() {
         // fetch the container image from the config
         def configImage = config.getContainer()
@@ -681,8 +693,10 @@ class TaskRun implements Cloneable {
         if( !configImage )
             configImage = null
 
-        final res = ContainerResolverProvider.load()
-        final info = res.resolveImage(this, configImage as String)
+        final info = containerResolver().resolveImage(this, configImage as String)
+        // track the key of the container used
+        this.containerKey = info.hashKey
+        // return the info
         return info
     }
 
@@ -699,6 +713,12 @@ class TaskRun implements Cloneable {
         return info?.hashKey
     }
 
+    boolean isContainerReady() {
+       return containerKey
+            ? containerResolver().isContainerReady(containerKey)
+            : true
+    }
+
     ResourcesBundle getModuleBundle() {
         return this.getProcessor().getModuleBundle()
     }

modules/nextflow/src/test/groovy/nextflow/processor/LocalPollingMonitorTest.groovy

@@ -96,6 +96,7 @@ class LocalPollingMonitorTest extends Specification {
         def handler = Mock(TaskHandler)
         handler.getTask() >> { task }
         handler.canForkProcess() >> true
+        handler.isReady() >> true
 
         expect:
         monitor.canSubmit(handler) == true
@@ -141,6 +142,7 @@ class LocalPollingMonitorTest extends Specification {
         def handler = Mock(TaskHandler)
         handler.getTask() >> { task }
         handler.canForkProcess() >> true
+        handler.isReady() >> true
 
         expect:
         monitor.canSubmit(handler) == true

modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy

@@ -104,6 +104,7 @@ class ParallelPollingMonitorTest extends Specification {
         def result = mon.canSubmit(handler)
         then:
         handler.canForkProcess() >> FORK
+        handler.isReady() >> true
         and:
         result == EXPECTED
 

plugins/nf-wave/build.gradle

@@ -36,8 +36,8 @@ dependencies {
     api 'org.apache.commons:commons-lang3:3.12.0'
     api 'com.google.code.gson:gson:2.10.1'
     api 'org.yaml:snakeyaml:2.2'
-    api 'io.seqera:wave-api:0.11.1'
-    api 'io.seqera:wave-utils:0.12.1'
+    api 'io.seqera:wave-api:0.13.2'
+    api 'io.seqera:wave-utils:0.14.1'
 
     testImplementation(testFixtures(project(":nextflow")))
     testImplementation "org.apache.groovy:groovy:4.0.23"

plugins/nf-wave/src/main/io/seqera/wave/plugin/SubmitContainerTokenRequest.groovy

@@ -21,7 +21,10 @@ package io.seqera.wave.plugin
 import groovy.transform.CompileStatic
 import groovy.transform.EqualsAndHashCode
 import groovy.transform.ToString
+import io.seqera.wave.api.ImageNameStrategy
 import io.seqera.wave.api.PackagesSpec
+import io.seqera.wave.api.ScanLevel
+import io.seqera.wave.api.ScanMode
 
 /**
  * Model a request for an augmented container
@@ -135,4 +138,26 @@ class SubmitContainerTokenRequest {
      */
     PackagesSpec packages
 
+    /**
+     * The strategy applied to name a container build by wave when using
+     * the freeze option.
+     */
+    ImageNameStrategy nameStrategy;
+
+    /**
+     * Whenever use container "mirror" mode
+     */
+    public boolean mirror;
+
+    /**
+     * The request security scan mode
+     */
+    ScanMode scanMode;
+
+    /**
+     * Define the allows security vulnerabilities in the container request.
+     * Empty or null means no vulnerabilities are allowed.
+     */
+    List<ScanLevel> scanLevels
+
 }

plugins/nf-wave/src/main/io/seqera/wave/plugin/SubmitContainerTokenResponse.groovy

@@ -20,6 +20,7 @@ package io.seqera.wave.plugin
 import groovy.transform.CompileStatic
 import groovy.transform.EqualsAndHashCode
 import groovy.transform.ToString
+import io.seqera.wave.api.ContainerStatus
 
 /**
  * Model a response for an augmented container
@@ -30,6 +31,12 @@ import groovy.transform.ToString
 @ToString(includeNames = true, includePackage = false)
 @CompileStatic
 class SubmitContainerTokenResponse {
+
+    /**
+     * Unique Id for this request
+     */
+    String requestId
+
     /**
      * A unique authorization token assigned to this request
      */
@@ -60,4 +67,19 @@ class SubmitContainerTokenResponse {
      */
     Boolean freeze;
 
+    /**
+     * When the result is a mirror container. Version v1alpha2 as later.
+     */
+    Boolean mirror
+
+    /**
+     * The id of the security scan associated with this container
+     */
+    String scanId
+
+    /**
+     * The status of the container for this request
+     */
+    ContainerStatus status
+
 }