From c253d30b5d2729605a193ad1893a60bb2323b48c Mon Sep 17 00:00:00 2001 From: Kevin Trogant Date: Mon, 10 Aug 2026 11:43:05 +0200 Subject: [PATCH] feat: kubernetes node resource queries --- nextflow/Dockerfile.dev | 68 ++-- .../resources/META-INF/build-info.properties | 4 +- .../main/nextflow/k8s/client/K8sClient.groovy | 113 +++++++ .../K8sDVFSSchedulingStrategy.groovy | 9 +- .../nextflow/k8s/client/K8sClientTest.groovy | 295 ++++++++++++++++++ test/nextflow.config | 2 +- 6 files changed, 439 insertions(+), 52 deletions(-) diff --git a/nextflow/Dockerfile.dev b/nextflow/Dockerfile.dev index d7b1bf1..33e7510 100644 --- a/nextflow/Dockerfile.dev +++ b/nextflow/Dockerfile.dev @@ -4,60 +4,34 @@ WORKDIR /opt/nextflow COPY . /opt/nextflow -RUN chmod +x ./gradlew ./launch.sh ./nextflow || true - RUN ./gradlew clean assemble exportClasspath +RUN chmod +x ./gradlew - RUN mkdir -p /opt/nextflow-runtime/lib \ - && i=0; \ - tr ':' '\n' < /opt/nextflow/.launch.classpath | while read -r path; do \ - if [ -f "$path" ]; then \ - name="$(basename "$path")"; \ - target="/opt/nextflow-runtime/lib/$name"; \ - runtime_path="/opt/nextflow/lib/$name"; \ - if [ -e "$target" ]; then \ - i=$((i+1)); \ - target="/opt/nextflow-runtime/lib/${i}-$name"; \ - runtime_path="/opt/nextflow/lib/${i}-$name"; \ - fi; \ - cp "$path" "$target"; \ - echo "$runtime_path" >> /opt/nextflow-runtime/.launch.classpath.lines; \ - fi; \ - done \ - && paste -sd ':' /opt/nextflow-runtime/.launch.classpath.lines > /opt/nextflow-runtime/.launch.classpath +RUN ./gradlew clean assemble pack - RUN mkdir -p /opt/nextflow-runtime/plugins \ - && cp /opt/nextflow/launch.sh /opt/nextflow-runtime/launch.sh \ - && cp /opt/nextflow/nextflow /opt/nextflow-runtime/nextflow \ - && cp -a /opt/nextflow/plugins/nf-k8s /opt/nextflow-runtime/plugins/nf-k8s +RUN mkdir -p /tmp/nf-unpack /opt/nf-dist \ + && cd /tmp/nf-unpack \ + && jar xf /opt/nextflow/modules/nextflow/build/distributions/nextflow-26.04.0.zip \ + && cp -a nextflow-*/* /opt/nf-dist/ \ + && chmod +x /opt/nf-dist/bin/nextflow +RUN mkdir -p /tmp/nxf-home/plugins/nf-k8s-1.5.2 \ + && cd /tmp/nxf-home/plugins/nf-k8s-1.5.2 \ + && jar xf /opt/nextflow/plugins/nf-k8s/build/distributions/nf-k8s-1.5.2.zip + +RUN mkdir -p /tmp/nxf-home/plugins/nf-amazon-3.9.0 \ + && cd /tmp/nxf-home/plugins/nf-amazon-3.9.0 \ + && jar xf /opt/nextflow/plugins/nf-amazon/build/distributions/nf-amazon-3.9.0.zip FROM eclipse-temurin:17-jre WORKDIR /opt/nextflow -COPY --from=build /opt/nextflow-runtime /opt/nextflow +COPY --from=build /opt/nf-dist /opt/nextflow +COPY --from=build /tmp/nxf-home /opt/nextflow/.nextflow -RUN chmod +x /opt/nextflow/launch.sh \ - && cp /opt/nextflow/launch.sh /opt/nextflow/nextflow \ - && chmod +x /opt/nextflow/nextflow \ - && echo "nextflow-dvfs-dev-image 0.1" > /opt/nextflow/DEV_IMAGE_MARKER +ENV PATH="/opt/nextflow/bin:${PATH}" +ENV NXF_HOME="/opt/nextflow/.nextflow" +ENV NXF_PLUGINS_DIR="/opt/nextflow/.nextflow/plugins" +ENV JAVA_OPTS="--add-opens java.base/java.nio.file.spi=ALL-UNNAMED --add-opens java.base/java.util=ALL-UNNAMED --add-opens java.base/java.lang.reflect=ALL-UNNAMED --add-opens java.base/java.lang=ALL-UNNAMED --add-opens java.base/java.util.concurrent=ALL-UNNAMED" - -RUN mkdir -p /opt/nextflow/conf \ - && cat > /opt/nextflow/conf/scm <<'EOF' -providers { - mygitea { - platform = 'gitea' - server = 'https://gitea.kleine.eulenhexe.de' - endpoint = 'https://gitea.kleine.eulenhexe.de/api/v1' - } -} -EOF - -ENV PATH="/opt/nextflow:${PATH}" -ENV NXF_HOME="/tmp/.nextflow" -ENV NXF_PLUGINS_MODE="dev" -ENV NXF_PLUGINS_DIR="/opt/nextflow/plugins" -ENV NXF_SCM_FILE="/opt/nextflow/conf/scm" - -ENTRYPOINT ["/opt/nextflow/launch.sh"] +ENTRYPOINT ["/opt/nextflow/bin/nextflow"] diff --git a/nextflow/modules/nextflow/src/main/resources/META-INF/build-info.properties b/nextflow/modules/nextflow/src/main/resources/META-INF/build-info.properties index 00d4739..06524c8 100644 --- a/nextflow/modules/nextflow/src/main/resources/META-INF/build-info.properties +++ b/nextflow/modules/nextflow/src/main/resources/META-INF/build-info.properties @@ -1,4 +1,4 @@ build=0 version=26.04.0 -timestamp=1778418181930 -commitId=ad0d91a +timestamp=1784045783078 +commitId=7b9188a 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 191f4d2..556a5b1 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 @@ -540,6 +540,119 @@ class K8sClient { new K8sResponseJson(resp.text) } + /** + * Get a specific node by name + * @param name The node name + * @return Response object containing the node details. + */ + K8sResponseJson nodeDescribe(String name) { + assert name + final action = "/api/v1/nodes/$name" + final resp = get(action) + trace('GET', action, resp.text) + new K8sResponseJson(resp.text) + } + + /** + * Query the CPU capacity of a node + * @param nodeName The name of the node + * @return The CPU capacity in cores (as a String representing the quantity, e.g., "4", "2.5") + */ + String getNodeCpuCapacity(String nodeName) { + assert nodeName + final resp = nodeDescribe(nodeName) + final status = resp.status as Map + final capacity = status?.capacity as Map + capacity?.cpu as String + } + + /** + * Query the currently used CPU of a node + * Note: This requires the metrics server to be installed in the cluster. + * Falls back to the allocated CPU from the node's allocated resources if metrics are not available. + * @param nodeName The name of the node + * @return The used CPU in cores (as a String representing the quantity, e.g., "1", "0.5") + */ + String getNodeCpuUsed(String nodeName) { + assert nodeName + + // First, try to get metrics from the metrics server + try { + final action = "/apis/metrics.k8s.io/v1beta1/nodes/$nodeName" + final resp = get(action) + trace('GET', action, resp.text) + final metrics = new K8sResponseJson(resp.text) + final usage = metrics.usage as Map + return usage?.cpu as String + } + catch (Exception e) { + // Fall back to allocated CPU from node status + log.debug("Metrics server not available or error fetching metrics for node $nodeName, falling back to allocated resources: ${e.message}") + final nodeResp = nodeDescribe(nodeName) + final status = nodeResp.status as Map + final allocatable = status?.allocatable as Map + final allocated = status?.allocated as Map + + // Calculate used as allocatable minus available (if we had that info) + // For now, return the allocated CPU if available + if (allocated?.cpu) { + return allocated.cpu as String + } + + // If we can't get used metrics, return null + return null + } + } + + /** + * Query the memory capacity of a node + * @param nodeName The name of the node + * @return The memory capacity in bytes (as a String representing the quantity, e.g., "16Gi", "8192Mi") + */ + String getNodeMemoryCapacity(String nodeName) { + assert nodeName + final resp = nodeDescribe(nodeName) + final status = resp.status as Map + final capacity = status?.capacity as Map + capacity?.memory as String + } + + /** + * Query the currently used memory of a node + * Note: This requires the metrics server to be installed in the cluster. + * Falls back to the allocated memory from the node's allocated resources if metrics are not available. + * @param nodeName The name of the node + * @return The used memory in bytes (as a String representing the quantity, e.g., "4Gi", "2048Mi") + */ + String getNodeMemoryUsed(String nodeName) { + assert nodeName + + // First, try to get metrics from the metrics server + try { + final action = "/apis/metrics.k8s.io/v1beta1/nodes/$nodeName" + final resp = get(action) + trace('GET', action, resp.text) + final metrics = new K8sResponseJson(resp.text) + final usage = metrics.usage as Map + return usage?.memory as String + } + catch (Exception e) { + // Fall back to allocated memory from node status + log.debug("Metrics server not available or error fetching metrics for node $nodeName, falling back to allocated resources: ${e.message}") + final nodeResp = nodeDescribe(nodeName) + final status = nodeResp.status as Map + final allocated = status?.allocated as Map + + // Return the allocated memory if available + if (allocated?.memory) { + return allocated.memory as String + } + + // If we can't get used metrics, return null + return null + } + } + 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 554d04b..1ef16f5 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 @@ -63,6 +63,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { this.allocatedCPUs = 0 this.allocatedMemory = 0 + this.allocatedMemory += 100 * 1024 * 1024 this.tasks = new ArrayList<>() } @@ -146,6 +147,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { private long finishedTaskCount private long globalMaxFrequency + private long globalMinFrequency K8sDVFSSchedulingStrategy(K8sRuntimeEstimator runtimeEstimator, K8sDVFSClient dvfsClient) { this.runtimeEstimator = runtimeEstimator @@ -184,13 +186,15 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { if (!isCriticalPath) { /* 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) + frequency = Math.min(frequency, globalMaxFrequency) } /* Step 2.2: Filter nodes based on task requirements */ ArrayList suitableNodes = filterNodes(req.task) if (suitableNodes.size() == 0) { if (!anyNode(req.task)) { - //log.error "[K8s] unable to schedule task ${req.task} - no node satisfies resource requirements" + log.error "[K8s] unable to schedule task ${req.task} - no node satisfies resource requirements ${getTaskMemoryRequirment(req.task)} bytes ${getTaskCPURequirement(req.task)} cpus" return null } /* No node can currently execute this task, but it should be possible in the future */ @@ -258,6 +262,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { continue } globalMaxFrequency = Long.min(globalMaxFrequency, max.asLong) + globalMinFrequency = Long.max(globalMinFrequency, min.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)) @@ -284,7 +289,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy { final long reqBytes = getTaskMemoryRequirment(task) final int reqCPUs = getTaskCPURequirement(task) for (WorkerNode n : nodes) { - if (n.memoryAmount >= reqBytes && n.availableCPUs >= reqCPUs) + if (n.memoryAmount >= reqBytes && n.cpuCount >= reqCPUs) return true } return false diff --git a/nextflow/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy b/nextflow/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy index 263a6e3..dc56862 100644 --- a/nextflow/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy +++ b/nextflow/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy @@ -1102,4 +1102,299 @@ class K8sClientTest extends Specification { result.terminated.exitCode == null result.terminated.exitcode == null } + + def 'should describe a node' () { + given: + def JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node", + "namespace": "default" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + }, + "allocatable": { + "cpu": "3.5", + "memory": "14Gi" + } + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + def RESP = Mock(K8sResponseApi) + RESP.getText() >> JSON + + when: + def result = client.nodeDescribe(NODE_NAME) + then: + 1 * client.get('/api/v1/nodes/test-node') >> RESP + result.kind == 'Node' + result.metadata.name == 'test-node' + } + + def 'should get node CPU capacity' () { + given: + def JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + } + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + def RESP = Mock(K8sResponseApi) + RESP.getText() >> JSON + + when: + def result = client.getNodeCpuCapacity(NODE_NAME) + then: + 1 * client.nodeDescribe(NODE_NAME) >> new K8sResponseJson(JSON) + result == '4' + } + + def 'should get node memory capacity' () { + given: + def JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + } + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + + when: + def result = client.getNodeMemoryCapacity(NODE_NAME) + then: + 1 * client.nodeDescribe(NODE_NAME) >> new K8sResponseJson(JSON) + result == '16Gi' + } + + def 'should get node CPU used from metrics server' () { + given: + def NODE_JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + } + } + } + ''' + def METRICS_JSON = ''' + { + "kind": "NodeMetrics", + "apiVersion": "metrics.k8s.io/v1beta1", + "metadata": { + "name": "test-node" + }, + "usage": { + "cpu": "1500m", + "memory": "2Gi" + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + def NODE_RESP = Mock(K8sResponseApi) + NODE_RESP.getText() >> NODE_JSON + def METRICS_RESP = Mock(K8sResponseApi) + METRICS_RESP.getText() >> METRICS_JSON + + when: + def result = client.getNodeCpuUsed(NODE_NAME) + then: + 1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> METRICS_RESP + result == '1500m' + } + + def 'should get node memory used from metrics server' () { + given: + def NODE_JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + } + } + } + ''' + def METRICS_JSON = ''' + { + "kind": "NodeMetrics", + "apiVersion": "metrics.k8s.io/v1beta1", + "metadata": { + "name": "test-node" + }, + "usage": { + "cpu": "1500m", + "memory": "2Gi" + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + def METRICS_RESP = Mock(K8sResponseApi) + METRICS_RESP.getText() >> METRICS_JSON + + when: + def result = client.getNodeMemoryUsed(NODE_NAME) + then: + 1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> METRICS_RESP + result == '2Gi' + } + + def 'should fallback when metrics server not available for CPU' () { + given: + def NODE_JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + }, + "allocated": { + "cpu": "2", + "memory": "8Gi" + } + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + def NODE_RESP = Mock(K8sResponseApi) + NODE_RESP.getText() >> NODE_JSON + + when: + def result = client.getNodeCpuUsed(NODE_NAME) + then: + // First attempt to metrics server fails + 1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> { throw new K8sResponseException("Metrics not available", new ByteArrayInputStream('{}'.bytes)) } + // Fallback to node describe - which calls get internally + 1 * client.get('/api/v1/nodes/test-node') >> NODE_RESP + result == '2' + } + + def 'should fallback when metrics server not available for memory' () { + given: + def NODE_JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + }, + "allocated": { + "cpu": "2", + "memory": "8Gi" + } + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + def NODE_RESP = Mock(K8sResponseApi) + NODE_RESP.getText() >> NODE_JSON + + when: + def result = client.getNodeMemoryUsed(NODE_NAME) + then: + // First attempt to metrics server fails + 1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> { throw new K8sResponseException("Metrics not available", new ByteArrayInputStream('{}'.bytes)) } + // Fallback to node describe - which calls get internally + 1 * client.get('/api/v1/nodes/test-node') >> NODE_RESP + result == '8Gi' + } + + def 'should return null when no allocated resources available' () { + given: + def NODE_JSON = ''' + { + "kind": "Node", + "apiVersion": "v1", + "metadata": { + "name": "test-node" + }, + "status": { + "capacity": { + "cpu": "4", + "memory": "16Gi" + } + } + } + ''' + + def client = Spy(K8sClient) + final NODE_NAME = 'test-node' + def NODE_RESP = Mock(K8sResponseApi) + NODE_RESP.getText() >> NODE_JSON + + when: + def resultCpu = client.getNodeCpuUsed(NODE_NAME) + def resultMem = client.getNodeMemoryUsed(NODE_NAME) + then: + // First attempt to metrics server fails for CPU + 1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> { throw new K8sResponseException("Metrics not available", new ByteArrayInputStream('{}'.bytes)) } + // Fallback to node describe for CPU + 1 * client.get('/api/v1/nodes/test-node') >> NODE_RESP + // First attempt to metrics server fails for Memory + 1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> { throw new K8sResponseException("Metrics not available", new ByteArrayInputStream('{}'.bytes)) } + // Fallback to node describe for Memory + 1 * client.get('/api/v1/nodes/test-node') >> NODE_RESP + resultCpu == null + resultMem == null + } } diff --git a/test/nextflow.config b/test/nextflow.config index 43fdbba..f5383ec 100644 --- a/test/nextflow.config +++ b/test/nextflow.config @@ -22,7 +22,7 @@ k8s { projectDir = '/workspace/projects' cleanup = false - nextflowImage = 'gitea.kleine.eulenhexe.de/kevin/ma/nextflow-dvfs:0.6.18' + nextflowImage = 'gitea.kleine.eulenhexe.de/kevin/ma/nextflow-dvfs:0.6.29' imagePullPolicy = 'IfNotPresent' schedulerInterval = '10s' recordTaskRuntimes = true