Merge pull request #1 from smartloli/master

update

README.md

@@ -88,7 +88,7 @@ Since this is an open source tool, please comply with the relevant laws and regu
 
 [Kafka Eagle Architecture PDF](http://pdf.kafka-eagle.org/)
 
-# Committers
+# Contributors
 
 Thanks to the following members for maintaining the project.
 

kafka-eagle-api/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>org.smartloli.kafka.eagle</groupId>
 		<artifactId>kafka-eagle</artifactId>
-		<version>1.3.5</version>
+		<version>1.3.6</version>
 	</parent>
 	<artifactId>kafka-eagle-api</artifactId>
 	<dependencies>

kafka-eagle-common/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>org.smartloli.kafka.eagle</groupId>
 		<artifactId>kafka-eagle</artifactId>
-		<version>1.3.5</version>
+		<version>1.3.6</version>
 	</parent>
 	<artifactId>kafka-eagle-common</artifactId>
 

kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/constant/JmxConstants.java

@@ -38,7 +38,8 @@ public interface KafkaNetWork {
 	public interface KafkaServer8 {
 		public static final String version = "kafka.common:type=AppInfo,name=Version";
 		public static final String value = "Value";
-		public static final String logSize = "kafka.log:type=Log,name=LogEndOffset,topic=%s,partition=%s";
+		public static final String endLogSize = "kafka.log:type=Log,name=LogEndOffset,topic=%s,partition=%s";
+		public static final String startLogSize = "kafka.log:type=Log,name=LogStartOffset,topic=%s,partition=%s";
 	}
 
 	public interface KafkaServer {
@@ -78,6 +79,13 @@ class ZookeeperClientMetrics {
 			public static String zooKeeperRequestLatencyMs = "kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs";
 		}
 
+
+		class OS{
+			public static String type = "java.lang:type=OperatingSystem";
+			public static String totalPhysicalMemorySize = "TotalPhysicalMemorySize";
+			public static String freePhysicalMemorySize = "FreePhysicalMemorySize";
+		}
+
 		public static final String version = "kafka.server:type=app-info,id=%s";
 		public static final String value = "Version";
 	}

kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/protocol/BrokersInfo.java

@@ -33,6 +33,15 @@ public class BrokersInfo extends BaseProtocol {
 	private String modify = "";
 	private int jmxPort = 0;
 	private String version = "";
+	private String ids = "";
+
+	public String getIds() {
+		return ids;
+	}
+
+	public void setIds(String ids) {
+		this.ids = ids;
+	}
 
 	public String getVersion() {
 		return version;

kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/protocol/PartitionsInfo.java

@@ -35,15 +35,15 @@ public class PartitionsInfo extends BaseProtocol {
 	private int partitionNumbers = 0;
 	private String created = "";
 	private String modify = "";
-	private long logSize = 0L;
-
-	public long getLogSize() {
-		return logSize;
-	}
-
-	public void setLogSize(long logSize) {
-		this.logSize = logSize;
-	}
+//	private long logSize = 0L;
+//
+//	public long getLogSize() {
+//		return logSize;
+//	}
+//
+//	public void setLogSize(long logSize) {
+//		this.logSize = logSize;
+//	}
 
 	public String getCreated() {
 		return created;

kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/protocol/topic/TopicRank.java

@@ -0,0 +1,68 @@
+/**
+ * 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.
+ */
+package org.smartloli.kafka.eagle.common.protocol.topic;
+
+import org.smartloli.kafka.eagle.common.protocol.BaseProtocol;
+
+/**
+ * Rank topic records logsize & capacity size.
+ * 
+ * @author smartloli.
+ *
+ *         Created by Jul 27, 2019
+ */
+public class TopicRank extends BaseProtocol {
+
+	private String cluster = "";
+	private String topic = "";
+	private String tkey = "";
+	private long tvalue = 0L;
+
+	public String getCluster() {
+		return cluster;
+	}
+
+	public void setCluster(String cluster) {
+		this.cluster = cluster;
+	}
+
+	public String getTopic() {
+		return topic;
+	}
+
+	public void setTopic(String topic) {
+		this.topic = topic;
+	}
+
+	public String getTkey() {
+		return tkey;
+	}
+
+	public void setTkey(String tkey) {
+		this.tkey = tkey;
+	}
+
+	public long getTvalue() {
+		return tvalue;
+	}
+
+	public void setTvalue(long tvalue) {
+		this.tvalue = tvalue;
+	}
+
+}

kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/protocol/topic/TopicSqlHistory.java

@@ -0,0 +1,86 @@
+/**
+ * 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.
+ */
+package org.smartloli.kafka.eagle.common.protocol.topic;
+
+import org.smartloli.kafka.eagle.common.protocol.BaseProtocol;
+
+/**
+ * Record topic query sql.
+ * 
+ * @author smartloli.
+ *
+ *         Created by Jul 27, 2019
+ */
+public class TopicSqlHistory extends BaseProtocol {
+
+	private String cluster = "";
+	private String username = "";
+	private String ksql = "";
+	private String status = "";
+	private long timespan = 0L;
+	private String tm = "";
+
+	public String getCluster() {
+		return cluster;
+	}
+
+	public void setCluster(String cluster) {
+		this.cluster = cluster;
+	}
+
+	public String getUsername() {
+		return username;
+	}
+
+	public void setUsername(String username) {
+		this.username = username;
+	}
+
+	public String getKsql() {
+		return ksql;
+	}
+
+	public void setKsql(String ksql) {
+		this.ksql = ksql;
+	}
+
+	public String getStatus() {
+		return status;
+	}
+
+	public void setStatus(String status) {
+		this.status = status;
+	}
+
+	public long getTimespan() {
+		return timespan;
+	}
+
+	public void setTimespan(long timespan) {
+		this.timespan = timespan;
+	}
+
+	public String getTm() {
+		return tm;
+	}
+
+	public void setTm(String tm) {
+		this.tm = tm;
+	}
+
+}

kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/util/KConstants.java

@@ -108,6 +108,8 @@ public interface MBean {
 		public final static String TOTALPRODUCEREQUESTSPERSEC = "total_produce_requests";
 		public final static String REPLICATIONBYTESINPERSEC = "replication_bytes_out";
 		public final static String REPLICATIONBYTESOUTPERSEC = "replication_bytes_in";
+		public final static String OSTOTALMEMORY = "os_total_memory";
+		public final static String OSFREEMEMORY = "os_free_memory";
 	}
 
 	public interface Linux {
@@ -171,6 +173,12 @@ public interface Topic {
 
 		public final static String SUCCESS = "SUCCESS";
 		public final static String FAILED = "FAILED";
+
+		public final static String LOGSIZE = "logsize";
+		public final static String CAPACITY = "capacity";
+
+		public final static int BATCH_SIZE = 500;
+
 	}
 
 	public interface Component {

kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/util/StrUtils.java

@@ -19,6 +19,8 @@
 
 import java.text.DecimalFormat;
 
+import com.alibaba.fastjson.JSONObject;
+
 /**
  * String conversion tool.
  * 
@@ -64,11 +66,43 @@ public static String stringify(long byteNumber) {
 		}
 	}
 
+	/** Formatter byte to kb,mb or gb etc. */
+	public static JSONObject stringifyByObject(long byteNumber) {
+		JSONObject object = new JSONObject();
+		if (byteNumber / TB_IN_BYTES > 0) {
+			object.put("size", df.format((double) byteNumber / (double) TB_IN_BYTES));
+			object.put("type", "TB");
+			return object;
+		} else if (byteNumber / GB_IN_BYTES > 0) {
+			object.put("size", df.format((double) byteNumber / (double) GB_IN_BYTES));
+			object.put("type", "GB");
+			return object;
+		} else if (byteNumber / MB_IN_BYTES > 0) {
+			object.put("size", df.format((double) byteNumber / (double) MB_IN_BYTES));
+			object.put("type", "MB");
+			return object;
+		} else if (byteNumber / KB_IN_BYTES > 0) {
+			object.put("size", df.format((double) byteNumber / (double) KB_IN_BYTES));
+			object.put("type", "KB");
+			return object;
+		} else {
+			object.put("size", String.valueOf(byteNumber));
+			object.put("type", "B");
+			return object;
+		}
+	}
+
 	/** Formmatter number. */
 	public static double numberic(String number) {
 		DecimalFormat formatter = new DecimalFormat("###.##");
 		return Double.valueOf(formatter.format(Double.valueOf(number)));
 	}
+
+	/** Formmatter number. */
+	public static double numberic(String number,String format) {
+		DecimalFormat formatter = new DecimalFormat(format);
+		return Double.valueOf(formatter.format(Double.valueOf(number)));
+	}
 
 	/** Convert string number to double. */
 	public static long integer(double number) {

kafka-eagle-core/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>org.smartloli.kafka.eagle</groupId>
 		<artifactId>kafka-eagle</artifactId>
-		<version>1.3.5</version>
+		<version>1.3.6</version>
 	</parent>
 	<artifactId>kafka-eagle-core</artifactId>
 

kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.Set;
 
+import org.smartloli.kafka.eagle.common.protocol.BrokersInfo;
 import org.smartloli.kafka.eagle.common.protocol.DisplayInfo;
 import org.smartloli.kafka.eagle.common.protocol.KafkaSqlInfo;
 import org.smartloli.kafka.eagle.common.protocol.MetadataInfo;
@@ -51,7 +52,7 @@ public interface KafkaService {
 	public Set<String> getActiveTopic(String clusterAlias, String group);
 
 	/** Get all broker list from zookeeper. */
-	public String getAllBrokersInfo(String clusterAlias);
+	public List<BrokersInfo> getAllBrokersInfo(String clusterAlias);
 
 	/**
 	 * Get all topic info from zookeeper. Deprecated this method in the v1.3.4
@@ -108,9 +109,12 @@ public interface KafkaService {
 	/** Get kafka consumer group & topic. */
 	public String getKafkaConsumerGroupTopic(String clusterAlias, String group);
 
-	/** Get kafka sasl logsize . */
+	/** Get kafka topic history logsize . */
 	public long getKafkaLogSize(String clusterAlias, String topic, int partitionid);
 
+	/** Get kafka topic real logsize. */
+	public long getKafkaRealLogSize(String clusterAlias, String topic, int partitionid);
+
 	/** Get kafka sasl topic metadate. */
 	public List<MetadataInfo> findKafkaLeader(String clusterAlias, String topic);
 
@@ -123,10 +127,19 @@ public interface KafkaService {
 	/** Get consumer group all topics lag. */
 	public long getLag(String clusterAlias, String group, String topic);
 
-	/** Get kafka logsize by old version. */
+	/** Get kafka history logsize by old version. */
 	public long getLogSize(String clusterAlias, String topic, int partitionid);
 
+	/** Get kafka real logsize by old version. */
+	public long getRealLogSize(String clusterAlias, String topic, int partitionid);
+
 	/** Get topic metadata. */
 	public String getReplicasIsr(String clusterAlias, String topic, int partitionid);
 
+	/** Get kafka version. */
+	public String getKafkaVersion(String host, int port, String ids, String clusterAlias);
+
+	/** Get kafka os memory. */
+	public long getOSMemory(String host, int port, String property);
+
 }