kevin/ma
1
0
Fork 0
Code Issues 1 Packages Activity

feat: kubernetes node resource queries

Browse Source
This commit is contained in:
Kevin
2026-08-10 11:43:05 +02:00
parent 7b9188ab54
commit c253d30b5d
6 changed files with 439 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
nextflow/Dockerfile.dev
View File

@@ -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"]

4
nextflow/modules/nextflow/src/main/resources/META-INF/build-info.properties
View File

@@ -1,4 +1,4 @@
build=0
version=26.04.0
timestamp=1778418181930
commitId=ad0d91a
timestamp=1784045783078
commitId=7b9188a

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

@@ -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"

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

@@ -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<WorkerNode> 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

295
nextflow/plugins/nf-k8s/src/test/nextflow/k8s/client/K8sClientTest.groovy
View File

@@ -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
}
}

2
test/nextflow.config
View File

@@ -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