Support Mesos fetcher cache (and other options).

docs/docs/api.md

@@ -222,7 +222,7 @@ curl -L -H 'Content-Type: application/json' -X POST -d '{json hash}' chronos-nod
   },
   "cpus": "0.5",
   "mem": "512",
-  "uris": [],
+  "fetch": [],
   "command": "while sleep 10; do date =u %T; done"
 }
 ```
@@ -335,7 +335,8 @@ When specifying the `command` field in your job hash, use `url-runner.bash` (mak
 | `mem`                 | Amount of Mesos Memory (in MB) for this job.                                                               | `128` or `--mesos_task_mem`    |
 | `disk`                | Amount of Mesos disk (in MB) for this job.                                                                 | `256` or `--mesos_task_disk`   |
 | `disabled`            | If set to `true`, this job will not be run.                                                              | `false`                        |
-| `uris`                | An array of URIs which Mesos will download when the task is started.                                     | -                              |
+| `uris`                | An array of URIs which Mesos will download when the task is started (deprecated).                         | -                             |
+| `fetch`               | An array of fetch configurations, one for each file that Mesos Fetcher will download when the task is started).| -                        |
 | `schedule`            | [ISO 8601][] repeating schedule for this job.  If specified, `parents` must not be specified.            | -                              |
 | `scheduleTimeZone`    | The time zone for the given schedule, specified in the [tz database](https://en.wikipedia.org/wiki/Tz_database) format. | -                              |
 | `parents`             | An array of parent jobs for a dependent job.  If specified, `schedule` must not be specified.            | -                              |
@@ -370,7 +371,14 @@ When specifying the `command` field in your job hash, use `url-runner.bash` (mak
   "disk": 10240,
   "mem": 1024,
   "disabled": false,
-  "uris": [],
+  "fetch": [
+    {
+      "uri": "https://url-to-file",
+      "cache": false,
+      "extract": false,
+      "executable": false
+    }
+  ],
   "schedule": "R/2014-03-08T20:00:00.000Z/PT2H",
   "environmentVariables": [
     {

pom.xml

@@ -31,7 +31,7 @@
     <joda-convert.version>1.7</joda-convert.version>
     <joda-time.version>2.3</joda-time.version>
     <lz4.version>1.2.0</lz4.version>
-    <mesos-utils.version>0.22.0-1</mesos-utils.version>
+    <mesos-utils.version>0.23.0</mesos-utils.version>
     <metrics.version>3.1.0</metrics.version>
     <raven.version>4.1.2</raven.version>
 

src/main/scala/org/apache/mesos/chronos/notification/HttpClient.scala

@@ -47,7 +47,8 @@ class HttpClient(val endpointUrl: String,
     generator.writeStringField("mem", job.mem.toString())
     generator.writeStringField("retries", job.retries.toString())
     generator.writeStringField("successCount", job.successCount.toString())
-    generator.writeStringField("uris", job.uris.mkString(","))
+    val uris = job.fetch.map { _.uri } ++ job.uris
+    generator.writeStringField("uris", uris.mkString(","))
 
 
     generator.writeEndObject()

src/main/scala/org/apache/mesos/chronos/notification/RavenClient.scala

@@ -17,6 +17,7 @@ class RavenClient(val dsn: String) extends NotificationClient {
 
   def sendNotification(job: BaseJob, to: String, subject: String, message: Option[String]) {
     val ravenMessage = subject + "\n\n" + message.getOrElse("")
+    val uris = job.fetch.map { _.uri } ++ job.uris
     val event = new EventBuilder()
       .setMessage(ravenMessage)
       .setLevel(Event.Level.ERROR)
@@ -36,7 +37,7 @@ class RavenClient(val dsn: String) extends NotificationClient {
       .addExtra("mem", job.mem)
       .addExtra("retries", job.retries)
       .addExtra("successCount", job.successCount)
-      .addExtra("uris", job.uris.mkString(","))
+      .addExtra("uris", uris.mkString(","))
       .build()
 
     raven.sendEvent(event)

src/main/scala/org/apache/mesos/chronos/scheduler/api/Iso8601JobResource.scala

@@ -46,6 +46,10 @@ class Iso8601JobResource @Inject()(
           "the job's name is invalid. Allowed names: '%s'".format(JobUtils.jobNamePattern.toString()))
         if (!Iso8601Expressions.canParse(newJob.schedule, newJob.scheduleTimeZone))
           return Response.status(Response.Status.BAD_REQUEST).build()
+        if(! JobUtils.isValidURIDefinition(newJob)) {
+          log.warning(s"Tried to add both uri (deprecated) and fetch parameters on ${newJob.name}")
+          return Response.status(Response.Status.BAD_REQUEST).build()
+        }
 
         //TODO(FL): Create a wrapper class that handles adding & removing jobs!
         jobScheduler.registerJob(List(newJob), persist = true)

src/main/scala/org/apache/mesos/chronos/scheduler/api/JobManagementResource.scala

@@ -91,6 +91,7 @@ class JobManagementResource @Inject()(val jobScheduler: JobScheduler,
                       disabled = childJob.disabled,
                       softError = childJob.softError,
                       uris = childJob.uris,
+                      fetch = childJob.fetch,
                       highPriority = childJob.highPriority
                     )
                     jobScheduler.updateJob(childJob, newChild)
@@ -201,13 +202,19 @@ class JobManagementResource @Inject()(val jobScheduler: JobScheduler,
   @Timed
   def list(): Response = {
     try {
-      val jobs = ListBuffer[BaseJob]()
       import scala.collection.JavaConversions._
-      jobGraph.dag.vertexSet().map({
-        job =>
-          jobs += jobGraph.getJobForName(job).get
-      })
-      Response.ok(jobs.toList).build
+      val jobs = jobGraph.dag.vertexSet()
+        .map { jobGraph.getJobForName }
+        .flatten
+        .map { // copies fetch in uris or uris in fetch (only one can be set) __only__ in REST get, for compatibility
+          case j : ScheduleBasedJob  =>
+            if(j.fetch.isEmpty) j.copy(fetch = j.uris.map { Fetch(_) })
+            else                j.copy(uris =  j.fetch.map { _.uri })
+          case j : DependencyBasedJob =>
+            if(j.fetch.isEmpty) j.copy(fetch = j.uris.map { Fetch(_) })
+            else                j.copy(uris =  j.fetch.map { _.uri })
+        }
+      Response.ok(jobs).build
     } catch {
       case ex: Exception =>
         log.log(Level.WARNING, "Exception while serving request", ex)

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

@@ -0,0 +1,12 @@
+package org.apache.mesos.chronos.scheduler.jobs
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+/**
+  * Created by Sylvain Veyrié on 17/02/2016.
+  */
+case class Fetch(
+                  @JsonProperty uri: String,
+                  @JsonProperty executable: Boolean = false,
+                  @JsonProperty cache: Boolean = false,
+                  @JsonProperty extract: Boolean = false)

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

@@ -56,6 +56,8 @@ object JobUtils {
     }
   }
 
+  def isValidURIDefinition(baseJob: BaseJob) = baseJob.uris.isEmpty || baseJob.fetch.isEmpty  // when you leave the deprecated one, then it should be empty
+
   //TODO(FL): Think about moving this back into the JobScheduler, though it might be a bit crowded.
   def loadJobs(scheduler: JobScheduler, store: PersistenceStore) {
     //TODO(FL): Create functions that map strings to jobs
@@ -183,6 +185,7 @@ object JobUtils {
           errorsSinceLastSuccess = job.errorsSinceLastSuccess,
           softError = job.softError,
           uris = job.uris,
+          fetch = job.fetch,
           highPriority = job.highPriority,
           runAsUser = job.runAsUser,
           container = job.container,
@@ -214,6 +217,7 @@ object JobUtils {
           errorsSinceLastSuccess = job.errorsSinceLastSuccess,
           softError = job.softError,
           uris = job.uris,
+          fetch = job.fetch,
           highPriority = job.highPriority,
           runAsUser = job.runAsUser,
           container = job.container,

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

@@ -61,8 +61,11 @@ trait BaseJob {
 
   def errorsSinceLastSuccess: Long = 0L
 
+  @Deprecated
   def uris: Seq[String] = List()
 
+  def fetch: Seq[Fetch] = List()
+
   def highPriority: Boolean = false
 
   def runAsUser: String = ""
@@ -104,7 +107,8 @@ case class ScheduleBasedJob(
                              @JsonProperty override val mem: Double = 0,
                              @JsonProperty override val disabled: Boolean = false,
                              @JsonProperty override val errorsSinceLastSuccess: Long = 0L,
-                             @JsonProperty override val uris: Seq[String] = List(),
+                             @Deprecated @JsonProperty override val uris: Seq[String] = List(),
+                             @JsonProperty override val fetch: Seq[Fetch] = List(),
                              @JsonProperty override val highPriority: Boolean = false,
                              @JsonProperty override val runAsUser: String = "",
                              @JsonProperty override val container: DockerContainer = null,
@@ -140,7 +144,8 @@ case class DependencyBasedJob(
                                @JsonProperty override val mem: Double = 0,
                                @JsonProperty override val disabled: Boolean = false,
                                @JsonProperty override val errorsSinceLastSuccess: Long = 0L,
-                               @JsonProperty override val uris: Seq[String] = List(),
+                               @Deprecated @JsonProperty override val uris: Seq[String] = List(),
+                               @JsonProperty override val fetch: Seq[Fetch] = List(),
                                @JsonProperty override val highPriority: Boolean = false,
                                @JsonProperty override val runAsUser: String = "",
                                @JsonProperty override val container: DockerContainer = null,

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

@@ -4,7 +4,7 @@ import java.util.logging.Logger
 import javax.inject.Inject
 
 import org.apache.mesos.chronos.scheduler.config.SchedulerConfiguration
-import org.apache.mesos.chronos.scheduler.jobs.BaseJob
+import org.apache.mesos.chronos.scheduler.jobs.{Fetch, BaseJob}
 import com.google.common.base.Charsets
 import com.google.protobuf.ByteString
 import org.apache.mesos.Protos.ContainerInfo.DockerInfo
@@ -83,14 +83,18 @@ class MesosTaskBuilder @Inject()(val conf: SchedulerConfiguration) {
       )
     }
 
-    val uriProtos = job.uris.map { uri =>
+    val fetch = job.fetch ++ job.uris.map { Fetch(_) }
+    val uriCommand = fetch.map { f =>
       CommandInfo.URI.newBuilder()
-        .setValue(uri)
+        .setValue(f.uri)
+        .setExtract(f.extract)
+        .setExecutable(f.executable)
+        .setCache(f.cache)
         .build()
     }
 
     if (job.executor.nonEmpty) {
-      appendExecutorData(taskInfo, job, environment, uriProtos)
+      appendExecutorData(taskInfo, job, environment, uriCommand)
     } else {
       val command = CommandInfo.newBuilder()
       if (job.command.startsWith("http") || job.command.startsWith("ftp")) {
@@ -106,7 +110,7 @@ class MesosTaskBuilder @Inject()(val conf: SchedulerConfiguration) {
           .setShell(job.shell)
           .setEnvironment(environment)
           .addAllArguments(job.arguments.asJava)
-          .addAllUris(uriProtos.asJava)
+          .addAllUris(uriCommand.asJava)
       }
       if (job.runAsUser.nonEmpty) {
         command.setUser(job.runAsUser)

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

@@ -121,6 +121,19 @@ class JobDeserializer extends JsonDeserializer[BaseJob] {
       }
     }
 
+    var fetch = scala.collection.mutable.ListBuffer[Fetch]()
+    if(node.has("fetch")) {
+      node.get("fetch").elements().map {
+        case node: ObjectNode => {
+          val uri = Option(node.get("uri")).map { _.asText() }.getOrElse("")
+          val executable = Option(node.get("executable")).map { _.asBoolean() }.getOrElse(false)
+          val cache = Option(node.get("cache")).map { _.asBoolean() }.getOrElse(false)
+          val extract = Option(node.get("extract")).map { _.asBoolean() }.getOrElse(false)
+          Fetch(uri, executable, cache, extract)
+        }
+      }.foreach(fetch.add)
+    }
+
     var arguments = scala.collection.mutable.ListBuffer[String]()
     if (node.has("arguments")) {
       for (argument <- node.path("arguments")) {
@@ -206,7 +219,7 @@ class JobDeserializer extends JsonDeserializer[BaseJob] {
         executor = executor, executorFlags = executorFlags, retries = retries, owner = owner,
         ownerName = ownerName, description = description, lastError = lastError, lastSuccess = lastSuccess,
         async = async, cpus = cpus, disk = disk, mem = mem, disabled = disabled,
-        errorsSinceLastSuccess = errorsSinceLastSuccess, uris = uris, highPriority = highPriority,
+        errorsSinceLastSuccess = errorsSinceLastSuccess, fetch = fetch, uris = uris, highPriority = highPriority,
         runAsUser = runAsUser, container = container, environmentVariables = environmentVariables, shell = shell,
         arguments = arguments, softError = softError, dataProcessingJobType = dataProcessingJobType,
         constraints = constraints)
@@ -217,7 +230,7 @@ class JobDeserializer extends JsonDeserializer[BaseJob] {
         executorFlags = executorFlags, retries = retries, owner = owner, ownerName = ownerName,
         description = description, lastError = lastError, lastSuccess = lastSuccess, async = async,
         cpus = cpus, disk = disk, mem = mem, disabled = disabled,
-        errorsSinceLastSuccess = errorsSinceLastSuccess, uris = uris,  highPriority = highPriority,
+        errorsSinceLastSuccess = errorsSinceLastSuccess, fetch = fetch, uris = uris,  highPriority = highPriority,
         runAsUser = runAsUser, container = container, scheduleTimeZone = scheduleTimeZone,
         environmentVariables = environmentVariables, shell = shell, arguments = arguments, softError = softError,
         dataProcessingJobType = dataProcessingJobType, constraints = constraints)
@@ -227,7 +240,7 @@ class JobDeserializer extends JsonDeserializer[BaseJob] {
         errorCount = errorCount, executor = executor, executorFlags = executorFlags, retries = retries, owner = owner,
         ownerName = ownerName, description = description, lastError = lastError, lastSuccess = lastSuccess,
         async = async, cpus = cpus, disk = disk, mem = mem, disabled = disabled,
-        errorsSinceLastSuccess = errorsSinceLastSuccess, uris = uris,  highPriority = highPriority,
+        errorsSinceLastSuccess = errorsSinceLastSuccess, fetch = fetch, uris = uris,  highPriority = highPriority,
         runAsUser = runAsUser, container = container, environmentVariables = environmentVariables, shell = shell,
         arguments = arguments, softError = softError, dataProcessingJobType = dataProcessingJobType,
         constraints = constraints)

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

@@ -79,6 +79,22 @@ class JobSerializer extends JsonSerializer[BaseJob] {
     json.writeFieldName("errorsSinceLastSuccess")
     json.writeNumber(baseJob.errorsSinceLastSuccess)
 
+    json.writeFieldName("fetch")
+    json.writeStartArray()
+    baseJob.fetch.foreach { f =>
+      json.writeStartObject()
+      json.writeFieldName("uri")
+      json.writeString(f.uri)
+      json.writeFieldName("executable")
+      json.writeBoolean(f.executable)
+      json.writeFieldName("cache")
+      json.writeBoolean(f.cache)
+      json.writeFieldName("extract")
+      json.writeBoolean(f.extract)
+      json.writeEndObject()
+    }
+    json.writeEndArray()
+
     json.writeFieldName("uris")
     json.writeStartArray()
     baseJob.uris.foreach(json.writeString)

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

@@ -43,10 +43,12 @@ class SerDeTest extends SpecificationWithJUnit {
         LikeConstraint("rack", "rack-[1-3]")
       )
 
+      val fetch = Seq(Fetch("https://mesos.github.io/chronos/", true, false, true))
+
       val a = new DependencyBasedJob(Set("B", "C", "D", "E"), "A", "noop", Minutes.minutes(5).toPeriod, 10L,
         20L, "fooexec", "fooflags", 7, "foo@bar.com", "Foo", "Test dependency-based job", "TODAY",
         "YESTERDAY", true, container = container, environmentVariables = environmentVariables,
-        shell = false, arguments = arguments, softError = true, constraints = constraints)
+        shell = false, arguments = arguments, softError = true, constraints = constraints, fetch = fetch)
 
       val aStr = objectMapper.writeValueAsString(a)
       val aCopy = objectMapper.readValue(aStr, classOf[DependencyBasedJob])
@@ -85,10 +87,12 @@ class SerDeTest extends SpecificationWithJUnit {
         LikeConstraint("rack", "rack-[1-3]")
       )
 
+      val fetch = Seq(Fetch("https://mesos.github.io/chronos/", true, false, true))
+
       val a = new ScheduleBasedJob("FOO/BAR/BAM", "A", "noop", Minutes.minutes(5).toPeriod, 10L, 20L,
         "fooexec", "fooflags", 7, "foo@bar.com", "Foo", "Test schedule-based job", "TODAY",
         "YESTERDAY", true, container = container, environmentVariables = environmentVariables,
-        shell = true, arguments = arguments, softError = true, constraints = constraints)
+        shell = true, arguments = arguments, softError = true, constraints = constraints, fetch = fetch)
 
       val aStr = objectMapper.writeValueAsString(a)
       val aCopy = objectMapper.readValue(aStr, classOf[ScheduleBasedJob])