A submitted Kubernetes job has succeeded with this code.

Error handling still incomplete.

Author: jaredjennings

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

@@ -15,7 +15,7 @@ import play.api.{Configuration, Logger}
 import akka.actor.ActorSystem
 import io.fabric8.kubernetes.client.{DefaultKubernetesClient, ConfigBuilder, Watcher}
 import io.fabric8.kubernetes.api.model.batch.{Job => KJob, JobBuilder => KJobBuilder}
-
+import io.fabric8.kubernetes.api.model.{PersistentVolumeClaimVolumeSourceBuilder}
 // import com.spotify.docker.client.DockerClient.LogsParam
 // import com.spotify.docker.client.messages.HostConfig.Bind
 // import com.spotify.docker.client.messages.{ContainerConfig, HostConfig}
@@ -58,42 +58,88 @@ class K8sJobRunnerSrv(
     }.get
 
   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()
+    // Spicy meatball: under Kubernetes, executions can fail for reasons
+    // other than bad inputs, for example, if a node dies. So we maybe
+    // can't say, "Kubernetes, only do this once! I will do any
+    // necessary retrying." But there may be quotas on analyzer usage,
+    // and retrying outside Cortex may use them up.
+
+    val cacertsFile = jobDirectory.resolve("input").resolve("cacerts")
+
+    // FIXME: this collapses case, jeopardizing the uniqueness of the
+    // identifier.
+    val kname = "_".r.replaceAllIn(job.id.map(_.toLower), "-")
+    val pvcvs = new PersistentVolumeClaimVolumeSourceBuilder()
+      .withClaimName(persistentVolumeClaimName)
+      .withReadOnly(false)
+      .build();
+    val kjob1 = new KJobBuilder()
+      .withApiVersion("batch/v1")
+      .withNewMetadata()
+      .withName(kname)
+      .withLabels(Map(
+        "cortex-job-id" -> job.id,
+        "cortex-worker-id" -> job.workerId(),
+        "cortex-job" -> "true").asJava)
+      .endMetadata()
+      .withNewSpec()
+      .withNewTemplate()
+      .withNewSpec()
+      .addNewVolume()
+      .withName("job-directory")
+      .withPersistentVolumeClaim(pvcvs)
+      .endVolume()
+      .addNewContainer()
+      .withName("neuron")
+      .withImage(dockerImage)
+      .withArgs("/job")
+      .addNewEnv()
+      .withName("CORTEX_JOB_FOLDER")
+      .withValue(jobBaseDirectory.relativize(jobDirectory).toString())
+      .endEnv();
+    val kjob2 = if (Files.exists(cacertsFile)) {
+      kjob1.addNewEnv()
+        .withName("REQUESTS_CA_BUNDLE")
+        .withValue("/job/input/cacerts")
+        .endEnv()
+    } else {
+      kjob1
+    }
+    val kjob3 = kjob2
+      .addNewVolumeMount()
+      .withName("job-directory")
+      .withSubPathExpr("$(CORTEX_JOB_FOLDER)/input")
+      .withMountPath("/job/input")
+      .withReadOnly(true)
+      .endVolumeMount()
+      .addNewVolumeMount()
+      .withName("job-directory")
+      .withSubPathExpr("$(CORTEX_JOB_FOLDER)/output")
+      .withMountPath("/job/output")
+      .withReadOnly(false)
+      .endVolumeMount()
+      .endContainer()
+      .withRestartPolicy("Never")
+      .endSpec()
     .endTemplate()
     .endSpec()
     .build();
-  logger.info(s"Constructed k8s Job ${job.workerId()}\n")
+    logger.info(s"Constructed k8s Job ${kjob3.getMetadata().getName()}\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
+    val execution = Future {
+      val created_kjob = client.batch().jobs().create(kjob3)
+      logger.info(s"Created k8s Job ${created_kjob.getMetadata().getName()}")
+      // FIXME: use the given timeout value
+      val ended_kjob = client.batch().jobs().withName(kname)
+        .waitUntilCondition(j => (j.getStatus().getFailed() > 0 || j.getStatus().getSucceeded() > 0),
+          5, TimeUnit.MINUTES );
+      ()
+    }.andThen {
+      case r =>
+        val foo_kjob = client.batch().jobs().withName(kname).get()
+        logger.info(s"k8s Job ${foo_kjob.getMetadata().getUid()} status ${foo_kjob.getStatus().toString()}")
+        Future {}
+    }
+    execution
   }
 }