//设置package包名称以及导入依赖的类
package com.galacticfog.lambda.scheduler

import java.util.UUID

import org.apache.mesos.{Protos, MesosSchedulerDriver}
import org.apache.mesos.Protos.FrameworkInfo
import play.api.{Logger, Play, Application, GlobalSettings}
import scala.collection.mutable
import scala.concurrent.Future
import scala.sys.SystemProperties
import scala.concurrent.duration._
import Play.current
import play.api.libs.concurrent.Execution.Implicits._
import scala.util.Try

case class CmdTask(uuid: UUID, handlerMethod: String, jarUrl: String)

object Global extends GlobalSettings {

  lazy val driver: Try[MesosSchedulerDriver] = Try {
    Logger.info("creating LambdaScheduler")
    val scheduler = new LambdaScheduler

    val master = current.configuration.getString("master") getOrElse "localhost:5050"
    Logger.info(s"registering with mesos-master: ${master}")

    val schedulerHostname = current.configuration.getString("hostname") getOrElse java.net.InetAddress.getLocalHost.getHostName
    Logger.info(s"scheduler on: ${schedulerHostname}")

    val frameworkInfoBuilder = FrameworkInfo.newBuilder()
      .setName("gestalt-lambda-scheduler")
      .setFailoverTimeout(60.seconds.toMillis)
      .setUser("")
      .setCheckpoint(true)
      .setHostname(schedulerHostname)

    val frameworkInfo = frameworkInfoBuilder.build()
    Logger.info(s"scheduler on: ${schedulerHostname}")
    val implicitAcknowledgements = false

    new MesosSchedulerDriver( scheduler, frameworkInfo, master, implicitAcknowledgements )
  }

  val taskQueue = new mutable.Queue[CmdTask]

  override def onStart(app: Application): Unit = {
    Logger.info("onStart")
    driver foreach { d =>
      Logger.info("starting driver")
      Future { d.run } map println
    }
  }

  override def onStop(app: Application): Unit = {
    Logger.info("onStop")
    driver foreach {
      Logger.info("stopping driver")
      _.stop()
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.galacticfog.gestalt.lambda.impl

import java.util.concurrent.TimeoutException

import org.apache.mesos.Protos
import org.joda.time.DateTime
import scala.collection.mutable
import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

case class EnvironmentCacheEntry( lambdaId : String, env : Protos.Environment, queuedTime : DateTime = DateTime.now )
class EnvironmentCache {

  val cache : mutable.Map[String, EnvironmentCacheEntry] = mutable.Map[String, EnvironmentCacheEntry]()
  val EXPIRATION_SECONDS = sys.env.getOrElse( "ENV_CACHE_EXPIRATION_SECONDS", "900" ).toInt

  def getEnvironment( lambdaId : String, env : Future[Map[String,String]] ) : Protos.Environment = {

    val cacheEntry = cache.get( lambdaId )

    if( !cacheEntry.isDefined || cacheEntry.get.queuedTime.plusSeconds( EXPIRATION_SECONDS ).isBeforeNow ) {
      //wait for the future
      try {
        val result = Await.result( env, 5 seconds )
        val builder = Protos.Environment.newBuilder
        result.foreach{ entry =>
          builder.addVariables( Protos.Environment.Variable.newBuilder
            .setName( entry._1 )
            .setValue( entry._2 )
          )
        }

        val newEnv =  builder.build
        cache( lambdaId ) = new EnvironmentCacheEntry( lambdaId, newEnv )
        newEnv
      }
      catch {
        case ex : TimeoutException => {
          println( "TIMEOUT" )
          Protos.Environment.newBuilder.build
        }
      }
    }
    else {
      cache( lambdaId ).env
    }
  }
} 

//设置package包名称以及导入依赖的类
package mesosphere.util.state

import mesosphere.marathon.state.{ MarathonState, Timestamp }
import org.apache.mesos.Protos
import org.apache.mesos.Protos.FrameworkID

//TODO: move logic from FrameworkID to FrameworkId (which also implies moving this class)
case class FrameworkId(id: String) extends MarathonState[Protos.FrameworkID, FrameworkId] {
  override def mergeFromProto(message: FrameworkID): FrameworkId = {
    FrameworkId(message.getValue)
  }
  override def mergeFromProto(bytes: Array[Byte]): FrameworkId = {
    mergeFromProto(Protos.FrameworkID.parseFrom(bytes))
  }
  override def toProto: FrameworkID = {
    Protos.FrameworkID.newBuilder().setValue(id).build()
  }
  override def version: Timestamp = Timestamp.zero
}

object FrameworkId {
  def fromProto(message: FrameworkID): FrameworkId = new FrameworkId(message.getValue)
  def fromProtoBytes(bytes: Array[Byte]): FrameworkId = fromProto(Protos.FrameworkID.parseFrom(bytes))
} 

//设置package包名称以及导入依赖的类
package mesosphere.mesos

import mesosphere.marathon.state.{ DiskSource, PersistentVolume }
import mesosphere.mesos.protos.Resource
import org.apache.mesos.Protos
import org.apache.mesos.Protos.Resource.DiskInfo.Source

// TODO - put this somewhere sensible
object ResourceHelpers {
  import mesosphere.marathon.api.v2.json.Formats.ConstraintFormat

  def requestedStringification(requested: Either[Double, PersistentVolume]): String = requested match {
    case Left(value) => s"disk:root:$value"
    case Right(vol) =>
      val constraintsJson = vol.persistent.constraints.map(ConstraintFormat.writes).toList
      s"disk:${vol.persistent.`type`.toString}:${vol.persistent.size}:[${constraintsJson.mkString(",")}]"
  }

  implicit class DiskRichResource(resource: Protos.Resource) {
    def getSourceOption: Option[Source] =
      if (resource.hasDisk && resource.getDisk.hasSource)
        Some(resource.getDisk.getSource)
      else
        None

    def getStringification: String = {
      require(resource.getName == Resource.DISK)
      val diskSource = DiskSource.fromMesos(getSourceOption)
      
      (List(
        resource.getName,
        diskSource.diskType.toString,
        resource.getScalar.getValue.toString) ++
        diskSource.path).mkString(":")
    }

    def afterAllocation(amount: Double): Option[Protos.Resource] = {
      val isMountDiskResource: Boolean =
        resource.hasDisk && resource.getDisk.hasSource &&
          (resource.getDisk.getSource.getType == Source.Type.MOUNT)

      if (isMountDiskResource || amount >= resource.getScalar.getValue)
        None
      else
        Some(
          resource.toBuilder.
          setScalar(
            Protos.Value.Scalar.newBuilder.
              setValue(resource.getScalar.getValue - amount)).
            build)
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.github.roberveral.mesosgocd.scheduler

import java.util

import com.github.roberveral.mesosgocd.spec.Task
import com.github.roberveral.mesosgocd.utils.OfferUtils
import org.apache.mesos.Protos
import org.apache.mesos.{Scheduler, SchedulerDriver}

import mesosphere.mesos.protos.Implicits._
import com.github.roberveral.mesosgocd.spec.Implicits._
import scala.collection.JavaConverters._


class GoCDScheduler(task: Task) extends Scheduler with OfferUtils {
  override def offerRescinded(schedulerDriver: SchedulerDriver, offerID: Protos.OfferID): Unit = {}

  override def disconnected(schedulerDriver: SchedulerDriver): Unit = {}

  override def reregistered(schedulerDriver: SchedulerDriver, masterInfo: Protos.MasterInfo): Unit = {}

  override def slaveLost(schedulerDriver: SchedulerDriver, slaveID: Protos.SlaveID): Unit = {}

  override def error(schedulerDriver: SchedulerDriver, s: String): Unit = {}

  override def statusUpdate(schedulerDriver: SchedulerDriver, taskStatus: Protos.TaskStatus): Unit =
    println(s"received status update $taskStatus")

  override def frameworkMessage(schedulerDriver: SchedulerDriver,
                                executorID: Protos.ExecutorID,
                                slaveID: Protos.SlaveID,
                                bytes: Array[Byte]): Unit = {}

  override def resourceOffers(schedulerDriver: SchedulerDriver, list: util.List[Protos.Offer]): Unit = {
    // https://github.com/apache/mesos/blob/1.2.0/include/mesos/mesos.proto
    val receivedOffers = list.asScala
    // Select a valid offer for the task to run
    val optionalAcceptOffer = receivedOffers.find(checkOffer(task, _))
    // If there is a valid offer, get the remaining offers
    val remainingOffers = receivedOffers diff optionalAcceptOffer.toList
    // Launche the task by accepting the offer
    optionalAcceptOffer.foreach(offer =>
      schedulerDriver.launchTasks(List(offer.getId).asJava,
        List(taskToProto(task).setSlaveId(offer.getSlaveId).build()).asJava))
    // Decline the other offers
    remainingOffers foreach(offer => schedulerDriver.declineOffer(offer.getId))
  }

  override def registered(schedulerDriver: SchedulerDriver,
                          frameworkID: Protos.FrameworkID,
                          masterInfo: Protos.MasterInfo): Unit =
    println(s"Framework registered with ID: $frameworkID")

  override def executorLost(schedulerDriver: SchedulerDriver,
                            executorID: Protos.ExecutorID,
                            slaveID: Protos.SlaveID,
                            i: Int): Unit = {}
} 

//设置package包名称以及导入依赖的类
package com.github.roberveral.mesosgocd.scheduler

import com.github.roberveral.mesosgocd.spec.{Command, Resources, Task}
import mesosphere.mesos.protos.{FrameworkID, Offer, OfferID, SlaveID}
import org.scalatest.{FlatSpec, Matchers}
import mesosphere.mesos.protos.Implicits._
import com.github.roberveral.mesosgocd.spec.Implicits._
import org.apache.mesos.Protos

import scala.collection.JavaConverters._



class GoCDSchedulerSpec extends FlatSpec with Matchers {
  val offers: List[Protos.Offer] =
    List(Offer(OfferID("1"), FrameworkID("1"), SlaveID("1"), "test-hostname", Resources(1.0, 512.0), Seq(), Seq()),
      Offer(OfferID("2"), FrameworkID("1"), SlaveID("2"), "test-hostname2", Resources(4.0, 1024.0), Seq(), Seq()))

  "The scheudler" should "launch the configured task when it receives an offer with enough resources" in {
    val task = Task("test-task", "test-image", Command("ls"), Resources(1.0, 512.0))

    val scheduler = new GoCDScheduler(task)
    val driver = new TestSchedulerDriver
    scheduler.resourceOffers(driver, offers.asJava)

    driver.acceptedOfferIds.size shouldBe 1
    driver.acceptedOfferIds.head.getValue shouldBe "1"
    driver.launchedTasks.size shouldBe 1
    driver.launchedTasks.head.getSlaveId.getValue shouldBe "1"
    driver.launchedTasks.head.getName shouldBe "test-task"
    driver.declinedOfferIds.size shouldBe 1
  }

  it should "launch the configured task when it receives an offer with enough resources (not he first one)" in {
    val task = Task("test-task", "test-image", Command("ls"), Resources(3.0, 512.0))

    val scheduler = new GoCDScheduler(task)
    val driver = new TestSchedulerDriver
    scheduler.resourceOffers(driver, offers.asJava)

    driver.acceptedOfferIds.size shouldBe 1
    driver.acceptedOfferIds.head.getValue shouldBe "2"
    driver.launchedTasks.size shouldBe 1
    driver.launchedTasks.head.getSlaveId.getValue shouldBe "2"
    driver.launchedTasks.head.getName shouldBe "test-task"
    driver.declinedOfferIds.size shouldBe 1
  }

  it should "decline all offers when no one fits in the task resources" in {
    val task = Task("test-task", "test-image", Command("ls"), Resources(8.0, 10000.0))

    val scheduler = new GoCDScheduler(task)
    val driver = new TestSchedulerDriver
    scheduler.resourceOffers(driver, offers.asJava)

    driver.acceptedOfferIds.size shouldBe 0
    driver.launchedTasks.size shouldBe 0
    driver.declinedOfferIds.size shouldBe 2
  }

} 

//设置package包名称以及导入依赖的类
package com.vivint.ceph
package model

import org.apache.mesos.Protos

case class TaskStatus(taskId: String, slaveId: String, state: TaskState.EnumVal) {
  def toMesos: Protos.TaskStatus = {
    ProtoHelpers.newTaskStatus(taskId, slaveId, Protos.TaskState.valueOf(state.id))
  }
}

object TaskStatus extends ((String, String, TaskState.EnumVal) => TaskStatus) {
  def fromMesos(p: Protos.TaskStatus) = {
    TaskStatus(
      p.getTaskId.getValue,
      p.getSlaveId.getValue,
      TaskState.valuesById(p.getState.getNumber))
  }
} 

//设置package包名称以及导入依赖的类
package com.vivint.ceph
package model

import org.apache.mesos.Protos
import scala.collection.breakOut

object TaskState extends lib.Enum {
  sealed trait EnumVal extends Value {
    val id: Int
  }

  sealed trait Active extends EnumVal
  sealed trait Limbo extends EnumVal
  sealed trait Terminal extends EnumVal

  case object TaskStarting extends Active   { val id = Protos.TaskState.TASK_STARTING_VALUE ; val name = "TASK_STARTING" }
  case object TaskStaging  extends Active   { val id = Protos.TaskState.TASK_STAGING_VALUE  ; val name = "TASK_STAGING" }
  case object TaskRunning  extends Active   { val id = Protos.TaskState.TASK_RUNNING_VALUE  ; val name = "TASK_RUNNING" }
  case object TaskKilling  extends Active   { val id = Protos.TaskState.TASK_KILLING_VALUE  ; val name = "TASK_KILLING" }
  case object TaskFinished extends Terminal { val id = Protos.TaskState.TASK_FINISHED_VALUE ; val name = "TASK_FINISHED" }
  case object TaskFailed   extends Terminal { val id = Protos.TaskState.TASK_FAILED_VALUE   ; val name = "TASK_FAILED" }
  case object TaskKilled   extends Terminal { val id = Protos.TaskState.TASK_KILLED_VALUE   ; val name = "TASK_KILLED" }
  case object TaskError    extends Terminal { val id = Protos.TaskState.TASK_ERROR_VALUE    ; val name = "TASK_ERROR" }
  case object TaskLost     extends Limbo    { val id = Protos.TaskState.TASK_LOST_VALUE     ; val name = "TASK_LOST" }

  val values = Vector(TaskStarting, TaskStaging, TaskRunning, TaskKilling, TaskFinished, TaskFailed, TaskKilled,
    TaskError, TaskLost)

  val valuesById: Map[Int, EnumVal] =
    values.map { v => v.id -> v}(breakOut)

  def fromMesos(p: Protos.TaskState): TaskState.EnumVal = {
    TaskState.valuesById(p.getNumber)
  }

} 

//设置package包名称以及导入依赖的类
package com.vivint.ceph.lib

import mesosphere.mesos.matcher.PortsMatchResult
import mesosphere.mesos.matcher.PortsMatchResult.PortWithRole
import mesosphere.mesos.matcher.ResourceMatcher.ResourceSelector
import mesosphere.mesos.matcher.{ MatchResult, ResourceMatcher }
import mesosphere.mesos.protos.Resource
import org.apache.mesos.Protos
import scala.annotation.tailrec
import com.vivint.ceph.ProtoHelpers


class ContiguousPortMatcher(ports: Int, resourceSelector: ResourceSelector) extends ResourceMatcher {
  import ProtoHelpers._
  val resourceName = Resource.PORTS
  def apply(offerId: String, resources: Iterable[Protos.Resource]): Iterable[MatchResult] =
    doMatch(resources.toList)

  @tailrec private def doMatch(resources: List[Protos.Resource]): Iterable[MatchResult] = resources match {
    case Nil =>
      List(
        PortsMatchResult(
          false,
          Nil,
          Nil))
    case resource :: rest =>
      if (!resourceSelector(resource))
        doMatch(rest)
      else
        resource.ranges.find(_.length >= ports) match {
          case Some(range) =>
            val reserveRange = range.min to (range.min + ports - 1)
            // ReservationInfo
            List(
              PortsMatchResult(
                true,
                reserveRange.map { n => Some(PortWithRole("main", n.toInt, resource.reservation)) },
                List(
                  resource.toBuilder.
                    setRanges(
                      newRanges(List(reserveRange))).
                    build)))
          case None =>
            doMatch(rest)
        }
  }
} 

//设置package包名称以及导入依赖的类
package com.vivint.ceph

import com.vivint.ceph.kvstore.KVStore
import scala.concurrent.{ ExecutionContext, Future, Promise }
import org.apache.mesos.Protos
import scaldi.Injector
import scaldi.Injectable._

case class FrameworkIdStore(implicit injector: Injector) {
  private val kvStore = inject[KVStore]
  private val ksPath = "frameworkId"

  private val getP = Promise[Protos.FrameworkID]
  val get = getP.future

  import ExecutionContext.Implicits.global

  val initial = kvStore.get(ksPath).map { _.map { bytes =>
    Protos.FrameworkID.newBuilder().setValue(new String(bytes, "UTF-8")).build
  }}

  initial.foreach {
    case Some(fId) => getP.trySuccess(fId)
    case None => ()
  }


  
  def set(value: Protos.FrameworkID): Future[Unit] = {
    val bytes = value.getValue.getBytes("UTF-8")
    kvStore.createOrSet(ksPath, bytes).andThen { case _ =>
      getP.trySuccess(value)
    }
  }
}