empia / akka-mesos

Mesos library built on top of Akka.

akka-mesos

Non-blocking driver for Apache Mesos based on Akka.

Usage

Registering a Framework with Mesos

import akka.actor.ActorSystem
import akka.libprocess.PID
import akka.mesos.Mesos
import akka.mesos.protos.FrameworkInfo

implicit val system = ActorSystem("Mesos", config)

val frameworkInfo = FrameworkInfo(
  name = "ExampleFramework",
  user = "user",
  failoverTimeout = Some(5.minutes),
  checkpoint = Some(true)
)

val framework = Mesos(system).registerFramework(Success(PID("127.0.0.1", 5050, "master")), frameworkInfo)

The registerFramework method returns a Framework object. This object contains a SchedulerDriver and a Source[SchedulerMessage]. The SchedulerDriver can be used to send scheduling messages to Mesos, while the Source[SchedulerMessages] contains the scheduling messages that Mesos sends to the framework.

Example

import akka.mesos.protos.TaskInfo
import akka.mesos.scheduler.SchedulerPublisher.{ StatusUpdate, ResourceOffers }
import akka.stream.ActorFlowMaterializer

// necessary to process the message stream
implicit val materializer = ActorFlowMaterializer()

// process incoming messages with the given function
framework.schedulerMessages.runForeach {
  case ResourceOffers(offers) =>
    // create a TaskInfo
    val taskInfo = TaskInfo(...)

    // find a matching offer
    val matchingOffer = offer.find(...)

    // if a matching offer is found, start the task with the offer
    matchingOffer.foreach { offer =>
      driver.launchTasks(Seq(task), Seq(offer.id))
    }

    // don't forget to decline the offers that didn't match
    (offers diff matchingOffer.toSeq).foreach { offer =>
      driver.declineOffer(offer.id)
    }

  case StatusUpdate(update) =>
    println(s"Task ${update.status.taskId} id now ${update.status.state}")
    // Status updates must be acknowledged so Mesos knows, that we
    // received and sucessfully processed this message. Otherwise
    // Mesos would keep sending this message
    framework.driver.acknowledgeStatusUpdate(update)

  case _ => // We are not interested in other messages for now
}