towards experiments

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<String, String> 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) {
+        try {
             HttpResponse<String> 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 {
             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"
     }
 }

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()
     }

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()
     }
 

nextflow/plugins/nf-k8s/src/main/nextflow/k8s/K8sRuntimeEstimator.groovy

@@ -34,7 +34,7 @@ class K8sRuntimeEstimator {
     K8sRuntimeEstimator(String dataFilePath) {
         HashMap<String, ArrayList<Tuple2<Long, Long>>> data = new HashMap<>();
         try {
-            BufferedReader reader = new BufferedReader(dataFilePath)
+            BufferedReader reader = new BufferedReader(new FileReader(dataFilePath))
             String line = reader.readLine()
             while (line != null) {
                 // <task-name>,<input-size>,<runtime-ms>

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<K8sSchedulingRequest> 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()

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
      */

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

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

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