feat: noise runtime estimator

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

@@ -248,6 +248,18 @@ class K8sConfig implements ConfigScope {
     """)
     final String schedulingStrategy
 
+    @ConfigOption
+    @Description("""
+        Selects a runtime estimator
+    """)
+    final String runtimeEstimator
+
+    @ConfigOption
+    @Description("""
+        Noise magnitude used for the noise runtime estimator
+    """)
+    final Duration noiseRuntimeEstimatorNoiseMagnitude
+
     /* required by extension point -- do not remove */
     K8sConfig() {
         this(Collections.emptyMap())
@@ -287,6 +299,8 @@ class K8sConfig implements ConfigScope {
         runtimeRecordPath = opts.runtimeRecordPath as String ?: "${workDir}/runtimes.csv"
 
         schedulingStrategy = opts.schedulingStrategy as String ?: "Hash"
+        runtimeEstimator = opts.runtimeEstimator as String ?: "LinearFit"
+        noiseRuntimeEstimatorNoiseMagnitude = opts.noiseRuntimeEstimatorNoiseMagnitude as Duration ?: new Duration(10, TimeUnit.SECONDS)
 
         // -- shortcut to pod image pull-policy
         if( imagePullPolicy )

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

@@ -97,7 +97,15 @@ class K8sExecutor extends Executor implements ExtensionPoint {
         log.debug "[K8s] config=$k8sConfig; API client config=$client.config"
 
         this.runtimeRecorder = new K8sRuntimeRecorder(k8sConfig.recordTaskRuntimes, k8sConfig.runtimeRecordPath)
-        this.runtimeEstimator = new K8sRuntimeEstimator(k8sConfig.runtimeRecordPath)
+
+        if (k8sConfig.runtimeEstimator == "LinearFit") {
+            this.runtimeEstimator = new K8sLinearFitRuntimeEstimator(k8sConfig.runtimeRecordPath)
+        } else if (k8sConfig.runtimeEstimator == "Noise") {
+            this.runtimeEstimator = new K8sNoiseRuntimeEstimator(k8sConfig.runtimeEstimator, k8sConfig.noiseRuntimeEstimatorNoiseMagnitude)
+        } else {
+            log.error "[K8s] invalid runtime estimator ${k8sConfig.runtimeEstimator} - falling back on \"LinearFit\""
+            this.runtimeEstimator = new K8sLinearFitRuntimeEstimator(k8sConfig.runtimeRecordPath)
+        }
 
         String[] nodes = getNodeList()
 
@@ -113,6 +121,17 @@ class K8sExecutor extends Executor implements ExtensionPoint {
             strategy = new K8sDVFSSchedulingStrategy(this.runtimeEstimator,
                 new K8sDVFSClient(nodes, ips),
                 () -> getClient())
+            strategy.fullSpeedMode = false
+        } else if (k8sConfig.schedulingStrategy == "DVFS-SPEED") {
+            String[] ips = new String[nodes.length]
+            for (int i = 0; i < nodes.length; i++) {
+                ips[i] = client.getPodIpAddress(K8sNodeInitDeployer.buildPodName(nodes[i]))
+                log.info "[K8s] node ${nodes[i]} -> ${ips[i]}"
+            }
+            strategy = new K8sDVFSSchedulingStrategy(this.runtimeEstimator,
+                new K8sDVFSClient(nodes, ips),
+                () -> getClient())
+            strategy.fullSpeedMode = true
         } else {
             log.error "[K8s] invalid scheduling strategy $k8sConfig.schedulingStrategy, falling back on \"Hash\""
             strategy = new K8sHashSchedulingStrategy()

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

@@ -0,0 +1,99 @@
+package nextflow.k8s
+
+import groovy.util.logging.Slf4j
+
+import java.nio.file.Files
+import java.nio.file.Path
+
+/**
+ * Provides runtime estimates for tasks based on task-name and input file size
+ */
+@Slf4j
+class K8sLinearFitRuntimeEstimator extends K8sRuntimeEstimator {
+
+    class Function {
+        private double m
+        private double n
+
+        Function(double m, double n) {
+            this.m = m
+            this.n = n
+        }
+
+        double estimate(long x) {
+            return m * (double)x + n
+        }
+    }
+
+    HashMap<String, Function> estimators;
+
+    /**
+     * Initialize the estimator with data recorded by K8sRuntimeRecorder
+     * @param dataFilePath
+     */
+    K8sLinearFitRuntimeEstimator(String dataFilePath) {
+        def data = parseDataFile(dataFilePath)
+        fit(data)
+    }
+
+    /**
+     * Initializes the runtime estimator with statically known data.
+     * @param data Map from task name to list of recordings, where each recording is a tuple (input-size, runtime-in-ms)
+     */
+    K8sLinearFitRuntimeEstimator(HashMap<String, ArrayList<Tuple2<Long, Long>>> data) {
+        fit(data)
+    }
+
+    /**
+     * Returns an estimation of the task runtime in milliseconds
+     * @param taskName the name of the task
+     * @param inputSize the total input size in bytes
+     * @return estimated task runtime in milliseconds OR infinity if the task is unknown.
+     */
+    double estimate(String taskName, long inputSize) {
+        Function estimator = estimators.get(taskName)
+        if (estimator == null) {
+            //log.warn "[K8s] Unable to estimate take ${taskName}: unknown task"
+            return Double.POSITIVE_INFINITY
+        }
+        return estimator.estimate(inputSize)
+    }
+
+    private void fit(HashMap<String, ArrayList<Tuple2<Long, Long>>> data) {
+        estimators = new HashMap<>()
+        for (Map.Entry<String, ArrayList<Tuple2<Long, Long>>> t : data) {
+            Function f = fit(t.value)
+            estimators.put(t.key, f)
+        }
+    }
+
+    /**
+     * Uses linear regression to fit a linear function (y = m * x + n) to the provided observations
+     * @param observations list of tuples (input size, runtime in ms)
+     * @return linear function fitted to the input
+     */
+    private Function fit(ArrayList<Tuple2<Long, Long>> observations) throws IllegalArgumentException {
+        int n = observations.size()
+        if (n > 1) {
+            double sumX = 0, sumY = 0, sumXY = 0, sumX2 = 0
+            for (Tuple2<Long, Long> o : observations) {
+                double x = (double) o.get(0)
+                double y = (double) o.get(1)
+                sumX += x
+                sumY += y
+                sumXY += x * y
+                sumX2 += x * x
+            }
+            double m = (n * sumXY - sumX * sumY) / (n * sumX2 - sumX * sumX)
+            return new Function(
+                m,
+                (sumY - m * sumX) / n
+            )
+        } else if (n == 1) {
+            // Special case: We only have 1 measurement. We will just assume that the runtime is constant,
+            // because in our observed data, it is.
+            return new Function(0.0, (double) observations[0].get(1))
+        }
+        throw new IllegalArgumentException("requires at least 1 observation")
+    }
+}

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

@@ -0,0 +1,50 @@
+package nextflow.k8s
+
+import nextflow.util.Duration
+
+class K8sNoiseRuntimeEstimator extends K8sRuntimeEstimator {
+    // Magnitude of the noise we add
+    private long noiseMag
+
+    private HashMap<String, ArrayList<Tuple2<Long, Long>>> data;
+    private synchronized Random rnd;
+
+    /**
+     * Initialize the estimator with data recorded by K8sRuntimeRecorder
+     * @param dataFilePath
+     * @param noiseMagnitude magnitude of the noise added to recordings
+     */
+    K8sNoiseRuntimeEstimator(String dataFilePath, Duration noiseMagnitude) {
+        this.data = parseDataFile(dataFilePath)
+        this.noiseMag = noiseMagnitude.toMillis()
+        this.rnd = new Random()
+    }
+
+    /**
+     * Initializes the runtime estimator with statically known data.
+     * @param data Map from task name to list of recordings, where each recording is a tuple (input-size, runtime-in-ms)
+     */
+    K8sNoiseRuntimeEstimator(HashMap<String, ArrayList<Tuple2<Long, Long>>> data, Duration noiseMagnitude) {
+        this.data = data;
+        this.noiseMag = noiseMagnitude.toMillis()
+    }
+
+    /**
+     * Returns an estimation of the task runtime in milliseconds
+     * @param taskName the name of the task
+     * @param inputSize the total input size in bytes
+     * @return estimated task runtime in milliseconds OR infinity if the task is unknown.
+     */
+    @Override
+    double estimate(String taskName, long inputSize) {
+        ArrayList recordings = data.get(taskName)
+        for (Tuple2<Long, Long> recording : recordings) {
+            if (recording.get(0).longValue() == inputSize) {
+                double noise = rnd.nextDouble(-1.0, 1.0) * (double)noiseMag
+                long ms = recording.get(1).longValue()
+                return (double)ms + noise
+            }
+        }
+        return 0
+    }
+}

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

@@ -9,29 +9,48 @@ import java.nio.file.Path
  * Provides runtime estimates for tasks based on task-name and input file size
  */
 @Slf4j
-class K8sRuntimeEstimator {
+abstract class K8sRuntimeEstimator {
-
+    /**
-    class Function {
+     * Returns an estimation of the task runtime in milliseconds
-        private double m
+     * @param handler the task handler
-        private double n
+     * @return estimated task runtime in milliseconds OR infinity if the task is unknown.
-
+     */
-        Function(double m, double n) {
+    double estimate(K8sTaskHandler handler) {
-            this.m = m
+        long x = getTaskHandlerInputSize(handler)
-            this.n = n
+        return estimate(handler.task.processor.name, x)
     }
 
-        double estimate(long x) {
-            return m * (double)x + n
-        }
-    }
-
-    HashMap<String, Function> estimators;
-
     /**
-     * Initialize the estimator with data recorded by K8sRuntimeRecorder
+     * Returns an estimation of the task runtime in milliseconds
-     * @param dataFilePath
+     * @param taskName the name of the task
+     * @param inputSize the total input size in bytes
+     * @return estimated task runtime in milliseconds OR infinity if the task is unknown.
      */
-    K8sRuntimeEstimator(String dataFilePath) {
+    abstract double estimate(String taskName, long inputSize);
+
+    protected long getTaskHandlerInputSize(K8sTaskHandler handler) {
+        long inputSizeSum = 0
+        // File input
+        def inputFiles = handler.task.getInputFilesMap()
+        for (Map.Entry<String, Path> f : inputFiles) {
+            try {
+                inputSizeSum += Files.size(f.value)
+            } catch (IOException ex) {
+                log.error "[K8s] failed to get size of input file ${f.value} of task ${task.task.name}: ${ex.message}"
+            }
+        }
+        // Non file input
+        def inputs = handler.task.getInputs()
+        for (Map.Entry i : inputs) {
+            inputSizeSum += i.value.toString().length()
+        }
+
+        return inputSizeSum
+    }
+
+    /// @brief Parses the runtime recording data file
+    /// @return Map from task name to list of recordings, where each recording is a tuple (input-size, runtime-in-ms)
+    protected HashMap<String, ArrayList<Tuple2<Long, Long>>> parseDataFile(String dataFilePath) {
         HashMap<String, ArrayList<Tuple2<Long, Long>>> data = new HashMap<>();
         try {
             BufferedReader reader = new BufferedReader(new FileReader(dataFilePath))
@@ -60,98 +79,6 @@ class K8sRuntimeEstimator {
         } catch (IOException ex) {
             log.error "[K8s] Failed to load ${dataFilePath}: ${ex.message}"
         }
-
+        return data
-        fit(data)
-    }
-
-    /**
-     * Initializes the runtime estimator with statically known data.
-     * @param data Map from task name to list of recordings, where each recording is a tuple (input-size, runtime-in-ms)
-     */
-    K8sRuntimeEstimator(HashMap<String, ArrayList<Tuple2<Long, Long>>> data) {
-        fit(data)
-    }
-
-    /**
-     * Returns an estimation of the task runtime in milliseconds
-     * @param handler the task handler
-     * @return estimated task runtime in milliseconds OR infinity if the task is unknown.
-     */
-    double estimate(K8sTaskHandler handler) {
-        long x = getTaskHandlerInputSize(handler)
-        return estimate(handler.task.processor.name, x)
-    }
-
-    /**
-     * Returns an estimation of the task runtime in milliseconds
-     * @param taskName the name of the task
-     * @param inputSize the total input size in bytes
-     * @return estimated task runtime in milliseconds OR infinity if the task is unknown.
-     */
-    double estimate(String taskName, long inputSize) {
-        Function estimator = estimators.get(taskName)
-        if (estimator == null) {
-            //log.warn "[K8s] Unable to estimate take ${taskName}: unknown task"
-            return Double.POSITIVE_INFINITY
-        }
-        return estimator.estimate(inputSize)
-    }
-
-    private void fit(HashMap<String, ArrayList<Tuple2<Long, Long>>> data) {
-        estimators = new HashMap<>()
-        for (Map.Entry<String, ArrayList<Tuple2<Long, Long>>> t : data) {
-            Function f = fit(t.value)
-            estimators.put(t.key, f)
-        }
-    }
-
-    /**
-     * Uses linear regression to fit a linear function (y = m * x + n) to the provided observations
-     * @param observations list of tuples (input size, runtime in ms)
-     * @return linear function fitted to the input
-     */
-    private Function fit(ArrayList<Tuple2<Long, Long>> observations) throws IllegalArgumentException {
-        int n = observations.size()
-        if (n > 1) {
-            double sumX = 0, sumY = 0, sumXY = 0, sumX2 = 0
-            for (Tuple2<Long, Long> o : observations) {
-                double x = (double) o.get(0)
-                double y = (double) o.get(1)
-                sumX += x
-                sumY += y
-                sumXY += x * y
-                sumX2 += x * x
-            }
-            double m = (n * sumXY - sumX * sumY) / (n * sumX2 - sumX * sumX)
-            return new Function(
-                m,
-                (sumY - m * sumX) / n
-            )
-        } else if (n == 1) {
-            // Special case: We only have 1 measurement. We will just assume that the runtime is constant,
-            // because in our observed data, it is.
-            return new Function(0.0, (double) observations[0].get(1))
-        }
-        throw new IllegalArgumentException("requires at least 1 observation")
-    }
-
-    private long getTaskHandlerInputSize(K8sTaskHandler handler) {
-        long inputSizeSum = 0
-        // File input
-        def inputFiles = handler.task.getInputFilesMap()
-        for (Map.Entry<String, Path> f : inputFiles) {
-            try {
-                inputSizeSum += Files.size(f.value)
-            } catch (IOException ex) {
-                log.error "[K8s] failed to get size of input file ${f.value} of task ${task.task.name}: ${ex.message}"
-            }
-        }
-        // Non file input
-        def inputs = handler.task.getInputs()
-        for (Map.Entry i : inputs) {
-            inputSizeSum += i.value.toString().length()
-        }
-
-        return inputSizeSum
     }
 }

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

@@ -46,6 +46,7 @@ class K8sRuntimeRecorder {
             inputSizeSum += i.value.toString().length()
         }
 
+        log.info "[K8s] task ${task.task.processor.name} - input ${inputSizeSum} bytes - ran ${runtimeMilis} ms (${task.startTimeMillis} -> ${task.completeTimeMillis})"
         records.add(new K8sRuntimeRecord(task.task.processor.name, inputSizeSum, runtimeMilis))
     }
 

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

@@ -1,9 +1,8 @@
 package nextflow.k8s
 
-import groovy.transform.CompileDynamic
+
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
-import nextflow.k8s.client.K8sClient
 import nextflow.util.Duration
 import java.util.concurrent.LinkedBlockingQueue
 
@@ -32,7 +31,7 @@ class K8sTaskScheduler implements Runnable {
         queue.add(new K8sSchedulingRequest(handler))
         final pending = new ArrayList<K8sSchedulingRequest>(queue)
         if (strategy.scheduleImmediately(this, pending))
-            drain()
+            schedule()
     }
 
     /**
@@ -45,24 +44,24 @@ class K8sTaskScheduler implements Runnable {
         /* If we have pending tasks, now would be a good time to schedule a new one.
          * Because resources were freed right now */
         if (queue.size() > 0)
-            drain()
+            schedule()
     }
 
-    protected synchronized void drain() {
+    protected synchronized void schedule() {
-        while( true ) {
         final pending = new ArrayList<K8sSchedulingRequest>(queue)
         final decision = strategy.schedule(this, pending)
 
         if (!decision)
             return
 
-            if ( !queue.remove(decision.request) )
+        if (!queue.remove(decision.request)) {
-                continue
+            log.warn "[K8s] failed to remove selected task from queue ${decision.request.task.name}"
+            return
+        }
 
         log.info "[K8s] launching queued task ${decision.request.task.name} on node: ${decision.nodeName}"
         decision.request.handler.submitNow(decision.nodeName)
     }
-    }
 
     /* Scheduling Strategy Interface */
     List<String> getNodes() {
@@ -75,9 +74,14 @@ class K8sTaskScheduler implements Runnable {
     void run() {
         this.shouldStop = false
         final interval = this.interval
+        if (interval.toMillis() == 0L) {
+            log.info("[K8s] scheduler loop disabled (interval is 0)")
+            return
+        }
+        log.info("[K8s] launched scheduler loop (${interval.toString()} interval)")
         while (!shouldStop) {
             sleep(interval.toMillis())
-            drain()
+            schedule()
         }
         log.info("[K8s] terminated scheduler loop")
     }

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

@@ -684,9 +684,9 @@ class K8sClient {
                        'kb', 'mb', 'gb', 'tb', 'pb', 'eb',
                        'k', 'm', 'g', 't', 'p', 'e']
 
-        // For CPU, we only care about 'm' (millicores) and 'n' (nanocores) suffixes
+        // For CPU, we only care about 'm' (millicores), 'u' (microcores) and 'n' (nanocores) suffixes
         if (forCpu) {
-            suffixes = ['m', 'n']
+            suffixes = ['m', 'u', 'n']
         }
 
         for (def s : suffixes) {
@@ -711,13 +711,16 @@ class K8sClient {
         }
 
         if (forCpu) {
-            // CPU: handle millicores (m) and nanocores (n) suffixes
+            // CPU: handle millicores (m), microcores (u) and nanocores (n) suffixes
             if (suffix == 'm') {
                 // Already in millicores
                 return (long)(numericValue)
             } else if (suffix == 'n') {
                 // Nanocores: convert to millicores (1 millicore = 1,000,000 nanocores)
                 return (long)(numericValue / 1000000.0)
+            } else if (suffix == 'u') {
+                // Microcores: convert to millicores (1 millicore = 1,000 microcores)
+                return (long)(numericValue / 1000)
             } else {
                 // Default is cores, convert to millicores
                 return (long)(numericValue * 1000)
@@ -853,6 +856,130 @@ class K8sClient {
         return result
     }
 
+    /**
+     * Query the total CPU allocated to pods on a node (sum of pod requests)
+     * @param nodeName The name of the node
+     * @return The allocated CPU in millicores as a Long, or null if unparseable
+     */
+    Long getNodeCpuAllocatedMillis(String nodeName) {
+        assert nodeName
+        try {
+            // List all pods on the node and sum their CPU requests
+            final action = "/api/v1/pods?fieldSelector=spec.nodeName=$nodeName"
+            final resp = get(action)
+            trace('GET', action, resp.text)
+            final podList = new K8sResponseJson(resp.text)
+            final items = podList.items as List<Map>
+            
+            long totalAllocated = 0L
+            if (items) {
+                for (Map pod : items) {
+                    final spec = pod.spec as Map
+                    if (spec) {
+                        final containers = spec.containers as List<Map>
+                        if (containers) {
+                            for (Map container : containers) {
+                                final resources = container.resources as Map
+                                final requests = resources?.requests as Map
+                                if (requests?.cpu) {
+                                    final cpuStr = requests.cpu as String
+                                    final cpuMillis = parseK8sQuantity(cpuStr, true)
+                                    if (cpuMillis != null) {
+                                        totalAllocated += cpuMillis
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+            return totalAllocated
+        }
+        catch (Exception e) {
+            log.warn("Failed to get allocated CPU for node $nodeName: ${e.message}")
+            return null
+        }
+    }
+
+    /**
+     * Query the available CPU on a node (allocatable - allocated)
+     * For scheduling purposes, we need allocated (requests), not usage (from metrics server)
+     * @param nodeName The name of the node
+     * @return The available CPU in millicores as a Long, or null if unparseable
+     */
+    Long getNodeCpuAvailableMillis(String nodeName) {
+        assert nodeName
+        final capacity = getNodeCpuCapacityMillis(nodeName)
+        final allocated = getNodeCpuAllocatedMillis(nodeName)
+        
+        if (capacity == null || allocated == null) {
+            return null
+        }
+        return capacity - allocated
+    }
+
+    /**
+     * Query the total memory allocated to pods on a node (sum of pod requests)
+     * @param nodeName The name of the node
+     * @return The allocated memory in bytes as a Long, or null if unparseable
+     */
+    Long getNodeMemoryAllocatedBytes(String nodeName) {
+        assert nodeName
+        try {
+            // List all pods on the node and sum their memory requests
+            final action = "/api/v1/pods?fieldSelector=spec.nodeName=$nodeName"
+            final resp = get(action)
+            trace('GET', action, resp.text)
+            final podList = new K8sResponseJson(resp.text)
+            final items = podList.items as List<Map>
+            
+            long totalAllocated = 0L
+            if (items) {
+                for (Map pod : items) {
+                    final spec = pod.spec as Map
+                    if (spec) {
+                        final containers = spec.containers as List<Map>
+                        if (containers) {
+                            for (Map container : containers) {
+                                final resources = container.resources as Map
+                                final requests = resources?.requests as Map
+                                if (requests?.memory) {
+                                    final memStr = requests.memory as String
+                                    final memBytes = parseK8sQuantity(memStr, false)
+                                    if (memBytes != null) {
+                                        totalAllocated += memBytes
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+            return totalAllocated
+        }
+        catch (Exception e) {
+            log.warn("Failed to get allocated memory for node $nodeName: ${e.message}")
+            return null
+        }
+    }
+
+    /**
+     * Query the available memory on a node (allocatable - allocated)
+     * For scheduling purposes, we need allocated (requests), not usage (from metrics server)
+     * @param nodeName The name of the node
+     * @return The available memory in bytes as a Long, or null if unparseable
+     */
+    Long getNodeMemoryAvailableBytes(String nodeName) {
+        assert nodeName
+        final capacity = getNodeMemoryCapacityBytes(nodeName)
+        final allocated = getNodeMemoryAllocatedBytes(nodeName)
+        
+        if (capacity == null || allocated == null) {
+            return null
+        }
+        return capacity - allocated
+    }
+
     protected void checkInvalidWaitingState( Map waiting, K8sResponseJson resp ) {
         if( waiting.reason == 'ErrImagePull' || waiting.reason == 'ImagePullBackOff') {
             def message = "K8s pod image cannot be pulled"

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sDVFSSchedulingStrategy.groovy

@@ -34,7 +34,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
     }
 
     private static int getTaskCPURequirement(TaskRun task) {
-        return task.config.hasCpus() ? task.config.getCpus()  : 1
+        return task.config.hasCpus() ? task.config.getCpus() * 1000 : 1000
     }
 
     @Slf4j
@@ -68,26 +68,35 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
         /* Return the number of available (unoccupied) bytes */
         long getAvailableMemory() {
             def k8sClient = clientGetter.getClient()
+            Long available = k8sClient.getNodeMemoryAvailableBytes(this.name)
+            if (available == null) {
+                log.warn "[K8s] failed to retrieve available memory for node ${name}"
+                // Fallback: try capacity - allocated
                 Long capacity = k8sClient.getNodeMemoryCapacityBytes(this.name)
-            Long used = k8sClient.getNodeMemoryUsedBytes(this.name)
+                Long allocated = k8sClient.getNodeMemoryAllocatedBytes(this.name)
-            if (capacity == null || used == null) {
+                if (capacity != null && allocated != null) {
-                log.warn "[K8s] failed to retrieve memory statistics for node ${name}"
+                    return capacity - allocated
+                }
                 return 0
             }
-            return capacity.longValue() - used.longValue()
+            return available
         }
 
         /* Return the number of available (unoccupied) cpu cores */
         long getAvailableCPUs() {
             def k8sClient = clientGetter.getClient()
+            Long available = k8sClient.getNodeCpuAvailableMillis(this.name)
+            if (available == null) {
+                log.warn "[K8s] failed to retrieve available CPU for node ${name}"
+                // Fallback: try capacity - allocated
                 Long capacity = k8sClient.getNodeCpuCapacityMillis(this.name)
-            Long used = k8sClient.getNodeCpuUsedMillis(this.name)
+                Long allocated = k8sClient.getNodeCpuAllocatedMillis(this.name)
-            if (capacity == null || used == null) {
+                if (capacity != null && allocated != null) {
-                log.warn "[K8s] failed to retrieve cpu statistics for node ${name}: capacity ${capacity}, used ${used}"
+                    return capacity - allocated
+                }
                 return 0
             }
-            long availMilis = capacity.longValue() - used.longValue()
+            return available
-            return availMilis.intdiv(1000)
         }
 
         /* Return the total amount of installed memory */
@@ -101,7 +110,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
         long getCPUCount() {
             def k8sClient = clientGetter.getClient()
             Long capacity = k8sClient.getNodeCpuCapacityMillis(this.name)
-            return capacity != null ? capacity.longValue().intdiv(1000) : 0
+            return capacity != null ? capacity.longValue(): 0
         }
 
         // Sets the frequency to the max. requested frequency of all currently running tasks.
@@ -170,6 +179,8 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
 
     private K8sClientGetter clientGetter
 
+    boolean fullSpeedMode
+
     K8sDVFSSchedulingStrategy(K8sRuntimeEstimator runtimeEstimator, K8sDVFSClient dvfsClient, K8sClientGetter clientGetter) {
         this.runtimeEstimator = runtimeEstimator
         this.dvfsClient = dvfsClient
@@ -205,7 +216,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
             final double taskEstimation = runtimeEstimator.estimate(req.handler)
             final boolean isCriticalPath = taskEstimation > averageRuntime
             long frequency = globalMaxFrequency
-            if (!isCriticalPath) {
+            if (!isCriticalPath && !fullSpeedMode) {
                 /* Set frequency so that we expect the runtime to be close to the mean runtime. */
                 frequency = (long)Math.floor((taskEstimation * globalMaxFrequency) / averageRuntime)
                 frequency = Math.max(frequency, globalMinFrequency)
@@ -220,6 +231,10 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
                     return null
                 }
                 /* No node can currently execute this task, but it should be possible in the future */
+                /* log.info "[K8s] ${req.task} can not be scheduled: ${getTaskMemoryRequirment(req.task)} bytes ${getTaskCPURequirement(req.task)} CPUs"
+                for (WorkerNode n : this.nodes) {
+                    log.info "[K8s] node ${n.name} - ${n.availableMemory}, ${n.availableCPUs}"
+                } */
                 continue
             }
 
@@ -236,15 +251,24 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
 
             closest.assignTask(req.task, frequency, dvfsClient)
             taskToNode.put(req.task.hash.toString(), closest)
+            log.info "[K8s] DVFS: Assigned task ${req.task} to node ${closest.name} - ${taskToNode.size()} assigned tasks"
             return new K8sSchedulingDecision(req, closest.name)
         }
 
+        log.info "[K8s] unable to schedule any task. The queue contains ${queue.size()} tasks."
         return null
     }
 
     @Override
     boolean scheduleImmediately(K8sTaskScheduler scheduler, List<K8sSchedulingRequest> queue) {
-        return queue != null && queue.size() > 0 && taskToNode.size() < nodes.size()
+        if (nodes.isEmpty()) {
+            if (!initNodes(scheduler))
+                return false
+        }
+        /* We want to schedule immediately if there are unoccupied nodes */
+        boolean doIt = queue != null && queue.size() > 0 && taskToNode.size() < nodes.size()
+        log.info "[K8s] scheduleImmediately: ${queue.size()} tasks in queue, ${taskToNode.size()} tasks running on ${nodes.size()} nodes: ${doIt}"
+        return doIt
     }
 
     @Override
@@ -267,9 +291,14 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
         } else {
             log.warn "[K8s] no node recorded for task ${task.toString()}"
         }
+
+        log.info "[K8s] task ${task.toString()} finished - ${taskToNode.size()} tasks running"
+        if (node != null) {
+            log.info "[K8s] task ran on node ${node.name} - ${node.availableMemory} bytes ${node.availableCPUs}"
+        }
     }
 
-    private boolean initNodes(K8sTaskScheduler scheduler) {
+    private synchronized boolean initNodes(K8sTaskScheduler scheduler) {
         this.globalMaxFrequency = Long.MAX_VALUE
         final nodes = scheduler.getNodes()
         for (String node : nodes) {
@@ -299,7 +328,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
         final int reqCPUs = getTaskCPURequirement(task)
         ArrayList<WorkerNode> suitableNodes = new ArrayList<>()
         for (WorkerNode n : nodes) {
-            if (n.availableMemory > reqBytes && n.availableCPUs > reqCPUs) {
+            if (n.availableMemory >= reqBytes && n.availableCPUs >= reqCPUs) {
                 log.info "[K8s] task ${task.name}: ${reqBytes} bytes ${reqCPUs} cpus: node ${n.name} has ${n.availableMemory} bytes, ${n.availableCPUs} cpus"
                 suitableNodes.add(n)
             }

test/nextflow.config

@@ -20,13 +20,13 @@ k8s {
 	storageMountPath = '/workspace'
     launchDir = '/workspace/launch'
     projectDir = '/workspace/projects'
-    cleanup = false
+    cleanup = true
 
-	nextflowImage = 'gitea.kleine.eulenhexe.de/kevin/ma/nextflow-dvfs:0.6.29'
+	nextflowImage = 'gitea.kleine.eulenhexe.de/kevin/ma/nextflow-dvfs:0.8.18'
 	imagePullPolicy = 'IfNotPresent'
-	schedulerInterval = '10s'
+	schedulerInterval = '1m'
 	recordTaskRuntimes = true
-	schedulingStrategy = 'DVFS'
+	schedulingStrategy = 'DVFS-SPEED'
 
 	nodeInit {
 		enabled = true