apply mesos#538

Author: vixns

docs/docs/api.md

@@ -180,7 +180,7 @@ Here is a more elaborate example of a dependent job hash:
 ## Adding a Docker Job
 
 A docker job takes the same format as a scheduled job or a dependency job and runs on a Docker container.
-To configure it, an additional `container` argument is required, which contains a type (required), an image (required), a network mode (optional), mounted volumes (optional) and whether Mesos should always pull the latest image before executing or not (optional).
+To configure it, an additional `container` argument is required, which contains a type (required), an image (required), a network mode (optional), mounted volumes (optional), parameters (optional) and whether Mesos should always pull the latest image before executing or not (optional).
 
 * Endpoint: __/scheduler/iso8601__ or __/scheduler/dependency__
 * Method: __POST__
@@ -229,6 +229,20 @@ Chronos by adding the `forcePullImage` boolean to your `container` configuration
 Chronos will default to not doing a `docker pull` if the image is already found on the executing node. The alternative approach is to use versions/tags for
 your images.
 
+There is also support for passing in arbitrary docker config options.
+
+```json
+{
+  "container": {
+    "type": "DOCKER",
+    "image": "libmesos/ubuntu",
+    "parameters": [
+      { "key": "a-docker-option", "value": "xxx" },
+      { "key": "b-docker-option", "value": "yyy" }
+    ]
+}
+```
+
 ## Updating Task Progress
 
 Task progress can be updated by providing the number of additional elements processed. This will increment the existing count of elements processed.

src/main/scala/org/apache/mesos/chronos/scheduler/jobs/Containers.scala

@@ -27,5 +27,6 @@ case class Volume(
 case class DockerContainer(
                             @JsonProperty image: String,
                             @JsonProperty volumes: Seq[Volume],
+                            @JsonProperty parameters: Seq[Parameter],
                             @JsonProperty network: NetworkMode = NetworkMode.HOST,
                             @JsonProperty forcePullImage: Boolean = false)

src/main/scala/org/apache/mesos/chronos/scheduler/jobs/Parameter.scala

@@ -0,0 +1,26 @@
+package org.apache.mesos.chronos.scheduler.jobs
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+import org.apache.mesos.{ Protos => mesos }
+
+/**
+ * Represents an environment variable definition for the job
+ */
+case class Parameter(
+    key: String,
+    value: String) {
+
+  def toProto(): mesos.Parameter =
+    mesos.Parameter.newBuilder
+      .setKey(key)
+      .setValue(value)
+      .build
+}
+object Parameter {
+  def apply(proto: mesos.Parameter): Parameter =
+    Parameter(
+      proto.getKey,
+      proto.getValue
+    )
+}

src/main/scala/org/apache/mesos/chronos/scheduler/mesos/MesosTaskBuilder.scala

@@ -170,6 +170,7 @@ class MesosTaskBuilder @Inject()(val conf: SchedulerConfiguration) {
       .setImage(job.container.image)
       .setNetwork(DockerInfo.Network.valueOf(job.container.network.toString.toUpperCase))
       .setForcePullImage(job.container.forcePullImage)
+      .addAllParameters(job.container.parameters.map(_.toProto).asJava)
       .build()).build
   }
 

src/main/scala/org/apache/mesos/chronos/utils/JobDeserializer.scala

@@ -171,7 +171,18 @@ class JobDeserializer extends JsonDeserializer[BaseJob] {
         if (containerNode.has("forcePullImage") && containerNode.get("forcePullImage") != null)
           Try(containerNode.get("forcePullImage").asText.toBoolean).getOrElse(false)
         else false
-      container = DockerContainer(containerNode.get("image").asText, volumes, networkMode, forcePullImage)
+
+
+      var parameters = scala.collection.mutable.ListBuffer[Parameter]()
+      if (containerNode.has("parameters")) {
+        containerNode.get("parameters").elements().map {
+          case node: ObjectNode =>
+          Parameter(node.get("key").asText(), node.get("value").asText)
+        }.foreach(parameters.add)
+      }
+
+
+      container = DockerContainer(containerNode.get("image").asText, volumes, parameters, networkMode, forcePullImage)
     }
 
     val constraints = scala.collection.mutable.ListBuffer[Constraint]()

src/main/scala/org/apache/mesos/chronos/utils/JobSerializer.scala

@@ -136,6 +136,19 @@ class JobSerializer extends JsonSerializer[BaseJob] {
       json.writeEndArray()
       json.writeFieldName("forcePullImage")
       json.writeBoolean(baseJob.container.forcePullImage)
+
+      json.writeFieldName("parameters")
+      json.writeStartArray()
+      baseJob.container.parameters.foreach { v =>
+        json.writeStartObject()
+        json.writeFieldName("key")
+        json.writeString(v.key)
+        json.writeFieldName("value")
+        json.writeString(v.value)
+        json.writeEndObject()
+      }
+      json.writeEndArray()
+
       json.writeEndObject()
     }
 

src/test/scala/org/apache/mesos/chronos/scheduler/api/SerDeTest.scala

@@ -29,7 +29,10 @@ class SerDeTest extends SpecificationWithJUnit {
       )
 
       val forcePullImage = false
-      val container = DockerContainer("dockerImage", volumes, NetworkMode.BRIDGE, forcePullImage)
+
+      var parameters = scala.collection.mutable.ListBuffer[Parameter]()
+
+      val container = DockerContainer("dockerImage", volumes, parameters, NetworkMode.BRIDGE, forcePullImage)
 
       val arguments = Seq(
         "-testOne"
@@ -69,7 +72,9 @@ class SerDeTest extends SpecificationWithJUnit {
       )
 
       val forcePullImage = true
-      val container = DockerContainer("dockerImage", volumes, NetworkMode.HOST, forcePullImage)
+      var parameters = scala.collection.mutable.ListBuffer[Parameter]()
+
+      val container = DockerContainer("dockerImage", volumes, parameters, NetworkMode.HOST, forcePullImage)
 
       val arguments = Seq(
         "-testOne"