Removed add clusters and modify cluster

.gitignore

@@ -12,3 +12,13 @@ dist
 activator*
 RUNNING_PID
 .DS_Store
+/bin
+.cache
+.classpath
+.project
+.sbtserver/
+.tmpBin
+project/.sbtserver
+project/.sbtserver.lock
+project/play-fork-run.sbt
+project/sbt-ui.sbt

app/GlobalKafkaManager.scala

@@ -3,27 +3,64 @@
  * See accompanying LICENSE file.
  */
 
+import collection.JavaConversions._
 import controllers.KafkaManagerContext
 import kafka.manager.KafkaManager
 import play.api._
+import com.typesafe.config.ConfigException
+import models.navigation.BreadCrumbs
 
 /**
  * @author hiral
  */
-object GlobalKafkaManager extends GlobalSettings {
+object Global extends GlobalSettings {
 
   private[this] var kafkaManager: KafkaManager = null
+  val ClusterZks = "managed-kafka-clusters"
 
   override def beforeStart(app: Application): Unit = {
     Logger.info("Init kafka manager...")
-    KafkaManagerContext.getKafkaManger
+    kafkaManager = KafkaManagerContext.getKafkaManger
+    initialize_with_clusters_from_conf()
     Thread.sleep(5000)
   }
 
   override def onStop(app: Application) {
     KafkaManagerContext.shutdown()
     Logger.info("Application shutdown...")
   }
+
+
+  def initialize_with_clusters_from_conf() {
+    Logger.info("initialize_with_clusters_from_conf(): Going to read kafka list from config and initialize")
+
+    val configWithDefaults = kafkaManager.getConfigWithDefaults
+
+    var member_zkhosts_list =  List[String]()
+    try {
+      member_zkhosts_list = configWithDefaults.getStringList(ClusterZks).toList
+    } catch {
+      case cfge: ConfigException => Logger.error("managed-kafka-clusters param not defined in the conf or is in wrong format")
+    }
+
+
+    member_zkhosts_list.foreach{ item => 
+
+      // Each item in the list from config will be a string of format
+      // "clustername, version, zk1:port1, zk2:port2...."
+      //
+      val mylist = item.split(",")
+      val name = mylist(0)
+      val version = mylist(1)
+      val zkhosts = mylist.takeRight(mylist.size - 2).mkString(",")
+
+
+      kafkaManager.addCluster(name, version, zkhosts)
+
+    }
+  }
+
+
 }
 
 

app/kafka/manager/KafkaManager.scala

@@ -146,6 +146,8 @@ class KafkaManager(akkaConfig: Config) {
     }
   }
 
+  def getConfigWithDefaults = configWithDefaults
+
   def addCluster(clusterName: String, version: String, zkHosts: String) : Future[ApiError \/ Unit] = {
     val cc = ClusterConfig(clusterName, version, zkHosts)
     tryWithKafkaManagerActor(KMAddCluster(cc)) { result: KMCommandResult =>

app/kafka/manager/KafkaManagerActor.scala

@@ -38,7 +38,8 @@ object KafkaVersion {
   val supportedVersions: Map[String,KafkaVersion] = Map(
     "0.8.1.1" -> Kafka_0_8_1_1, 
     "0.8.2-beta" -> Kafka_0_8_2_0,
-    "0.8.2.0" -> Kafka_0_8_2_0)
+    "0.8.2.0" -> Kafka_0_8_2_0,
+    "0.8.2.1" -> Kafka_0_8_2_0)
 
   val formSelectList : IndexedSeq[(String,String)] = supportedVersions.toIndexedSeq.filterNot(_._1.contains("beta")).map(t => (t._1,t._2.toString))
 
@@ -82,8 +83,8 @@ object ClusterConfig {
     validateName(name)
     //validate zk hosts
     validateZkHosts(zkHosts)
-    val cleanZkHosts = zkHosts.replaceAll(" ","").toLowerCase
-    new ClusterConfig(name.toLowerCase, CuratorConfig(cleanZkHosts, zkMaxRetry), true, kafkaVersion)
+    val cleanZkHosts = zkHosts.replaceAll(" ", "")
+    new ClusterConfig(name, CuratorConfig(cleanZkHosts, zkMaxRetry), true, kafkaVersion)
   }
 
   def customUnapply(cc: ClusterConfig) : Option[(String, String, String, Int)] = {

app/models/navigation/BreadCrumbs.scala

@@ -27,8 +27,7 @@ object BreadCrumbs {
   import models.navigation.QuickRoutes._
 
   val baseBreadCrumbs: Map[String, IndexedSeq[BreadCrumb]] = Map(
-    "Clusters" -> IndexedSeq.empty[BreadCrumb],
-    "Add Cluster" -> IndexedSeq("Clusters".baseRouteBreadCrumb)
+    "Clusters" -> IndexedSeq.empty[BreadCrumb]
   )
 
   val clusterBreadCrumbs: Map[String, IndexedSeq[BreadCrumb]] = Map(

app/models/navigation/Menus.scala

@@ -13,8 +13,7 @@ object Menus {
   def clusterMenus(cluster: String) : IndexedSeq[Menu] = IndexedSeq(
     Menu("Cluster",IndexedSeq(
       "Summary".clusterRouteMenuItem(cluster),
-      "List".baseRouteMenuItem,
-      "Add Cluster".baseRouteMenuItem),
+      "List".baseRouteMenuItem),
       None),
     "Brokers".clusterMenu(cluster),
     Menu("Topic",IndexedSeq(
@@ -27,8 +26,7 @@ object Menus {
 
   def indexMenu : IndexedSeq[Menu] = IndexedSeq(
     Menu("Cluster",IndexedSeq(
-      "List".baseRouteMenuItem,
-      "Add Cluster".baseRouteMenuItem),
+      "List".baseRouteMenuItem),
       None)
   )
 }

app/models/navigation/QuickRoutes.scala

@@ -15,8 +15,7 @@ object QuickRoutes {
 
   val baseRoutes : Map[String, Call] = Map(
     "Clusters" -> controllers.routes.Application.index(),
-    "List" -> controllers.routes.Application.index(),
-    "Add Cluster" -> controllers.routes.Cluster.addCluster()
+    "List" -> controllers.routes.Application.index()
   )
   val clusterRoutes : Map[String, String => Call] = Map(
     "Update Cluster" -> controllers.routes.Cluster.updateCluster,

app/views/cluster/clusterList.scala.html

@@ -23,7 +23,6 @@
                     <td>
                         <div class="btn-group-horizontal" role="group" aria-label="...">
                         @if(cluster.enabled) {
-                            <a href="@routes.Cluster.updateCluster(cluster.name)" class="btn btn-default ops-button" role="button">Modify</a>
                             @b3.form(routes.Cluster.handleUpdateCluster(cluster.name)) {
                             <input type="hidden" name="name" value="@cluster.name">
                             <input type="hidden" name="kafkaVersion" value="@cluster.version.toString">
@@ -39,13 +38,6 @@
                             <input type="hidden" name="operation" value="Enable">
                             @b3.submit('class -> "btn btn-success ops-button"){ Enable }
                             }
-                            @b3.form(routes.Cluster.handleUpdateCluster(cluster.name)) {
-                            <input type="hidden" name="name" value="@cluster.name">
-                            <input type="hidden" name="kafkaVersion" value="@cluster.version.toString">
-                            <input type="hidden" name="zkHosts" value="@cluster.curatorConfig.zkConnect">
-                            <input type="hidden" name="operation" value="Delete">
-                            @b3.submit('class -> "btn btn-danger ops-button"){ Delete }
-                            }
                         }
                         </div>
                     </td>

build.sbt

@@ -5,7 +5,7 @@
 name := """kafka-manager"""
 
 /* For packaging purposes, -SNAPSHOT MUST contain a digit */
-version := "1.1"
+version := "1.2"
 
 scalaVersion := "2.11.5"
 
@@ -82,3 +82,6 @@ rpmLicense := Some("Apache")
 
 
 
+
+
+fork in run := true

conf/application.conf

@@ -65,3 +65,9 @@ kafka-manager.zkhosts=${?ZK_HOSTS}
 pinned-dispatcher.type="PinnedDispatcher"
 pinned-dispatcher.executor="thread-pool-executor"
 
+# These will be added into Kafka Manager at startup 
+# Value of this param is a list of strings representing info about kafka clusters to be managed
+#    ["clusterNmae, Version, zk1:port1, zk2:port2,...", ...]
+# Supported versions currently are "0.8.1.1"  and "0.8.2-beta"  only
+# (line 38:  kafka/manager/KafkaManagerActor.scala)
+managed-kafka-clusters=["kmgr1,0.8.2-beta,zk1:2181,zk2:2181"]