Added Elasticsearch REST API and JMX metrics(for now, just BrokerTopicMetrics)

src/main/java/com/chickling/kmonitor/alert/DaemonThreadFactory.java

@@ -0,0 +1,25 @@
+package com.chickling.kmonitor.alert;
+
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * @author Hulva Luva.H
+ *
+ */
+public class DaemonThreadFactory implements ThreadFactory {
+	private String prefix = "";
+
+	public DaemonThreadFactory(String prefix) {
+		this.prefix = prefix;
+	}
+
+	@Override
+	public Thread newThread(Runnable r) {
+		Thread t = new Thread(r, "DaemonThread-" + prefix);
+		t.setDaemon(true);
+		if (t.getPriority() != Thread.NORM_PRIORITY)
+			t.setPriority(Thread.NORM_PRIORITY);
+		return t;
+	}
+
+}

src/main/java/com/chickling/kmonitor/alert/KafkaNodeListener.java

@@ -0,0 +1,49 @@
+package com.chickling.kmonitor.alert;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.I0Itec.zkclient.IZkChildListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.chickling.kmonitor.utils.ZKUtils;
+
+import kafka.utils.ZkUtils;
+
+/**
+ * @author Hulva Luva.H
+ * 
+ *         Listening Kafka node on ZK
+ *
+ */
+public class KafkaNodeListener {
+	private static Logger LOG = LoggerFactory.getLogger(KafkaNodeListener.class);
+	private static ExecutorService exec = null;
+
+
+	public KafkaNodeListener() {
+		exec = Executors.newCachedThreadPool(new DaemonThreadFactory("KafkaNodeOffLineListener"));
+	}
+
+	public void startListener() {
+		LOG.info("Starting Kafka ZK node listener...");
+		exec.execute(new Runnable() {
+
+			@Override
+			public void run() {
+				ZKUtils.getZKClient().subscribeChildChanges(ZkUtils.BrokerIdsPath(), new IZkChildListener() {
+
+					@Override
+					public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
+
+					}
+
+				});
+			}
+
+		});
+	}
+
+}

src/main/java/com/chickling/kmonitor/alert/TaskManager.java

@@ -53,24 +53,7 @@ public static boolean exits(TaskContent taskContent) {
 		return false;
 	}
 
-	public static void refreshTask(TaskContent taskContent) {
-		String taskFilePath = taskFolder + "/" + taskContent.getGroup() + "-" + taskContent.getTopic() + ".task";
-		try {
-			PrintWriter writer = new PrintWriter(taskFilePath);
-			writer.println(new Gson().toJson(taskContent));
-			writer.close();
-			Map<String, TaskContent> task = new HashMap<String, TaskContent>();
-			task.put(taskContent.getTopic(), taskContent);
-			tasks.put(taskContent.getGroup(), task);
-		} catch (Exception e) {
-			logger.error("refreshTask to file failed!", e);
-		}
-	}
-
 	public static void addTask(TaskContent taskContent) throws JSONException {
-		if (exits(taskContent)) {
-			return;
-		}
 		Map<String, TaskContent> task = null;
 		if (tasks.containsKey(taskContent.getGroup())) {
 			task = tasks.get(taskContent.getGroup());

src/main/java/com/chickling/kmonitor/config/AppConfig.java

@@ -5,8 +5,10 @@
  *
  */
 public class AppConfig {
+	private String apiType;
 	private String esHosts;
 	private String esIndex;
+	private String docTypeForOffset;
 
 	private Integer dataCollectFrequency = 1;
 
@@ -29,6 +31,14 @@ public class AppConfig {
 
 	private Long excludeByLastSeen = 2592000L;
 
+	public String getApiType() {
+		return apiType;
+	}
+
+	public void setApiType(String apiType) {
+		this.apiType = apiType;
+	}
+
 	public String getZkHosts() {
 		return zkHosts;
 	}
@@ -61,6 +71,14 @@ public void setEsIndex(String esIndex) {
 		this.esIndex = esIndex;
 	}
 
+	public String getDocTypeForOffset() {
+		return docTypeForOffset;
+	}
+
+	public void setDocTypeForOffset(String docTypeForOffset) {
+		this.docTypeForOffset = docTypeForOffset;
+	}
+
 	public Integer getDataCollectFrequency() {
 		return dataCollectFrequency;
 	}

src/main/java/com/chickling/kmonitor/controller/AlertController.java

@@ -1,5 +1,6 @@
 package com.chickling.kmonitor.controller;
 
+import java.util.HashSet;
 import java.util.Set;
 
 import org.json.JSONObject;
@@ -26,7 +27,10 @@ public class AlertController {
 
 	@RequestMapping(value = "/tasks", method = RequestMethod.GET)
 	public Set<TaskContent> get() {
-		return TaskManager.getTasks();
+		if(SystemManager.getConfig().getIsAlertEnabled()) {
+			return TaskManager.getTasks();
+		}
+		return new HashSet<TaskContent>();
 	}
 
 	@RequestMapping(value = "/isAlertEnabled", method = RequestMethod.GET)
@@ -36,15 +40,9 @@ public String isAlertEnabled() {
 		return response.toString();
 	}
 
-	@RequestMapping(value = "/new", method = RequestMethod.POST)
-	public Set<TaskContent> post(@RequestBody TaskContent taskContent) {
-		TaskManager.saveTaskToFileAndAddToTasks(taskContent);
-		return TaskManager.getTasks();
-	}
-
-	@RequestMapping(value = { "/update" }, method = RequestMethod.POST)
+	@RequestMapping(value = { "/task" }, method = RequestMethod.POST)
 	public Set<TaskContent> put(@RequestBody TaskContent taskContent) {
-		TaskManager.refreshTask(taskContent);
+		TaskManager.saveTaskToFileAndAddToTasks(taskContent);
 		return TaskManager.getTasks();
 	}