From b0a5447797b88157b8b881ad60d29e93ef57df0e Mon Sep 17 00:00:00 2001 From: Kevin Trogant Date: Fri, 21 Aug 2026 12:46:55 +0200 Subject: [PATCH] feat: noise runtime estimator --- .../src/main/nextflow/k8s/K8sConfig.groovy | 14 ++ .../src/main/nextflow/k8s/K8sExecutor.groovy | 21 ++- .../k8s/K8sLinearFitRuntimeEstimator.groovy | 99 ++++++++++++ .../k8s/K8sNoiseRuntimeEstimator.groovy | 50 ++++++ .../nextflow/k8s/K8sRuntimeEstimator.groovy | 151 +++++------------- .../nextflow/k8s/K8sRuntimeRecorder.groovy | 1 + .../main/nextflow/k8s/K8sTaskScheduler.groovy | 36 +++-- .../main/nextflow/k8s/client/K8sClient.groovy | 133 ++++++++++++++- .../K8sDVFSSchedulingStrategy.groovy | 63 ++++++-- test/nextflow.config | 8 +- 10 files changed, 423 insertions(+), 153 deletions(-) create mode 100644 nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sLinearFitRuntimeEstimator.groovy create mode 100644 nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNoiseRuntimeEstimator.groovy diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sConfig.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sConfig.groovy index add4f2e..5261ab9 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sConfig.groovy +++ b/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 ) diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy index 793f798..ceff9b9 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy +++ b/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() diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sLinearFitRuntimeEstimator.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sLinearFitRuntimeEstimator.groovy new file mode 100644 index 0000000..8dfd2b7 --- /dev/null +++ b/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 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>> 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>> data) { + estimators = new HashMap<>() + for (Map.Entry>> 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> observations) throws IllegalArgumentException { + int n = observations.size() + if (n > 1) { + double sumX = 0, sumY = 0, sumXY = 0, sumX2 = 0 + for (Tuple2 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") + } +} diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNoiseRuntimeEstimator.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNoiseRuntimeEstimator.groovy new file mode 100644 index 0000000..fcdc4d1 --- /dev/null +++ b/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>> 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>> 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 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 + } +} diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeEstimator.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeEstimator.groovy index 71a8d6d..86eebc2 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeEstimator.groovy +++ b/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 { - - 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 - } +abstract class K8sRuntimeEstimator { + /** + * 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) } - HashMap estimators; - /** - * Initialize the estimator with data recorded by K8sRuntimeRecorder - * @param dataFilePath + * 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. */ - 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 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>> parseDataFile(String dataFilePath) { HashMap>> 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}" } - - 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>> 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>> data) { - estimators = new HashMap<>() - for (Map.Entry>> 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> observations) throws IllegalArgumentException { - int n = observations.size() - if (n > 1) { - double sumX = 0, sumY = 0, sumXY = 0, sumX2 = 0 - for (Tuple2 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 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 + return data } } diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeRecorder.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeRecorder.groovy index 3d7e4c6..48e9375 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeRecorder.groovy +++ b/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)) } diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy index 55e2c88..6de43b6 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy +++ b/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(queue) if (strategy.scheduleImmediately(this, pending)) - drain() + schedule() } /** @@ -45,23 +44,23 @@ 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() { - while( true ) { - final pending = new ArrayList(queue) - final decision = strategy.schedule(this, pending) + protected synchronized void schedule() { + final pending = new ArrayList(queue) + final decision = strategy.schedule(this, pending) - if ( !decision ) - return + if (!decision) + return - if ( !queue.remove(decision.request) ) - continue - - log.info "[K8s] launching queued task ${decision.request.task.name} on node: ${decision.nodeName}" - decision.request.handler.submitNow(decision.nodeName) + if (!queue.remove(decision.request)) { + 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 */ @@ -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") } diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy index b3d96df..e7e8395 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/client/K8sClient.groovy +++ b/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 + + long totalAllocated = 0L + if (items) { + for (Map pod : items) { + final spec = pod.spec as Map + if (spec) { + final containers = spec.containers as List + 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 + + long totalAllocated = 0L + if (items) { + for (Map pod : items) { + final spec = pod.spec as Map + if (spec) { + final containers = spec.containers as List + 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" diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sDVFSSchedulingStrategy.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sDVFSSchedulingStrategy.groovy index ce2e967..2b14d44 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/strategies/K8sDVFSSchedulingStrategy.groovy +++ b/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 capacity = k8sClient.getNodeMemoryCapacityBytes(this.name) - Long used = k8sClient.getNodeMemoryUsedBytes(this.name) - if (capacity == null || used == null) { - log.warn "[K8s] failed to retrieve memory statistics for node ${name}" + 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 allocated = k8sClient.getNodeMemoryAllocatedBytes(this.name) + if (capacity != null && allocated != null) { + 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 capacity = k8sClient.getNodeCpuCapacityMillis(this.name) - Long used = k8sClient.getNodeCpuUsedMillis(this.name) - if (capacity == null || used == null) { - log.warn "[K8s] failed to retrieve cpu statistics for node ${name}: capacity ${capacity}, used ${used}" + 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 allocated = k8sClient.getNodeCpuAllocatedMillis(this.name) + if (capacity != null && allocated != null) { + return capacity - allocated + } return 0 } - long availMilis = capacity.longValue() - used.longValue() - return availMilis.intdiv(1000) + return available } /* 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 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 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) } diff --git a/test/nextflow.config b/test/nextflow.config index f5383ec..5f28eb1 100644 --- a/test/nextflow.config +++ b/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