create a Kubernetes job. the job does not run properly yet

Author: jaredjennings

app/org/thp/cortex/services/K8sJobRunnerSrv.scala

@@ -1,17 +1,21 @@
 package org.thp.cortex.services
 
+import java.util.concurrent.TimeUnit
 import java.nio.charset.StandardCharsets
 import java.nio.file._
 
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.Try
+import scala.collection.JavaConverters._
 
 import play.api.libs.json.Json
 import play.api.{Configuration, Logger}
 
 import akka.actor.ActorSystem
-import io.fabric8.kubernetes.client.{DefaultKubernetesClient, ConfigBuilder}
+import io.fabric8.kubernetes.client.{DefaultKubernetesClient, ConfigBuilder, Watcher}
+import io.fabric8.kubernetes.api.model.batch.{Job => KJob, JobBuilder => KJobBuilder}
+
 // import com.spotify.docker.client.DockerClient.LogsParam
 // import com.spotify.docker.client.messages.HostConfig.Bind
 // import com.spotify.docker.client.messages.{ContainerConfig, HostConfig}
@@ -44,14 +48,52 @@ class K8sJobRunnerSrv(
 
   lazy val isAvailable: Boolean =
     Try {
-      logger.info(s"Kubernetes is available:\n${client.getVersion().toString()}")
+      val ver = client.getVersion()
+      logger.info(s"Kubernetes is available: major ${ver.getMajor()} minor ${ver.getMinor()} git ${ver.getGitCommit()}")
       true
     }.recover {
       case error =>
         logger.info(s"Kubernetes is not available", error)
         false
     }.get
 
-  def run(jobDirectory: Path, dockerImage: String, job: Job, timeout: Option[FiniteDuration])(implicit ec: ExecutionContext): Future[Unit] = { Future {} }
+  def run(jobDirectory: Path, dockerImage: String, job: Job, timeout: Option[FiniteDuration])(implicit ec: ExecutionContext): Future[Unit] = {
+  val kjob = new KJobBuilder()
+    .withApiVersion("batch/v1")
+    .withNewMetadata()
+    .withName(job.workerId())
+    .withLabels(Map("cortex_workerId" -> job.workerId()).asJava)
+    .endMetadata()
+    .withNewSpec()
+    .withNewTemplate()
+    .withNewSpec()
+    .addNewContainer()
+    .withName("neuron")
+    .withImage(dockerImage)
+    .withArgs("/job")
+    .endContainer()
+    .withRestartPolicy("Never")
+    .endSpec()
+    .endTemplate()
+    .endSpec()
+    .build();
+  logger.info(s"Constructed k8s Job ${job.workerId()}\n")
 
+  val execution = Future {
+    val created_kjob = client.batch().jobs().create(kjob)
+    logger.info(s"Created k8s Job ${created_kjob.getMetadata().getUid()}")
+    logger.info(s"the status at creation time is ${created_kjob.getStatus().toString()}")
+    // FIXME: use the given timeout value
+    val ended_kjob = client.batch().jobs().withName(job.workerId())
+      .waitUntilCondition(j => (j.getStatus().getFailed() > 0 || j.getStatus().getSucceeded() > 0),
+      5, TimeUnit.MINUTES );
+    ()
+  }.andThen {
+    case r =>
+      val foo_kjob = client.batch().jobs().withName(job.workerId()).get()
+      logger.info(s"k8s Job ${foo_kjob.getMetadata().getUid()} status ${foo_kjob.getStatus().toString()}")
+      Future {}
+  }
+  execution
+  }
 }