diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sDVFSClient.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sDVFSClient.groovy index 4f0b292..11b74be 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sDVFSClient.groovy +++ b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sDVFSClient.groovy @@ -5,6 +5,7 @@ import groovy.util.logging.Slf4j import java.net.http.HttpClient import java.net.http.HttpRequest import java.net.http.HttpResponse +import java.nio.channels.UnresolvedAddressException /** * The DVFS client uses the deployed DVFS agents to control the operating frequency of the worker nodes. @@ -12,9 +13,15 @@ import java.net.http.HttpResponse @Slf4j class K8sDVFSClient { private HttpClient httpClient + private Map ipTable - K8sDVFSClient() { + K8sDVFSClient(String[] nodes, String[] ips) { + assert(nodes.length == ips.length) this.httpClient = HttpClient.newBuilder().build() + this.ipTable = new HashMap<>() + for (int i = 0; i < nodes.length; i++) { + this.ipTable.put(nodes[i], ips[i]) + } } OptionalLong getNodeCurrentFrequency(String node) { @@ -62,28 +69,34 @@ class K8sDVFSClient { } OptionalLong getMemoryAmount(String node) { - return getInt(HttpRequest.newBuilder() + return getLong(HttpRequest.newBuilder() .uri(new URI("http://${agentAddress(node)}/mem/amount")) .GET() .build()) } private OptionalLong getLong(HttpRequest request) { - HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()) - if (response.statusCode() != 200) { - log.error("Request GET ${request.uri().toString()} returned ${response.statusCode()}") - return OptionalLong.empty() - } try { + HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()) + if (response.statusCode() != 200) { + log.error("Request GET ${request.uri().toString()} returned ${response.statusCode()}") + return OptionalLong.empty() + } long parsed = Long.parseLong(response.body()) return OptionalLong.of(parsed) } catch (NumberFormatException ex) { log.error("Unexpected response ${response.body()} - ${ex.message}") return OptionalLong.empty() + } catch (ConnectException ex) { + log.error("Request failed ${request.uri().toString()}: ${ex.message}") + return OptionalLong.empty() + } catch (UnresolvedAddressException ex) { + log.error("Failed to resolve address ${request.uri().toString()}: ${ex.message}") + return OptionalLong.empty() } } private String agentAddress(String node) { - return K8sNodeInitDeployer.buildPodName(node) + return ipTable.get(node) + ":8080" } } 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 9a42705..74a2915 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy +++ b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy @@ -99,17 +99,24 @@ class K8sExecutor extends Executor implements ExtensionPoint { this.runtimeRecorder = new K8sRuntimeRecorder(k8sConfig.recordTaskRuntimes, k8sConfig.runtimeRecordPath) this.runtimeEstimator = new K8sRuntimeEstimator(k8sConfig.runtimeRecordPath) + String[] nodes = getNodeList() + K8sSchedulingStrategy strategy = null if (k8sConfig.schedulingStrategy == "Hash") { strategy = new K8sHashSchedulingStrategy() } else if (k8sConfig.schedulingStrategy == "DVFS") { - strategy = new K8sDVFSSchedulingStrategy(this.runtimeEstimator, new K8sDVFSClient()) + 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)) } else { log.error "[K8s] invalid scheduling strategy $k8sConfig.schedulingStrategy, falling back on \"Hash\"" strategy = new K8sHashSchedulingStrategy() } - this.taskScheduler = new K8sTaskScheduler(getNodeList(), strategy) + this.taskScheduler = new K8sTaskScheduler(nodes, strategy, k8sConfig.schedulerInterval) this.schedulerThread = new Thread(this.taskScheduler) this.schedulerThread.start() } diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNodeInitDeployer.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNodeInitDeployer.groovy index 6df9683..020ed12 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNodeInitDeployer.groovy +++ b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sNodeInitDeployer.groovy @@ -136,6 +136,7 @@ exit 0 .withPrivileged(true) .withHostMounts(mounts) .withPodName(buildPodName(nodeName)) + .withPort(8080) .build() } 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 b3bd847..e554b7a 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeEstimator.groovy +++ b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeEstimator.groovy @@ -34,7 +34,7 @@ class K8sRuntimeEstimator { K8sRuntimeEstimator(String dataFilePath) { HashMap>> data = new HashMap<>(); try { - BufferedReader reader = new BufferedReader(dataFilePath) + BufferedReader reader = new BufferedReader(new FileReader(dataFilePath)) String line = reader.readLine() while (line != null) { // ,, 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 bb0ded6..55e2c88 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy +++ b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskScheduler.groovy @@ -4,20 +4,21 @@ 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 @Slf4j @CompileStatic class K8sTaskScheduler implements Runnable { - private final K8sExecutor executor + private final Duration interval private final K8sSchedulingStrategy strategy private final LinkedBlockingQueue queue = new LinkedBlockingQueue<>() private String[] nodes private synchronized boolean shouldStop - K8sTaskScheduler(String[] nodes, K8sSchedulingStrategy strategy) { - this.executor = executor + K8sTaskScheduler(String[] nodes, K8sSchedulingStrategy strategy, Duration interval) { + this.interval = interval this.strategy = strategy this.nodes = nodes } @@ -73,7 +74,7 @@ class K8sTaskScheduler implements Runnable { * */ void run() { this.shouldStop = false - def interval = this.executor.getK8sConfig().schedulerInterval + final interval = this.interval while (!shouldStop) { sleep(interval.toMillis()) drain() 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 e38fb83..191f4d2 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 @@ -332,6 +332,12 @@ class K8sClient { return new K8sResponseJson(resp.text) } + String getPodIpAddress(String podName) { + assert podName + final K8sResponseJson resp = podStatus0(podName) + (resp?.status as Map)?.podIP as String + } + /* * https://v1-8.docs.kubernetes.io/docs/api-reference/v1.8/#read-status-69 */ diff --git a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/model/PodSpecBuilder.groovy b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/model/PodSpecBuilder.groovy index d8a6607..a6e8bde 100644 --- a/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/model/PodSpecBuilder.groovy +++ b/nextflow/plugins/nf-k8s/src/main/nextflow/k8s/model/PodSpecBuilder.groovy @@ -128,6 +128,8 @@ class PodSpecBuilder { String nodeName + Integer port = null + /** * @return A sequential volume unique identifier */ @@ -406,6 +408,11 @@ class PodSpecBuilder { return this } + PodSpecBuilder withPort(int value) { + this.port = new Integer(value) + return this + } + @PackageScope List createPullSecret() { def result = new ArrayList(1) def entry = new LinkedHashMap(1) @@ -443,6 +450,12 @@ class PodSpecBuilder { if( imagePullPolicy ) container.imagePullPolicy = imagePullPolicy + if( this.port ) { + List ports = [] + ports << [containerPort: this.port.intValue()] + container.ports = ports + } + final secContext = new LinkedHashMap(10) if( privileged ) { // note: privileged flag needs to be defined in the *container* securityContext 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 4a213fc..2bc46c6 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 @@ -20,6 +20,15 @@ import nextflow.util.ArrayTuple @CompileStatic class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { + private static long getTaskMemoryRequirment(TaskRun task) { + return task.config.getMemory() ? task.config.getMemory().bytes : 64 * 1024 * 1024 + } + + private static int getTaskCPURequirement(TaskRun task) { + return task.config.hasCpus() ? task.config.getCpus() : 1 + } + + @Slf4j private class WorkerNode { private class AssignedTask { TaskRun task @@ -78,19 +87,23 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { } void assignTask(TaskRun task, long frequency, K8sDVFSClient dvfsClient) { - final long reqBytes = task.config.getMemory().bytes - final int reqCPUs = task.config.hasCpus() ? task.config.getCpus() : 1 + final long reqBytes = getTaskMemoryRequirment(task) + final int reqCPUs = getTaskCPURequirement(task) this.allocatedMemory += reqBytes this.allocatedCPUs += reqCPUs + + log.info "[K8s] node ${name}: task ${task.name} assigned, now have ${availableMemory} bytes and ${availableCPUs} cpus [${reqBytes}, ${reqCPUs}]" this.tasks.add(new AssignedTask(task, frequency)) updateFrequency(dvfsClient) } void taskFinished(TaskRun task, K8sDVFSClient dvfsClient) { - final long reqBytes = task.config.getMemory().bytes - final int reqCPUs = task.config.hasCpus() ? task.config.getCpus() : 1 + final long reqBytes = getTaskMemoryRequirment(task) + final int reqCPUs = getTaskCPURequirement(task) this.allocatedMemory -= reqBytes this.allocatedCPUs -= reqCPUs + + log.info "[K8s] node ${name}: task ${task.name} finished, now have ${availableMemory} bytes and ${availableCPUs} cpus [${reqBytes}, ${reqCPUs}]" this.tasks.removeIf {it.task == task} updateFrequency(dvfsClient) } @@ -244,6 +257,8 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { continue } globalMaxFrequency = Long.min(globalMaxFrequency, max.asLong) + + log.info "[K8s] node ${node}: ${cpus.asLong} CPUs, ${mem.asLong} bytes RAM ${min.asLong} Hz - ${max.asLong} Hz current ${cur.asLong}" this.nodes.add(new WorkerNode(node, max.asLong, min.asLong, cur.asLong, cpus.asLong, mem.asLong)) } return !this.nodes.isEmpty() @@ -252,19 +267,21 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { /* Returns a list of nodes that fulfill the tasks resource requirements */ private ArrayList filterNodes(TaskRun task) { - final long reqBytes = task.config.getMemory().bytes - final int reqCPUs = task.config.hasCpus() ? task.config.getCpus() : 1 + final long reqBytes = getTaskMemoryRequirment(task) + 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) + } } return suitableNodes } private boolean anyNode(TaskRun task) { - final long reqBytes = task.config.getMemory().bytes - final int reqCPUs = task.config.hasCpus() ? task.config.getCpus() : 1 + final long reqBytes = getTaskMemoryRequirment(task) + final int reqCPUs = getTaskCPURequirement(task) for (WorkerNode n : nodes) { if (n.memoryAmount >= reqBytes && n.availableCPUs >= reqCPUs) return true diff --git a/test/nextflow.config b/test/nextflow.config index f876739..0d8efde 100644 --- a/test/nextflow.config +++ b/test/nextflow.config @@ -22,16 +22,18 @@ k8s { projectDir = '/workspace/projects' cleanup = false - nextflowImage = 'gitea.kleine.eulenhexe.de/kevin/ma/nextflow-dvfs:0.5.5' + nextflowImage = 'gitea.kleine.eulenhexe.de/kevin/ma/nextflow-dvfs:0.6.15' imagePullPolicy = 'IfNotPresent' schedulerInterval = '10s' recordTaskRuntimes = true + schedulingStrategy = 'DVFS' nodeInit { - enabled = false - image = 'gitea.kleine.eulenhexe.de/kevin/ma/dvfs-agent:0.1' - command = ['/bin/agent'] + enabled = true + image = 'gitea.kleine.eulenhexe.de/kevin/ma/dvfs-agent:0.2.0' + command = ['/usr/local/bin/agent'] wait = 'Running' cleanup = false } + }