added CapacityError to be used for failing task launch promises after some number of unmatched offer cycles (#40)

Author: tysonnorris

src/main/resources/reference.conf

@@ -2,4 +2,5 @@
 mesos-actor {
   agent-stats-ttl = 1 minutes //minimum time that agent stats remain valid without an update; incrase time in case offer cycle is slow; decrease time ignore offer data that may be stale
   agent-stats-pruning-period = 5 seconds //time period between pruning of expired agent stats
+  //fail-pending-offer-cycles = 2 //fail tasks that are not matched within this number of offer cycles
 }

src/main/scala/com/adobe/api/platform/runtime/mesos/MesosClient.scala

@@ -43,6 +43,7 @@ import org.apache.mesos.v1.scheduler.Protos.Call
 import org.apache.mesos.v1.scheduler.Protos.Call._
 import org.apache.mesos.v1.scheduler.Protos.Event
 import pureconfig._
+import pureconfig.loadConfigOrThrow
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.Buffer
@@ -119,7 +120,7 @@ case class CommandURIDef(uri: URI, extract: Boolean = true, cache: Boolean = fal
 
 //task states
 sealed abstract class TaskState()
-case class SubmitPending(reqs: TaskDef, promise: Promise[Running]) extends TaskState
+case class SubmitPending(reqs: TaskDef, promise: Promise[Running], offerCycles: Int = 1) extends TaskState
 case class Submitted(pending: SubmitPending,
                      taskInfo: TaskInfo,
                      offer: OfferID,
@@ -137,10 +138,19 @@ case class DeletePending(taskId: String, promise: Promise[Deleted]) extends Task
 case class Deleted(taskId: String, taskStatus: TaskStatus) extends TaskState
 case class Failed(taskId: String, agentId: String) extends TaskState
 
-case class MesosActorConfig(agentStatsTTL: FiniteDuration, agentStatsPruningPeriod: FiniteDuration)
+case class MesosActorConfig(agentStatsTTL: FiniteDuration,
+                            agentStatsPruningPeriod: FiniteDuration,
+                            failPendingOfferCycles: Option[Int])
 case class AgentStats(mem: Double, cpu: Double, ports: Int, expiration: Instant)
 
 case class MesosAgentStats(stats: Map[String, AgentStats])
+
+case class CapacityFailure(requiredMem: Float,
+                           requiredCpu: Float,
+                           requiredPorts: Int,
+                           remainingResources: List[(Float, Float, Int)])
+    extends MesosException("cluster does not have capacity")
+
 //TODO: mesos authentication
 trait MesosClientActor extends Actor with ActorLogging with MesosClientConnection {
   implicit val ec: ExecutionContext = context.dispatcher
@@ -167,7 +177,7 @@ trait MesosClientActor extends Actor with ActorLogging with MesosClientConnectio
   var agentOfferHistory = Map.empty[String, AgentStats] // Map[<agent hostname> -> <stats>] track the most recent offer stats per agent hostname
   val listener: Option[ActorRef]
 
-  val config = loadConfigOrThrow[MesosActorConfig]("mesos-actor")
+  val config: MesosActorConfig
 
   if (autoSubscribe) {
     log.info(s"auto-subscribing ${self} to mesos master at ${master}")
@@ -191,13 +201,14 @@ trait MesosClientActor extends Actor with ActorLogging with MesosClientConnectio
   }
   case object PruneStats
 
-  val statsPruner =
+  override def preStart() = {
     actorSystem.scheduler.schedule(30.seconds, config.agentStatsPruningPeriod, context.actorOf(Props(new Actor {
       override def receive: Receive = {
         case PruneStats =>
           context.parent ! PruneStats //client actor needs to handle PruneStats to avoid concurrent update to stats map
       }
     })), PruneStats)
+  }
   //cache the framework id, so that in case this actor restarts we can reconnect
   if (MesosClient.frameworkID.isEmpty) MesosClient.frameworkID = Some(FrameworkID.newBuilder().setValue(id()).build())
   private val frameworkID = MesosClient.frameworkID.get
@@ -234,7 +245,7 @@ trait MesosClientActor extends Actor with ActorLogging with MesosClientConnectio
       tasks.get(taskID.getValue) match {
         case Some(taskDetails) =>
           taskDetails match {
-            case SubmitPending(taskDef, promise) => {
+            case SubmitPending(taskDef, promise, _) => {
               log.info(s"deleting unlaunched task ${taskDef.taskId}")
               tasks.remove(taskDef.taskId)
             }
@@ -434,7 +445,8 @@ trait MesosClientActor extends Actor with ActorLogging with MesosClientConnectio
 
       log.debug(s"agent offer stats: {}", agentOfferHistory)
 
-      val matchedTasks = taskMatcher.matchTasksToOffers(role, pending, event.getOffersList.asScala.toList, taskBuilder)
+      val (matchedTasks, remaining) =
+        taskMatcher.matchTasksToOffers(role, pending, event.getOffersList.asScala.toList, taskBuilder)
 
       val matchedCount = matchedTasks.foldLeft(0)(_ + _._2.size)
       log.info(s"matched ${matchedCount} tasks to ${matchedTasks.size} offers out of ${pending.size} pending tasks")
@@ -474,7 +486,7 @@ trait MesosClientActor extends Actor with ActorLogging with MesosClientConnectio
             matchedTasks.foreach(entry => {
               entry._2.foreach(task => {
                 tasks(task._1.getTaskId.getValue) match {
-                  case s @ SubmitPending(reqs, promise) =>
+                  case s @ SubmitPending(reqs, promise, _) =>
                     //dig the hostname out of the offer whose agent id matches the agent id in the task info
                     val hostname =
                       event.getOffersList.asScala.find(p => p.getAgentId == task._1.getAgentId).get.getHostname
@@ -493,6 +505,27 @@ trait MesosClientActor extends Actor with ActorLogging with MesosClientConnectio
         }
 
       }
+      //generate failures for pending tasks that did not fit any offers
+      config.failPendingOfferCycles.foreach { maxOfferCycles =>
+        val submitPending = tasks.collect { case (_, s: SubmitPending) => s }
+        if (submitPending.nonEmpty) {
+          submitPending.foreach { task =>
+            //println(s"task offerCycles:${task.offerCycles}   ${maxOfferCycles}")
+            if (task.offerCycles > maxOfferCycles) {
+              log.warning(s"failing task ${task.reqs.taskId} after ${task.offerCycles} unmatching offer cycles")
+              task.promise.failure(
+                new CapacityFailure(
+                  task.reqs.mem.toFloat,
+                  task.reqs.cpus.toFloat,
+                  task.reqs.ports.size,
+                  remaining.values.toList))
+              tasks.remove(task.reqs.taskId)
+            } else {
+              tasks.update(task.reqs.taskId, task.copy(offerCycles = task.offerCycles + 1)) //increase the offer cycles this task has seen
+            }
+          }
+        }
+      }
 
       //store a reference of last memory offer (total) from each agent
       val newOfferStats = MesosClient.getOfferStats(config, role, agentOfferMap)
@@ -691,7 +724,8 @@ class MesosClient(val id: () => String,
                   val tasks: TaskStore,
                   val refuseSeconds: Double,
                   val heartbeatMaxFailures: Int,
-                  val listener: Option[ActorRef])
+                  val listener: Option[ActorRef],
+                  val config: MesosActorConfig)
     extends MesosClientActor
     with MesosClientHttpConnection {}
 
@@ -710,7 +744,8 @@ object MesosClient {
             taskStore: TaskStore,
             refuseSeconds: Double = 5.0,
             heartbeatMaxFailures: Int = 2,
-            listener: Option[ActorRef] = None): Props =
+            listener: Option[ActorRef] = None,
+            config: MesosActorConfig = loadConfigOrThrow[MesosActorConfig]("mesos-actor")): Props =
     Props(
       new MesosClient(
         id,
@@ -724,7 +759,8 @@ object MesosClient {
         taskStore,
         refuseSeconds,
         heartbeatMaxFailures,
-        listener))
+        listener,
+        config))
 
   //TODO: allow task persistence/reconcile
 

src/main/scala/com/adobe/api/platform/runtime/mesos/TaskMatcher.scala

@@ -23,15 +23,16 @@ import org.apache.mesos.v1.Protos.TaskInfo
 import org.apache.mesos.v1.Protos.Value
 import org.apache.mesos.v1.Protos.Value.Ranges
 import scala.collection.JavaConverters._
+import scala.collection.mutable.Buffer
 import scala.collection.mutable.ListBuffer
 
 trait TaskMatcher {
   def matchTasksToOffers(role: String, t: Iterable[TaskDef], o: Iterable[Offer], builder: TaskBuilder)(
-    implicit logger: LoggingAdapter): Map[OfferID, Seq[(TaskInfo, Seq[Int])]]
+    implicit logger: LoggingAdapter): (Map[OfferID, Seq[(TaskInfo, Seq[Int])]], Map[OfferID, (Float, Float, Int)])
 }
 class DefaultTaskMatcher(isValid: Offer => Boolean = _ => true) extends TaskMatcher {
   override def matchTasksToOffers(role: String, t: Iterable[TaskDef], o: Iterable[Offer], builder: TaskBuilder)(
-    implicit logger: LoggingAdapter): Map[OfferID, Seq[(TaskInfo, Seq[Int])]] = {
+    implicit logger: LoggingAdapter): (Map[OfferID, Seq[(TaskInfo, Seq[Int])]], Map[OfferID, (Float, Float, Int)]) = {
     //we can launch many tasks on a single offer
 
     var tasksInNeed: ListBuffer[TaskDef] = t.to[ListBuffer]
@@ -40,7 +41,7 @@ class DefaultTaskMatcher(isValid: Offer => Boolean = _ => true) extends TaskMatc
       _.getResourcesList.asScala
         .find(_.getName == "cpus")
         .map(_.getScalar.getValue))
-
+    var remaining: Map[OfferID, (Float, Float, Int)] = Map.empty
     sortedOffers.foreach(offer => {
       try {
         //for testing worst case scenario...
@@ -68,6 +69,17 @@ class DefaultTaskMatcher(isValid: Offer => Boolean = _ => true) extends TaskMatc
             var remainingOfferMem = scalarResources("mem")
             var usedPorts = ListBuffer[Int]()
             var acceptedTasks = ListBuffer[(TaskInfo, Seq[Int])]()
+            //check for a good fit
+            //collect ranges from ports resources
+            val offerPortsRanges = offer.getResourcesList.asScala
+              .filter(res => res.getName == "ports")
+              .filter(_.getRole == role) //ignore resources with other roles
+              .map(res => res.getRanges)
+
+            //before matching, consider all resources remaining
+            remaining = remaining + (offer.getId -> (remainingOfferMem.toFloat, remainingOfferCpus.toFloat, countPorts(
+              offerPortsRanges) - usedPorts.size))
+
             tasksInNeed.map(task => {
 
               val taskCpus = task.cpus
@@ -86,12 +98,6 @@ class DefaultTaskMatcher(isValid: Offer => Boolean = _ => true) extends TaskMatc
                 }
               })
               logger.debug(s"constraintChecks ${constraintChecks}")
-              //check for a good fit
-              //collect ranges from ports resources
-              val offerPortsRanges = offer.getResourcesList.asScala
-                .filter(res => res.getName == "ports")
-                .filter(_.getRole == role) //ignore resources with other roles
-                .map(res => res.getRanges)
               //pluck the number of ports needed for this task
               val hostPorts = pluckPorts(offerPortsRanges, task.ports.size, usedPorts)
               val matchedResources = remainingOfferCpus > taskCpus &&
@@ -105,12 +111,13 @@ class DefaultTaskMatcher(isValid: Offer => Boolean = _ => true) extends TaskMatc
                 logger.info(
                   s"offer did not match resource requirements cpu:${taskCpus} (${remainingOfferCpus}), mem: ${taskMem}  (${remainingOfferMem}), ports: ${task.ports.size} (${hostPorts.size})")
               } else {
-
                 //mark resources as used
                 remainingOfferCpus -= taskCpus
                 remainingOfferMem -= taskMem
                 usedPorts ++= hostPorts
-
+                //update remaining for all resource usage
+                remaining = remaining + (offer.getId -> (remainingOfferMem.toFloat, remainingOfferCpus.toFloat, countPorts(
+                  offerPortsRanges) - usedPorts.size))
                 //build port mappings
                 val portMappings =
                   if (task.ports.isEmpty) List()
@@ -180,7 +187,7 @@ class DefaultTaskMatcher(isValid: Offer => Boolean = _ => true) extends TaskMatc
         case t: Exception => logger.error(s"task matching failed, ignoring offer ${offer.getId} ${t}")
       }
     })
-    result
+    (result, remaining)
   }
 
   def pluckPorts(rangesList: Iterable[org.apache.mesos.v1.Protos.Value.Ranges],
@@ -201,4 +208,7 @@ class DefaultTaskMatcher(isValid: Offer => Boolean = _ => true) extends TaskMatc
     })
     ports.toList
   }
+  def countPorts(portRanges: Buffer[Ranges]) = {
+    portRanges.foldLeft(0)(_ + _.getRangeList.asScala.foldLeft(0)((a, b) => a + b.getEnd.toInt - b.getBegin.toInt + 1))
+  }
 }

src/test/resources/emptyoffer.json

@@ -0,0 +1,83 @@
+{
+  "offers": [{
+    "id": {
+      "value": "7168e411-c3e4-4e29-b292-9b12eda4aaca-O58"
+    },
+    "frameworkId": {
+      "value": "sample-11fdf320-311f-42b0-a284-dbc38bb8d191"
+    },
+    "agentId": {
+      "value": "db6b062d-84e3-4a2e-a8c5-98ffa944a304-S0"
+    },
+    "hostname": "192.168.99.100",
+    "resources": [{
+      "name": "ports",
+      "type": "RANGES",
+      "ranges": {
+        "range": [{
+          "begin": "11001",
+          "end": "11199"
+        }]
+      },
+      "role": "*",
+      "allocationInfo": {
+        "role": "*"
+      }
+    }, {
+      "name": "cpus",
+      "type": "SCALAR",
+      "scalar": {
+        "value": 0.0
+      },
+      "role": "*",
+      "allocationInfo": {
+        "role": "*"
+      }
+    }, {
+      "name": "mem",
+      "type": "SCALAR",
+      "scalar": {
+        "value": 0.0
+      },
+      "role": "*",
+      "allocationInfo": {
+        "role": "*"
+      }
+    }, {
+      "name": "disk",
+      "type": "SCALAR",
+      "scalar": {
+        "value": 0.0
+      },
+      "role": "*",
+      "allocationInfo": {
+        "role": "*"
+      }
+    }],
+    "attributes": [{
+      "name": "att1",
+      "type": "TEXT",
+      "text": {
+        "value": "att1valueslave1"
+      }
+    }, {
+      "name": "att2",
+      "type": "TEXT",
+      "text": {
+        "value": "att2valueslave1"
+      }
+    }],
+    "url": {
+      "scheme": "http",
+      "address": {
+        "hostname": "192.168.99.100",
+        "ip": "192.168.99.100",
+        "port": 5051
+      },
+      "path": "/slave(1)"
+    },
+    "allocationInfo": {
+      "role": "*"
+    }
+  }]
+}