feat: simpler test workflow with configuration

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sConfig.groovy

@@ -212,10 +212,15 @@ class K8sConfig implements ConfigScope {
     """)
     final String workDir
 
-    @ConfigOption
     @Description("Node initialization config")
     final K8sNodeInitConfig nodeInit
 
+    @ConfigOption
+    @Description("""
+        The image name of the nextflow launcher image
+    """)
+    final String nextflowImage
+
     /* required by extension point -- do not remove */
     K8sConfig() {
         this(Collections.emptyMap())
@@ -245,6 +250,7 @@ class K8sConfig implements ConfigScope {
         storageMountPath = opts.storageMountPath ?: '/workspace'
         storageSubPath = opts.storageSubPath
         userName = opts.userName
+        nextflowImage = opts.nextflowImage ?: "nextflow/nextflow:${BuildInfo.version}"
 
         launchDir = opts.launchDir ?: "${storageMountPath}/${getUserName()}"
         projectDir = opts.projectDir ?: "${storageMountPath}/projects"
@@ -266,6 +272,7 @@ class K8sConfig implements ConfigScope {
         else if( securityContext )
             pod.securityContext = new PodSecurityContext(securityContext)
 
+        log.info("Hello sailor")
         nodeInit = new K8sNodeInitConfig(opts.nodeInit as Map ?: Collections.emptyMap())
     }
 
@@ -327,7 +334,7 @@ class K8sConfig implements ConfigScope {
     boolean useJobResource() { ResourceType.Job.name() == computeResourceType }
 
     String getNextflowImageName() {
-        return "nextflow/nextflow:${BuildInfo.version}"
+        return nextflowImage
     }
 
     PodOptions getPodOptions() {

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNodeInitDeployer.groovy

@@ -1,9 +1,11 @@
 package nextflow.k8s
 
+import groovy.util.logging.Slf4j
 import nextflow.k8s.client.K8sClient
 import nextflow.k8s.model.PodHostMount
 import nextflow.k8s.model.PodSpecBuilder
 
+@Slf4j
 class K8sNodeInitDeployer {
     private K8sClient client
     private K8sConfig config
@@ -20,12 +22,16 @@ class K8sNodeInitDeployer {
         if ( !init?.enabled )
             return
 
+        log.info("deploying init pods")
+
         final nodes = getNodes()
         for ( String nodeName : nodes ) {
+            log.info(" ... deploying to " + nodeName)
             final spec = makePodSpec(init, nodeName)
             client.podCreate(spec)
         }
 
+        log.info("waiting for init pods")
         waitForPods(nodes)
     }
 

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy

@@ -20,7 +20,10 @@ class K8sTaskScheduler {
     }
 
     void submit(K8sTaskHandler handler) {
+        log.debug "[K8s] received queued task ${handler.task.name}"
         queue.add(new K8sSchedulingRequest(handler))
+
+        drain()
     }
 
     protected synchronized void drain() {
@@ -34,7 +37,7 @@ class K8sTaskScheduler {
             if ( !queue.remove(decision.request) )
                 continue
 
-            log.trace "[K8s] launching queued task ${decision.request.task.name} on node: ${decision.nodeName}"
+            log.debug "[K8s] launching queued task ${decision.request.task.name} on node: ${decision.nodeName}"
             decision.request.handler.submitNow(decision.nodeName)
         }
     }