Support for Mesos in Kafka Manager

Dockerfile

@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM java:7
+
+ENV app /kafka-manager
+WORKDIR ${app}
+
+COPY sbt $app/sbt
+COPY build.sbt $app/build.sbt
+COPY app $app/app
+COPY conf $app/conf
+COPY img $app/img
+COPY project/build.properties $app/project/build.properties
+COPY project/plugins.sbt $app/project/plugins.sbt
+COPY public $app/public
+COPY src $app/src
+COPY test $app/test
+
+RUN ./sbt assembly
+EXPOSE 9000
+CMD [ "./sbt", "run" ]

app/controllers/Application.scala

@@ -21,8 +21,8 @@ object Application extends Controller {
   private[this] implicit val af: ApplicationFeatures = ApplicationFeatures.features
 
   def index = Action.async {
-    kafkaManager.getClusterList.map { errorOrClusterList =>
-      Ok(views.html.index(errorOrClusterList))
-    }
+    for {errorOrSchedulerList <- kafkaManager.getSchedulerList
+         errorOrClusterList <- kafkaManager.getClusterList
+    } yield Ok(views.html.index(errorOrClusterList, errorOrSchedulerList))
   }
 }

app/controllers/Cluster.scala

@@ -6,7 +6,7 @@
 package controllers
 
 import features.{KMClusterManagerFeature, ApplicationFeatures}
-import kafka.manager.{KafkaVersion, ApiError, ClusterConfig}
+import kafka.manager.{SchedulerConfig, KafkaVersion, ApiError, ClusterConfig}
 import models.FollowLink
 import models.form._
 import play.api.data.Form
@@ -76,6 +76,17 @@ object Cluster extends Controller {
     )(ClusterConfig.apply)(ClusterConfig.customUnapply)
   )
 
+  val schedulerConfigForm = Form(
+    mapping(
+      "name" -> nonEmptyText.verifying(maxLength(250), validateName),
+      "kafkaVersion" -> nonEmptyText.verifying(validateKafkaVersion),
+      "apiUrl" -> nonEmptyText.verifying(validateZkHosts),
+      "zkHosts" -> nonEmptyText.verifying(validateZkHosts),
+      "zkMaxRetry" -> ignored(100 : Int),
+      "jmxEnabled" -> boolean
+    )(SchedulerConfig.apply)(SchedulerConfig.customUnapply)
+  )
+
   val updateForm = Form(
     mapping(
       "operation" -> nonEmptyText.verifying(validateOperation),
@@ -114,6 +125,28 @@ object Cluster extends Controller {
     }
   }
 
+  def addScheduler = Action.async { implicit request =>
+    Future.successful(Ok(scheduler.views.html.scheduler.addScheduler(schedulerConfigForm)))
+  }
+
+  def handleAddScheduler = Action.async { implicit request =>
+    schedulerConfigForm.bindFromRequest.fold(
+      formWithErrors => Future.successful(BadRequest(scheduler.views.html.scheduler.addScheduler(formWithErrors))),
+      schedulerConfig => {
+        kafkaManager.addScheduler(schedulerConfig.name, schedulerConfig.version.toString, schedulerConfig.apiUrl, schedulerConfig.curatorConfig.zkConnect, jmxEnabled = true).map { errorOrSuccess =>
+          Ok(views.html.common.resultOfCommand(
+            views.html.navigation.defaultMenu(),
+            models.navigation.BreadCrumbs.withView("Add Scheduler"),
+            errorOrSuccess,
+            "Add Scheduler",
+            FollowLink("Go to scheduler view.",scheduler.controllers.routes.SchedulerApplication.getScheduler(schedulerConfig.name).toString()),
+            FollowLink("Try again.",routes.Cluster.addScheduler().toString())
+          ))
+        }
+      }
+    )
+  }
+
   def updateCluster(c: String) = Action.async { implicit request =>
     featureGate(KMClusterManagerFeature) {
       kafkaManager.getClusterConfig(c).map { errorOrClusterConfig =>

app/kafka/manager/ActorModel.scala

@@ -5,11 +5,12 @@
 
 package kafka.manager
 
-import java.util.Properties
+import java.util.{Date, Properties}
 
 import org.joda.time.DateTime
 import kafka.common.TopicAndPartition
 import org.slf4j.LoggerFactory
+import scheduler.models.form.Failover
 
 import scala.collection.immutable.Queue
 import scala.concurrent.{Await, Future}
@@ -639,4 +640,100 @@ object ActorModel {
       LogkafkaIdentity(hostname, lct.isDefined, identitySet.toMap)
     }
   }
+
+  case class SMGetBrokerIdentity(id: Int) extends BVRequest
+
+  case object SMGetView extends QueryRequest
+  case class SMView(topicsCount: Int, brokersCount: Int, schedulerConfig: SchedulerConfig) extends QueryResponse
+
+  case class SMAddBroker(id: Int, cpus: Option[Double], mem: Option[Long], heap: Option[Long], port: Option[String],
+                         bindAddress: Option[String], constraints: Option[String], options: Option[String],
+                         log4jOptions: Option[String], jvmOptions: Option[String], stickinessPeriod: Option[String],
+                         failover: Failover) extends CommandRequest
+
+  case class SMUpdateBroker(id: Int, cpus: Option[Double], mem: Option[Long], heap: Option[Long], port: Option[String],
+                            bindAddress: Option[String], constraints: Option[String], options: Option[String],
+                            log4jOptions: Option[String], jvmOptions: Option[String], stickinessPeriod: Option[String],
+                            failover: Failover) extends CommandRequest
+
+  case class SMCommandResult(result: Try[Unit]) extends CommandResponse
+
+  case class KSCAddBroker(id: Int, cpus: Option[Double], mem: Option[Long], heap: Option[Long], port: Option[String],
+                          bindAddress: Option[String], constraints: Option[String], options: Option[String],
+                          log4jOptions: Option[String], jvmOptions: Option[String], stickinessPeriod: Option[String],
+                          failover: Failover) extends CommandRequest
+
+  case class KSCUpdateBroker(id: Int, cpus: Option[Double], mem: Option[Long], heap: Option[Long], port: Option[String],
+                             bindAddress: Option[String], constraints: Option[String], options: Option[String],
+                             log4jOptions: Option[String], jvmOptions: Option[String], stickinessPeriod: Option[String],
+                             failover: Failover) extends CommandRequest
+
+  case class SMStartBroker(brokerId: Int) extends CommandRequest
+  case class SMStopBroker(brokerId: Int) extends CommandRequest
+  case class SMRemoveBroker(brokerId: Int) extends CommandRequest
+  case class SMRebalanceTopics(ids: String, topics:Option[String]) extends CommandRequest
+
+  case class KSCStartBroker(id: Int) extends CommandRequest
+  case class KSCStopBroker(id: Int) extends CommandRequest
+  case class KSCRemoveBroker(id: Int) extends CommandRequest
+  case class KSCRebalanceTopics(ids: String, topics:Option[String]) extends CommandRequest
+
+  case class KMSchedulerCommandRequest(scheduler: String, request: CommandRequest) extends CommandRequest
+  case class KMSchedulerList(active: IndexedSeq[SchedulerConfig], pending : IndexedSeq[SchedulerConfig]) extends QueryResponse
+  case object KMGetAllSchedulers extends QueryRequest
+  case class KMAddScheduler(config: SchedulerConfig) extends CommandRequest
+  case class KMSchedulerQueryRequest(schedulerName: String, request: QueryRequest) extends QueryRequest
+  case class KMGetSchedulerConfig(schedulerName: String) extends QueryRequest
+  case class KMSchedulerConfigResult(result: Try[SchedulerConfig]) extends QueryResponse
+
+  case object SchedulerKSGetBrokers extends KSRequest
+  case class SchedulerBrokerList(list: Seq[SchedulerBrokerIdentity], schedulerConfig: SchedulerConfig) extends QueryResponse
+
+  case class SchedulerBrokerTaskIdentity(id: String,
+                                         slaveId: String,
+                                         executorId: String,
+                                         hostname: String,
+                                         endpoint: Option[String],
+                                         state: String)
+
+  case class SchedulerBrokerStickinessIdentity(period: String,
+                                               stopTime: Option[Date],
+                                               hostname: Option[String])
+
+  case class SchedulerBrokerFailoverIdentity(delay: String,
+                                             maxDelay: String,
+                                             maxTries: Option[Int],
+                                             failures: Option[Int],
+                                             failureTime: Option[Date])
+
+  case class SchedulerBrokerIdentity(id: Int, active: Boolean, cpus: Double, mem: Long, heap: Long, port: Option[String],
+                                     bindAddress: Option[String], constraints: Seq[(String, String)], options: Seq[(String, String)],
+                                     log4jOptions: Seq[(String, String)], jvmOptions: Option[String],
+                                     stickiness: SchedulerBrokerStickinessIdentity,
+                                     failover: SchedulerBrokerFailoverIdentity,
+                                     task: Option[SchedulerBrokerTaskIdentity],
+                                     schedulerConfig: SchedulerConfig = null,
+                                     metrics: Option[BrokerMetrics] = None,
+                                     stats: Option[BrokerClusterStats] = None) {
+
+    def actualHost(): Option[String] = task.flatMap(t => t.endpoint.map(_.split(":")(0)))
+
+    def actualPort(): Option[String] = task.flatMap(t => t.endpoint.map(_.split(":")(1)))
+
+    def numTopics() = 0
+    def numPartitions() = 0
+
+    def topicPartitions() = Seq.empty[(TopicIdentity,  IndexedSeq[Int])]
+
+    def constraintsDesc = constraints.map { case (k, v) => s"$k=$v" }.mkString
+    def optionsDesc = options.map { case (k, v) => s"$k=$v" }.mkString
+    def log4jOptionsDesc = log4jOptions.map { case (k, v) => s"$k=$v" }.mkString
+
+    def state(): String = {
+      if (active)
+        if (actualHost().isEmpty) "starting" else "running"
+      else
+        "stopped|failed"
+    }
+  }
 }

app/kafka/manager/KafkaManager.scala

@@ -13,6 +13,7 @@ import akka.util.Timeout
 import com.typesafe.config.{ConfigFactory, Config}
 import kafka.manager.ActorModel._
 import org.slf4j.{LoggerFactory, Logger}
+import scheduler.models.form.Failover
 
 import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.duration._
@@ -33,6 +34,7 @@ case class BrokerListExtended(list: IndexedSeq[BrokerIdentity],
                               clusterContext: ClusterContext)
 case class ConsumerListExtended(list: IndexedSeq[(String, Option[ConsumerIdentity])], clusterContext: ClusterContext)
 case class LogkafkaListExtended(list: IndexedSeq[(String, Option[LogkafkaIdentity])], deleteSet: Set[String])
+case class SchedulerBrokerListExtended(list: Seq[SchedulerBrokerIdentity], metrics: Map[Int,BrokerMetrics], combinedMetric: Option[BrokerMetrics], schedulerConfig: SchedulerConfig)
 
 case class ApiError(msg: String)
 object ApiError {
@@ -778,4 +780,129 @@ class KafkaManager(akkaConfig: Config)
       )
     }
   }
+
+  def addScheduler(schedulerName: String, version: String, apiUrl: String, zkHosts: String, jmxEnabled: Boolean): Future[ApiError \/
+    Unit] =
+  {
+    val sc = SchedulerConfig(schedulerName, apiUrl, CuratorConfig(zkHosts), enabled = true, KafkaVersion(version), jmxEnabled = jmxEnabled)
+
+    tryWithKafkaManagerActor(KMAddScheduler(sc)) { result: KMCommandResult =>
+      result.result.get
+    }
+  }
+
+  def getSchedulerView(schedulerName: String): Future[ApiError \/ SMView] = {
+    tryWithKafkaManagerActor(KMSchedulerQueryRequest(schedulerName, SMGetView))(identity[SMView])
+  }
+
+  def getSchedulerBrokerList(schedulerName: String): Future[ApiError \/ SchedulerBrokerListExtended] = {
+    implicit val ec = apiExecutionContext
+
+    val futureBrokerList = tryWithKafkaManagerActor(KMSchedulerQueryRequest(schedulerName, SchedulerKSGetBrokers))(identity[SchedulerBrokerList])
+    futureBrokerList.map {
+      case \/-(SchedulerBrokerList(identities, config)) =>
+        \/-(SchedulerBrokerListExtended(identities, Map.empty, None, config))
+      case a : -\/[ApiError] =>
+        a
+    }
+  }
+
+  def getSchedulerConfig(schedulerName: String): Future[ApiError \/ SchedulerConfig] = {
+    tryWithKafkaManagerActor(KMGetSchedulerConfig(schedulerName)) { result: KMSchedulerConfigResult =>
+      result.result.get
+    }
+  }
+
+  def getSchedulerList: Future[ApiError \/ KMSchedulerList] = {
+    tryWithKafkaManagerActor(KMGetAllSchedulers)(identity[KMSchedulerList])
+  }
+
+
+  def addBroker(schedulerName:String, id:Int, cpus: Option[Double], mem: Option[Long], heap: Option[Long], port: Option[String],
+                bindAddress: Option[String], constraints: Option[String], options: Option[String],
+                log4jOptions: Option[String], jvmOptions: Option[String], stickinessPeriod: Option[String],
+                failover: Failover): Future[ApiError \/ Unit] =
+  {
+    implicit val ec = apiExecutionContext
+    withKafkaManagerActor(KMSchedulerCommandRequest(schedulerName, SMAddBroker(id, cpus, mem, heap, port, bindAddress,
+      constraints, options, log4jOptions, jvmOptions, stickinessPeriod, failover))) {
+      result: Future[SMCommandResult] =>
+        result.map(cmr => toDisjunction(cmr.result))
+    }
+
+  }
+
+  def updateBroker(schedulerName:String, id:Int, cpus: Option[Double], mem: Option[Long], heap: Option[Long], port: Option[String],
+                   bindAddress: Option[String], constraints: Option[String], options: Option[String],
+                   log4jOptions: Option[String], jvmOptions: Option[String], stickinessPeriod: Option[String],
+                   failover: Failover): Future[ApiError \/ Unit] =
+  {
+    implicit val ec = apiExecutionContext
+    withKafkaManagerActor(KMSchedulerCommandRequest(schedulerName, SMUpdateBroker(id, cpus, mem, heap, port, bindAddress,
+      constraints, options, log4jOptions, jvmOptions, stickinessPeriod, failover))) {
+      result: Future[SMCommandResult] =>
+        result.map(cmr => toDisjunction(cmr.result))
+    }
+
+  }
+
+  def getBrokerIdentity(schedulerName: String, brokerId: Int): Future[ApiError \/ SchedulerBrokerIdentity] = {
+    val futureView = tryWithKafkaManagerActor(
+      KMSchedulerQueryRequest(
+        schedulerName,
+        SMGetBrokerIdentity(brokerId)
+      )
+    )(identity[Option[SchedulerBrokerIdentity]])
+
+    implicit val ec = apiExecutionContext
+    futureView.flatMap[ApiError \/ SchedulerBrokerIdentity] { errOrView =>
+      errOrView.fold(
+      { err: ApiError =>
+        Future.successful(-\/[ApiError](err))
+      }, { viewOption: Option[SchedulerBrokerIdentity] =>
+        viewOption.fold {
+          Future.successful[ApiError \/ SchedulerBrokerIdentity](-\/(ApiError(s"Broker not found $brokerId for scheduler $schedulerName")))
+        } { view =>
+          Future.successful(\/-(view))
+        }
+      }
+      )
+    }
+  }
+
+  def startBroker(schedulerName: String, brokerId: Int): Future[ApiError \/ Unit] =
+  {
+    implicit val ec = apiExecutionContext
+    withKafkaManagerActor(KMSchedulerCommandRequest(schedulerName, SMStartBroker(brokerId))) {
+      result: Future[SMCommandResult] =>
+        result.map(cmr => toDisjunction(cmr.result))
+    }
+  }
+
+  def stopBroker(schedulerName: String, brokerId: Int): Future[ApiError \/ Unit] =
+  {
+    implicit val ec = apiExecutionContext
+    withKafkaManagerActor(KMSchedulerCommandRequest(schedulerName, SMStopBroker(brokerId))) {
+      result: Future[SMCommandResult] =>
+        result.map(cmr => toDisjunction(cmr.result))
+    }
+  }
+
+  def removeBroker(schedulerName: String, brokerId: Int): Future[ApiError \/ Unit] =
+  {
+    implicit val ec = apiExecutionContext
+    withKafkaManagerActor(KMSchedulerCommandRequest(schedulerName, SMRemoveBroker(brokerId))) {
+      result: Future[SMCommandResult] =>
+        result.map(cmr => toDisjunction(cmr.result))
+    }
+  }
+
+  def rebalanceTopics(schedulerName: String, ids: String, topics: Option[String]): Future[ApiError \/ Unit] =
+  {
+    implicit val ec = apiExecutionContext
+    withKafkaManagerActor(KMSchedulerCommandRequest(schedulerName, SMRebalanceTopics(ids, topics))) {
+      result: Future[SMCommandResult] =>
+        result.map(cmr => toDisjunction(cmr.result))
+    }
+  }
 }