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

fix: Support "nano" cpus

Browse Source
This commit is contained in:
Kevin
2026-08-10 19:41:30 +02:00
parent c253d30b5d
commit fb85e20f44
4 changed files with 490 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sExecutor.groovy
View File

@@ -110,7 +110,9 @@ class K8sExecutor extends Executor implements ExtensionPoint {
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))
strategy = new K8sDVFSSchedulingStrategy(this.runtimeEstimator,
new K8sDVFSClient(nodes, ips),
() -> getClient())
} else {
log.error "[K8s] invalid scheduling strategy $k8sConfig.schedulingStrategy, falling back on \"Hash\""
strategy = new K8sHashSchedulingStrategy()

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

@@ -653,6 +653,206 @@ class K8sClient {
}
}
/**
* Parse a Kubernetes quantity string (e.g., "16Gi", "8192Mi", "1024Ki", "1500m") into a long value.
* For memory: supports binary (Ki, Mi, Gi, Ti, Pi, Ei) and the value is returned in bytes.
* For CPU: supports millicores (e.g., "1500m" = 1.5 cores = 1500 millicores) and cores (e.g., "2" = 2000 millicores).
*
* @param quantity The Kubernetes quantity string (e.g., "16Gi", "1500m", "2")
* @param forCpu If true, parses as CPU (millicores); if false, parses as memory (bytes)
* @return The value as a Long, or null if unparseable
*/
static Long parseK8sQuantity(String quantity, boolean forCpu=false) {
if (!quantity) return null
// Reject quantities with spaces - Kubernetes quantity strings don't have spaces
if (quantity.contains(' ')) {
return null
}
// Extract numeric value and suffix separately
// Kubernetes quantities: https://kubernetes.io/docs/reference/kubernetes-api/common-definitions/#quantity
def suffix = null
def numericStr = quantity
// Try to extract suffix - check for the longest suffixes first
// Binary: KiB, MiB, GiB, TiB, PiB, EiB, Ki, Mi, Gi, Ti, Pi, Ei
// Decimal: KB, MB, GB, TB, PB, EB, K, M, G, T, P, E
// CPU: m (millicores), n (nanocores)
def suffixes = ['kib', 'mib', 'gib', 'tib', 'pib', 'eib',
'ki', 'mi', 'gi', 'ti', 'pi', 'ei',
'kb', 'mb', 'gb', 'tb', 'pb', 'eb',
'k', 'm', 'g', 't', 'p', 'e']
// For CPU, we only care about 'm' (millicores) and 'n' (nanocores) suffixes
if (forCpu) {
suffixes = ['m', 'n']
}
for (def s : suffixes) {
if (quantity.toLowerCase().endsWith(s)) {
suffix = s
numericStr = quantity.substring(0, quantity.length() - s.length())
break
}
}
// If we couldn't extract a suffix, try to parse the whole thing as a number
// Validate that numericStr only contains valid numeric characters
if (numericStr && !numericStr.matches(/^[+-]?[0-9]*(\.[0-9]*)?$/)) {
return null
}
def numericValue
try {
numericValue = numericStr.toDouble()
} catch (Exception e) {
return null
}
if (forCpu) {
// CPU: handle millicores (m) 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 {
// Default is cores, convert to millicores
return (long)(numericValue * 1000)
}
} else {
// Memory: convert to bytes
def multiplier = 1L
if (suffix == null || suffix == '') {
// No suffix - assume bytes
return (long)numericValue
}
// Binary suffixes (Ki, Mi, Gi, Ti, Pi, Ei)
switch (suffix) {
case 'ki': case 'kib':
multiplier = 1024L
break
case 'mi': case 'mib':
multiplier = 1024L * 1024L
break
case 'gi': case 'gib':
multiplier = 1024L * 1024L * 1024L
break
case 'ti': case 'tib':
multiplier = 1024L * 1024L * 1024L * 1024L
break
case 'pi': case 'pib':
multiplier = 1024L * 1024L * 1024L * 1024L * 1024L
break
case 'ei': case 'eib':
multiplier = 1024L * 1024L * 1024L * 1024L * 1024L * 1024L
break
// Decimal suffixes (K, M, G, T, P, E) - Kubernetes uses binary by default
// but we support decimal for completeness
case 'k': case 'kb':
multiplier = 1000L
break
case 'm': case 'mb':
multiplier = 1000L * 1000L
break
case 'g': case 'gb':
multiplier = 1000L * 1000L * 1000L
break
case 't': case 'tb':
multiplier = 1000L * 1000L * 1000L * 1000L
break
case 'p': case 'pb':
multiplier = 1000L * 1000L * 1000L * 1000L * 1000L
break
case 'e': case 'eb':
multiplier = 1000L * 1000L * 1000L * 1000L * 1000L * 1000L
break
}
return (long)(numericValue * multiplier)
}
}
/**
* Query the CPU capacity of a node in millicores (1 core = 1000 millicores)
* @param nodeName The name of the node
* @return The CPU capacity in millicores as a Long, or null if unparseable
*/
Long getNodeCpuCapacityMillis(String nodeName) {
assert nodeName
final capacity = getNodeCpuCapacity(nodeName)
if (!capacity) {
log.debug("Node $nodeName CPU capacity not available")
return null
}
final result = parseK8sQuantity(capacity, true)
if (result == null) {
log.warn("Unable to parse CPU capacity for node $nodeName: '$capacity'")
}
return result
}
/**
* Query the currently used CPU of a node in millicores (1 core = 1000 millicores)
* @param nodeName The name of the node
* @return The used CPU in millicores as a Long, or null if unparseable
*/
Long getNodeCpuUsedMillis(String nodeName) {
assert nodeName
final used = getNodeCpuUsed(nodeName)
if (!used) {
log.debug("Node $nodeName CPU usage not available (metrics server not installed or node has no allocated resources)")
return null
}
final result = parseK8sQuantity(used, true)
if (result == null) {
log.warn("Unable to parse CPU usage for node $nodeName: '$used'")
}
return result
}
/**
* Query the memory capacity of a node in bytes
* @param nodeName The name of the node
* @return The memory capacity in bytes as a Long, or null if unparseable
*/
Long getNodeMemoryCapacityBytes(String nodeName) {
assert nodeName
final capacity = getNodeMemoryCapacity(nodeName)
if (!capacity) {
log.debug("Node $nodeName memory capacity not available")
return null
}
final result = parseK8sQuantity(capacity, false)
if (result == null) {
log.warn("Unable to parse memory capacity for node $nodeName: '$capacity'")
}
return result
}
/**
* Query the currently used memory of a node in bytes
* @param nodeName The name of the node
* @return The used memory in bytes as a Long, or null if unparseable
*/
Long getNodeMemoryUsedBytes(String nodeName) {
assert nodeName
final used = getNodeMemoryUsed(nodeName)
if (!used) {
log.debug("Node $nodeName memory usage not available (metrics server not installed or node has no allocated resources)")
return null
}
final result = parseK8sQuantity(used, false)
if (result == null) {
log.warn("Unable to parse memory usage for node $nodeName: '$used'")
}
return result
}
protected void checkInvalidWaitingState( Map waiting, K8sResponseJson resp ) {
if( waiting.reason == 'ErrImagePull' || waiting.reason == 'ImagePullBackOff') {
def message = "K8s pod image cannot be pulled"

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

@@ -9,6 +9,7 @@ import nextflow.k8s.K8sSchedulingRequest
import nextflow.k8s.K8sSchedulingStrategy
import nextflow.k8s.K8sTaskHandler
import nextflow.k8s.K8sTaskScheduler
import nextflow.k8s.client.K8sClient
import nextflow.processor.TaskRun
import nextflow.util.ArrayTuple
@@ -20,6 +21,14 @@ import nextflow.util.ArrayTuple
@CompileStatic
class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
/** Used for passing K8sExecutor.getClient.
* We cannot pass the client directly, becaue it can be refreshed
* during workflow execution.
*/
public interface K8sClientGetter {
K8sClient getClient()
}
private static long getTaskMemoryRequirment(TaskRun task) {
return task.config.getMemory() ? task.config.getMemory().bytes : 64 * 1024 * 1024
}
@@ -45,34 +54,54 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
long maxFrequency
long minFrequency
long currentFrequency
long cpuCount
long memoryAmount
long allocatedMemory
long allocatedCPUs
ArrayList<AssignedTask> tasks
WorkerNode(String name, long maxF, long minF, long curF, long cpus, long mem) {
WorkerNode(String name, long maxF, long minF, long curF, K8sClientGetter clientGetter) {
this.name = name
this.maxFrequency = maxF
this.minFrequency = minF
this.currentFrequency = curF
this.cpuCount = cpus
this.memoryAmount = mem
this.allocatedCPUs = 0
this.allocatedMemory = 0
this.allocatedMemory += 100 * 1024 * 1024
this.tasks = new ArrayList<>()
}
/* Return the number of available (unoccupied) bytes */
long getAvailableMemory() {
memoryAmount - allocatedMemory
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}"
return 0
}
return capacity.longValue() - used.longValue()
}
/* Return the number of available (unoccupied) cpu cores */
long getAvailableCPUs() {
cpuCount - allocatedCPUs
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}"
return 0
}
long availMilis = capacity.longValue() - used.longValue()
return availMilis.intdiv(1000)
}
/* Return the total amount of installed memory */
long getMemoryAmount() {
def k8sClient = clientGetter.getClient()
Long capacity = k8sClient.getNodeMemoryCapacityBytes(this.name)
return capacity != null ? capacity.longValue() : 0
}
/* Return the total number of installed cpu cores */
long getCPUCount() {
def k8sClient = clientGetter.getClient()
Long capacity = k8sClient.getNodeCpuCapacityMillis(this.name)
return capacity != null ? capacity.longValue().intdiv(1000) : 0
}
// Sets the frequency to the max. requested frequency of all currently running tasks.
@@ -89,23 +118,13 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
}
void assignTask(TaskRun task, long frequency, K8sDVFSClient dvfsClient) {
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}]: at ${frequency}/${maxFrequency}"
log.info "[K8s] node ${name}: task ${task.name} assigned with ${frequency}/${maxFrequency}"
this.tasks.add(new AssignedTask(task, frequency))
updateFrequency(dvfsClient)
}
void taskFinished(TaskRun task, K8sDVFSClient dvfsClient) {
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}]"
log.info "[K8s] node ${name}: task ${task.name} finished"
this.tasks.removeIf {it.task == task}
updateFrequency(dvfsClient)
}
@@ -149,11 +168,14 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
private long globalMaxFrequency
private long globalMinFrequency
K8sDVFSSchedulingStrategy(K8sRuntimeEstimator runtimeEstimator, K8sDVFSClient dvfsClient) {
private K8sClientGetter clientGetter
K8sDVFSSchedulingStrategy(K8sRuntimeEstimator runtimeEstimator, K8sDVFSClient dvfsClient, K8sClientGetter clientGetter) {
this.runtimeEstimator = runtimeEstimator
this.dvfsClient = dvfsClient
this.nodes = new ArrayList<>()
this.taskToNode = new HashMap<>();
this.clientGetter = clientGetter
}
@Override
@@ -265,7 +287,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
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))
this.nodes.add(new WorkerNode(node, max.asLong, min.asLong, cur.asLong, clientGetter))
}
return !this.nodes.isEmpty()
}
@@ -289,7 +311,7 @@ class K8sDVFSSchedulingStrategy implements K8sSchedulingStrategy {
final long reqBytes = getTaskMemoryRequirment(task)
final int reqCPUs = getTaskCPURequirement(task)
for (WorkerNode n : nodes) {
if (n.memoryAmount >= reqBytes && n.cpuCount >= reqCPUs)
if (n.memoryAmount >= reqBytes && n.CPUCount >= reqCPUs)
return true
}
return false

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

@@ -1397,4 +1397,234 @@ class K8sClientTest extends Specification {
resultCpu == null
resultMem == null
}
def 'should parse memory quantity strings' () {
expect:
K8sClient.parseK8sQuantity(null) == null
K8sClient.parseK8sQuantity('') == null
K8sClient.parseK8sQuantity('1024') == 1024L
K8sClient.parseK8sQuantity('1Ki') == 1024L
K8sClient.parseK8sQuantity('1KiB') == 1024L
K8sClient.parseK8sQuantity('1Mi') == 1024L * 1024L
K8sClient.parseK8sQuantity('1MiB') == 1024L * 1024L
K8sClient.parseK8sQuantity('1Gi') == 1024L * 1024L * 1024L
K8sClient.parseK8sQuantity('1GiB') == 1024L * 1024L * 1024L
K8sClient.parseK8sQuantity('16Gi') == 16L * 1024L * 1024L * 1024L
K8sClient.parseK8sQuantity('8192Mi') == 8192L * 1024L * 1024L
K8sClient.parseK8sQuantity('0.5Gi') == (long)(0.5 * 1024L * 1024L * 1024L)
}
def 'should parse CPU quantity strings' () {
expect:
K8sClient.parseK8sQuantity(null, true) == null
K8sClient.parseK8sQuantity('', true) == null
K8sClient.parseK8sQuantity('1', true) == 1000L // 1 core = 1000 millicores
K8sClient.parseK8sQuantity('2', true) == 2000L // 2 cores = 2000 millicores
K8sClient.parseK8sQuantity('0.5', true) == 500L // 0.5 cores = 500 millicores
K8sClient.parseK8sQuantity('1500m', true) == 1500L // 1500 millicores
K8sClient.parseK8sQuantity('100m', true) == 100L // 100 millicores
// Nanocores: 1 core = 1,000,000,000 nanocores, so we divide by 1,000,000 to get millicores
K8sClient.parseK8sQuantity('1000000n', true) == 1L // 1,000,000 nanocores = 1 millicore
K8sClient.parseK8sQuantity('1000000000n', true) == 1000L // 1,000,000,000 nanocores = 1000 millicores = 1 core
K8sClient.parseK8sQuantity('298332671n', true) == 298L // 298,332,671 nanocores ≈ 298.33 millicores
K8sClient.parseK8sQuantity('434126984n', true) == 434L // 434,126,984 nanocores ≈ 434.13 millicores
}
def 'should get node CPU capacity in millicores' () {
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'
when:
def result = client.getNodeCpuCapacityMillis(NODE_NAME)
then:
1 * client.nodeDescribe(NODE_NAME) >> new K8sResponseJson(NODE_JSON)
result == 4000L // 4 cores = 4000 millicores
}
def 'should get node CPU used in millicores' () {
given:
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.getNodeCpuUsedMillis(NODE_NAME)
then:
1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> METRICS_RESP
result == 1500L // 1500 millicores
}
def 'should get node memory capacity in bytes' () {
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'
when:
def result = client.getNodeMemoryCapacityBytes(NODE_NAME)
then:
1 * client.nodeDescribe(NODE_NAME) >> new K8sResponseJson(NODE_JSON)
result == 16L * 1024L * 1024L * 1024L // 16 GiB in bytes
}
def 'should get node memory used in bytes' () {
given:
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.getNodeMemoryUsedBytes(NODE_NAME)
then:
1 * client.get('/apis/metrics.k8s.io/v1beta1/nodes/test-node') >> METRICS_RESP
result == 2L * 1024L * 1024L * 1024L // 2 GiB in bytes
}
def 'should handle fractional CPU values' () {
expect:
K8sClient.parseK8sQuantity('2.5', true) == 2500L // 2.5 cores = 2500 millicores
K8sClient.parseK8sQuantity('0.25', true) == 250L // 0.25 cores = 250 millicores
}
def 'should handle fractional memory values' () {
expect:
K8sClient.parseK8sQuantity('1.5Gi') == (long)(1.5 * 1024L * 1024L * 1024L)
K8sClient.parseK8sQuantity('0.5Mi') == (long)(0.5 * 1024L * 1024L)
}
def 'should handle decimal suffixes for memory' () {
expect:
K8sClient.parseK8sQuantity('1K') == 1000L
K8sClient.parseK8sQuantity('1M') == 1000L * 1000L
K8sClient.parseK8sQuantity('1G') == 1000L * 1000L * 1000L
}
def 'should return null for invalid quantity strings' () {
expect:
K8sClient.parseK8sQuantity('invalid') == null
K8sClient.parseK8sQuantity('16 GiB') == null // space not allowed
K8sClient.parseK8sQuantity('Gi16') == null // suffix before number (doesn't match any suffix pattern)
K8sClient.parseK8sQuantity('16.5.5') == null // multiple decimal points
K8sClient.parseK8sQuantity('abc123') == null // non-numeric start
K8sClient.parseK8sQuantity('123abc') == null // non-numeric end (no valid suffix)
}
def 'should handle large memory values like 8058776Ki' () {
expect:
K8sClient.parseK8sQuantity('8058776Ki') == 8058776L * 1024L
}
def 'should handle large memory values like 8058776ki' () {
expect:
K8sClient.parseK8sQuantity('8058776ki') == 8058776L * 1024L
}
def 'should handle memory values with B suffix' () {
expect:
K8sClient.parseK8sQuantity('1KiB') == 1024L
K8sClient.parseK8sQuantity('1MiB') == 1024L * 1024L
K8sClient.parseK8sQuantity('1GiB') == 1024L * 1024L * 1024L
}
def 'should handle decimal memory suffixes' () {
expect:
K8sClient.parseK8sQuantity('1K') == 1000L
K8sClient.parseK8sQuantity('1KB') == 1000L
K8sClient.parseK8sQuantity('1M') == 1000L * 1000L
K8sClient.parseK8sQuantity('1MB') == 1000L * 1000L
K8sClient.parseK8sQuantity('1G') == 1000L * 1000L * 1000L
}
def 'should handle real world memory value 8058776Ki' () {
expect:
K8sClient.parseK8sQuantity('8058776Ki') == 8058776L * 1024L
K8sClient.parseK8sQuantity('8058776ki') == 8058776L * 1024L
K8sClient.parseK8sQuantity('8058776KI') == 8058776L * 1024L
}
def 'should handle CPU values without suffix' () {
expect:
K8sClient.parseK8sQuantity('1', true) == 1000L
K8sClient.parseK8sQuantity('2', true) == 2000L
K8sClient.parseK8sQuantity('0.5', true) == 500L
K8sClient.parseK8sQuantity('2.5', true) == 2500L
}
def 'should handle fractional memory values with suffixes' () {
expect:
K8sClient.parseK8sQuantity('1.5Gi') == (long)(1.5 * 1024 * 1024 * 1024)
K8sClient.parseK8sQuantity('0.5Mi') == (long)(0.5 * 1024 * 1024)
K8sClient.parseK8sQuantity('2.25Ki') == (long)(2.25 * 1024)
}
def 'should handle edge cases' () {
expect:
K8sClient.parseK8sQuantity('0') == 0L
K8sClient.parseK8sQuantity('0Ki') == 0L
K8sClient.parseK8sQuantity('0m', true) == 0L
K8sClient.parseK8sQuantity('+100') == 100L
K8sClient.parseK8sQuantity('-50') == -50L
}
}