From c38b1dc8d4a6e64d49bc8704e7e921ceb6621eb0 Mon Sep 17 00:00:00 2001 From: jeff Date: Fri, 17 Apr 2020 13:58:00 +0800 Subject: [PATCH 01/12] Save --- .gitignore | 3 +++ kafka-eagle-api/.gitignore | 1 + kafka-eagle-common/.gitignore | 1 + kafka-eagle-core/.gitignore | 1 + kafka-eagle-plugin/.gitignore | 1 + kafka-eagle-web/.gitignore | 1 + 6 files changed, 8 insertions(+) create mode 100644 .gitignore create mode 100644 kafka-eagle-api/.gitignore create mode 100644 kafka-eagle-common/.gitignore create mode 100644 kafka-eagle-core/.gitignore create mode 100644 kafka-eagle-plugin/.gitignore create mode 100644 kafka-eagle-web/.gitignore diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..58129053 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.classpath +.settings +.project diff --git a/kafka-eagle-api/.gitignore b/kafka-eagle-api/.gitignore new file mode 100644 index 00000000..b83d2226 --- /dev/null +++ b/kafka-eagle-api/.gitignore @@ -0,0 +1 @@ +/target/ diff --git a/kafka-eagle-common/.gitignore b/kafka-eagle-common/.gitignore new file mode 100644 index 00000000..b83d2226 --- /dev/null +++ b/kafka-eagle-common/.gitignore @@ -0,0 +1 @@ +/target/ diff --git a/kafka-eagle-core/.gitignore b/kafka-eagle-core/.gitignore new file mode 100644 index 00000000..b83d2226 --- /dev/null +++ b/kafka-eagle-core/.gitignore @@ -0,0 +1 @@ +/target/ diff --git a/kafka-eagle-plugin/.gitignore b/kafka-eagle-plugin/.gitignore new file mode 100644 index 00000000..b83d2226 --- /dev/null +++ b/kafka-eagle-plugin/.gitignore @@ -0,0 +1 @@ +/target/ diff --git a/kafka-eagle-web/.gitignore b/kafka-eagle-web/.gitignore new file mode 100644 index 00000000..b83d2226 --- /dev/null +++ b/kafka-eagle-web/.gitignore @@ -0,0 +1 @@ +/target/ From 9ea08f59ef4cd5aae13c7e2fbbfea769bd81648b Mon Sep 17 00:00:00 2001 From: jeff Date: Fri, 24 Apr 2020 16:03:10 +0800 Subject: [PATCH 02/12] Updata local config --- .../src/main/resources/system-config.properties | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/kafka-eagle-common/src/main/resources/system-config.properties b/kafka-eagle-common/src/main/resources/system-config.properties index aafe35ba..40db4a56 100644 --- a/kafka-eagle-common/src/main/resources/system-config.properties +++ b/kafka-eagle-common/src/main/resources/system-config.properties @@ -2,8 +2,8 @@ # multi zookeeper & kafka cluster list ###################################### kafka.eagle.zk.cluster.alias=cluster1 -cluster1.zk.list=127.0.0.1:2181 -cluster1_0_1.zk.list=127.0.0.1:2181/v_1_0_1 +cluster1.zk.list=10.100.173.206:2181,10.100.173.207:2181,10.100.173.208:2181 +#cluster1_0_1.zk.list=127.0.0.1:2181/v_1_0_1 cluster2.zk.list=127.0.0.1:2181/plain cluster3.zk.list=127.0.0.1:2181/scram cluster4.zk.list=dn1:2181 @@ -55,13 +55,14 @@ kafka.eagle.topic.token=keadmin ###################################### # kafka sasl authenticate ###################################### -cluster1.kafka.eagle.sasl.enable=false +cluster1.kafka.eagle.sasl.enable=true cluster1.kafka.eagle.sasl.protocol=SASL_PLAINTEXT -cluster1.kafka.eagle.sasl.mechanism=SCRAM-SHA-256 -cluster1.kafka.eagle.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret"; +#cluster1.kafka.eagle.sasl.mechanism=SCRAM-SHA-256 +cluster1.kafka.eagle.sasl.mechanism=PLAIN +cluster1.kafka.eagle.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin_not_used_tcyx"; cluster1.kafka.eagle.sasl.client.id= cluster1.kafka.eagle.sasl.cgroup.enable=false -cluster1.kafka.eagle.sasl.cgroup.topics=kafka20191230,kafka20191217 +cluster1.kafka.eagle.sasl.cgroup.topics= cluster4.kafka.eagle.sasl.enable=false cluster4.kafka.eagle.sasl.protocol=SASL_PLAINTEXT @@ -73,8 +74,8 @@ cluster4.kafka.eagle.sasl.client.id= # kafka jdbc driver address ###################################### kafka.eagle.driver=com.mysql.jdbc.Driver -kafka.eagle.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull -kafka.eagle.username=root +kafka.eagle.url=jdbc:mysql://10.100.172.33:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull +kafka.eagle.username=dev kafka.eagle.password=123456 #kafka.eagle.driver=org.sqlite.JDBC From be057d32a02edac979a57b295352df69430be14f Mon Sep 17 00:00:00 2001 From: jeff Date: Mon, 11 May 2020 17:43:39 +0800 Subject: [PATCH 03/12] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E7=94=A8=E6=88=B7ACL?= =?UTF-8?q?=E6=8E=88=E6=9D=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eagle/core/factory/KafkaService.java | 2 +- .../eagle/core/factory/KafkaServiceImpl.java | 51 ++++++++++++++++++- .../eagle/web/controller/TopicController.java | 8 ++- .../webapp/WEB-INF/views/topic/create.jsp | 14 +++++ 4 files changed, 72 insertions(+), 3 deletions(-) diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java index 1b4e8ed0..35d1c572 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java @@ -75,7 +75,7 @@ public interface KafkaService { public Map getKafkaOffset(String clusterAlias,String group,String topic, Set partitionids); /** Use kafka console comand to create topic. */ - public Map create(String clusterAlias, String topicName, String partitions, String replic); + public Map create(String clusterAlias, String topicName, String partitions, String replic, String username); /** Use kafka console command to delete topic. */ public Map delete(String clusterAlias, String topicName); diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java index 736bb560..89b263ff 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java @@ -44,6 +44,7 @@ import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.ConsumerGroupListing; +import org.apache.kafka.clients.admin.CreateAclsResult; import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult; import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions; import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult; @@ -58,7 +59,14 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.acl.AccessControlEntry; +import org.apache.kafka.common.acl.AclBinding; +import org.apache.kafka.common.acl.AclOperation; +import org.apache.kafka.common.acl.AclPermissionType; import org.apache.kafka.common.config.SaslConfigs; +import org.apache.kafka.common.resource.PatternType; +import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourceType; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.zookeeper.data.Stat; @@ -450,7 +458,7 @@ public JSONObject zkCliStatus(String clusterAlias) { * Replic numbers. * @return Map. */ - public Map create(String clusterAlias, String topicName, String partitions, String replic) { + public Map create(String clusterAlias, String topicName, String partitions, String replic, String username) { Map targets = new HashMap(); int brokers = getAllBrokersInfo(clusterAlias).size(); if (Integer.parseInt(replic) > brokers) { @@ -470,6 +478,11 @@ public Map create(String clusterAlias, String topicName, String adminClient = AdminClient.create(prop); NewTopic newTopic = new NewTopic(topicName, Integer.valueOf(partitions), Short.valueOf(replic)); adminClient.createTopics(Collections.singleton(newTopic)).all().get(); + + if (!"".equalsIgnoreCase(username)) { + createAclForTopic(adminClient, topicName, username, "group_" + username); + } + } catch (Exception e) { LOG.info("Create kafka topic has error, msg is " + e.getMessage()); e.printStackTrace(); @@ -481,6 +494,42 @@ public Map create(String clusterAlias, String topicName, String targets.put("info", "Create topic[" + topicName + "] has successed,partitions numbers is [" + partitions + "],replication-factor numbers is [" + replic + "]"); return targets; } + + /** + * + * @param client + * @param topicName + * @param username + */ + private void createAclForTopic(AdminClient client, String topicName, String username, String groupnname) { + String principal = "User:" + username; + String host = "*"; + ArrayList as = new ArrayList<>(); + + ResourcePattern pattern = new ResourcePattern(ResourceType.TOPIC, topicName, PatternType.LITERAL); + + AclBinding aclBindingW = new AclBinding(pattern, new AccessControlEntry(principal, host, AclOperation.WRITE, AclPermissionType.ALLOW)); + AclBinding aclBindingR = new AclBinding(pattern, new AccessControlEntry(principal, host, AclOperation.READ, AclPermissionType.ALLOW)); + AclBinding aclBindingD = new AclBinding(pattern, new AccessControlEntry(principal, host, AclOperation.DESCRIBE, AclPermissionType.ALLOW)); + as.add(aclBindingW); + as.add(aclBindingR); + as.add(aclBindingD); + + //group + ResourcePattern resourcePatternGroup = new ResourcePattern(ResourceType.GROUP, groupnname, PatternType.LITERAL); + AclBinding aclBindingG = new AclBinding(resourcePatternGroup, new AccessControlEntry(principal, host, AclOperation.READ, AclPermissionType.ALLOW)); + as.add(aclBindingG); + + CreateAclsResult result = client.createAcls(as); + + try { + result.values().get(aclBindingW).get(30, TimeUnit.SECONDS); + } catch (Exception e) { + LOG.info("Error creating acl for topic " + topicName, e); + e.printStackTrace(); + } + } + /** Delete topic to kafka cluster. */ public Map delete(String clusterAlias, String topicName) { diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/TopicController.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/TopicController.java index 3744f814..bff4a6bc 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/TopicController.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/TopicController.java @@ -526,8 +526,14 @@ public ModelAndView topicAddForm(HttpSession session, HttpServletResponse respon String ke_topic_name = request.getParameter("ke_topic_name"); String ke_topic_partition = request.getParameter("ke_topic_partition"); String ke_topic_repli = request.getParameter("ke_topic_repli"); + String ke_topic_username = request.getParameter("ke_topic_username"); + String clusterAlias = session.getAttribute(KConstants.SessionAlias.CLUSTER_ALIAS).toString(); - Map respons = kafkaService.create(clusterAlias, ke_topic_name, ke_topic_partition, ke_topic_repli); + + Map respons = kafkaService.create(clusterAlias, ke_topic_name, ke_topic_partition, ke_topic_repli, ke_topic_username); + + + if ("success".equals(respons.get("status"))) { session.removeAttribute("Submit_Status"); session.setAttribute("Submit_Status", respons.get("info")); diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/topic/create.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/topic/create.jsp index c0bbf502..b7b69924 100644 --- a/kafka-eagle-web/src/main/webapp/WEB-INF/views/topic/create.jsp +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/topic/create.jsp @@ -77,6 +77,20 @@ class="fa fa-info-circle"> Replication Factor parameters must be numeric . Pay attention to available brokers must be larger than replication factor . + +
+ + + +
+ - - - -
-
-
-
- Topic Property -
-
- -
-
-
-
-
- - -
-
- -
-
- -
-
- - - -
- -
- -
- + +
+
@@ -131,4 +122,4 @@ return true; } - + \ No newline at end of file From 0fb7d60a18fa2ff1fe7ee5739995f58600eba97e Mon Sep 17 00:00:00 2001 From: jeff Date: Sat, 20 Feb 2021 10:47:13 +0800 Subject: [PATCH 08/12] =?UTF-8?q?ACL=20=E5=88=97=E8=A1=A8=E9=A1=B5?= =?UTF-8?q?=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eagle/core/factory/KafkaService.java | 8 + .../eagle/core/factory/KafkaServiceImpl.java | 66 +++++ .../eagle/web/controller/AclController.java | 68 +++++ .../kafka/eagle/web/service/AclService.java | 33 +++ .../web/service/impl/AclServiceImpl.java | 52 ++++ .../main/webapp/WEB-INF/views/acls/acls.jsp | 118 ++++++++ .../WEB-INF/views/public/plus/navbar.jsp | 7 + .../src/main/webapp/media/js/main/acl/acls.js | 41 +++ .../media/js/main/acl/offset.consumer.js | 74 +++++ .../media/js/main/acl/offset.realtime.js | 258 ++++++++++++++++++ .../webapp/media/js/public/plus/navbar.js | 2 + 11 files changed, 727 insertions(+) create mode 100644 kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java create mode 100644 kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java create mode 100644 kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java create mode 100644 kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp create mode 100644 kafka-eagle-web/src/main/webapp/media/js/main/acl/acls.js create mode 100644 kafka-eagle-web/src/main/webapp/media/js/main/acl/offset.consumer.js create mode 100644 kafka-eagle-web/src/main/webapp/media/js/main/acl/offset.realtime.js diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java index 0d01f136..70077c96 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java @@ -20,6 +20,8 @@ import org.apache.kafka.common.TopicPartition; import org.smartloli.kafka.eagle.common.protocol.*; +import com.alibaba.fastjson.JSONArray; + import java.util.List; import java.util.Map; import java.util.Properties; @@ -257,5 +259,11 @@ public interface KafkaService { * Set kafka ssl acl. */ public void ssl(Properties props, String clusterAlias); + + /** + * Get kafka acls . + */ + public JSONArray getKafkaAcl(String clusterAlias); + } diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java index e7c4d470..5e38fc6e 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java @@ -40,13 +40,16 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.acl.AccessControlEntry; +import org.apache.kafka.common.acl.AccessControlEntryFilter; import org.apache.kafka.common.acl.AclBinding; +import org.apache.kafka.common.acl.AclBindingFilter; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.config.SslConfigs; import org.apache.kafka.common.resource.PatternType; import org.apache.kafka.common.resource.ResourcePattern; +import org.apache.kafka.common.resource.ResourcePatternFilter; import org.apache.kafka.common.resource.ResourceType; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; @@ -1782,5 +1785,68 @@ public String getUsedMemory(String clusterAlias, String host, int port) { } return memory; } + + + /** + * Get kafka 0.10.x, 1.x, 2.x acl metadata. + */ + public JSONArray getKafkaAcl(String clusterAlias) { + + Properties prop = new Properties(); + JSONArray acls = new JSONArray(); + prop.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, parseBrokerServer(clusterAlias)); + + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.sasl.enable")) { + sasl(prop, clusterAlias); + } + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.ssl.enable")) { + ssl(prop, clusterAlias); + } + + AdminClient adminClient = null; + try { + adminClient = AdminClient.create(prop); + + AclBindingFilter aclBindingFilter = new AclBindingFilter( + ResourcePatternFilter.ANY, + new AccessControlEntryFilter( + null, + "*", + AclOperation.ANY, + AclPermissionType.ANY + ) + ); + + java.util.Iterator itorA = adminClient.describeAcls(aclBindingFilter).values().get().iterator(); + while (itorA.hasNext()) { + AclBinding ab = itorA.next(); + AccessControlEntry e = ab.entry(); + JSONObject entry = new JSONObject(); + entry.put("host", e.host()); + entry.put("operation",e.operation().name()); + entry.put("principal", e.principal()); + entry.put("permissionType",e.permissionType().name()); + + ResourcePattern p = ab.pattern(); + JSONObject pattern = new JSONObject(); + pattern.put("name", p.name()); + pattern.put("patternType", p.patternType().name()); + pattern.put("resourceType", p.resourceType().name()); + + JSONObject acl = new JSONObject(); + acl.put("entry", entry); + acl.put("pattern", pattern); + acls.add(acl); + } + + } catch (Exception e) { + LOG.error("Get kafka consumer has error,msg is " + e.getMessage()); + e.printStackTrace(); + } finally { + adminClient.close(); + } + + return acls; + } } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java new file mode 100644 index 00000000..b8eb9fa7 --- /dev/null +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.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.web.controller; + +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; +import javax.servlet.http.HttpSession; + +import org.smartloli.kafka.eagle.common.util.KConstants; +import org.smartloli.kafka.eagle.web.service.AclService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.ResponseBody; +import org.springframework.web.servlet.ModelAndView; + +import com.alibaba.fastjson.JSONArray; + +/** + * Kafka acl controller to viewer data. + * + * @author jeff + * + */ +@Controller +public class AclController { + + /** Kafka acl service interface. */ + @Autowired + private AclService aclService; + + /** acls viewer. */ + @RequestMapping(value = "/acls", method = RequestMethod.GET) + public ModelAndView aclsView(HttpSession session) { + ModelAndView mav = new ModelAndView(); + mav.setViewName("/acls/acls"); + return mav; + } + + + /** Get acl data by ajax. */ + @RequestMapping(value = "/acls/list", method = RequestMethod.GET) + @ResponseBody + public JSONArray consumersGraphAjax(HttpServletResponse response, HttpServletRequest request) { + HttpSession session = request.getSession(); + String clusterAlias = session.getAttribute(KConstants.SessionAlias.CLUSTER_ALIAS).toString(); + JSONArray result = aclService.getAcls(clusterAlias); + + return result; + } + +} diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java new file mode 100644 index 00000000..adabee37 --- /dev/null +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java @@ -0,0 +1,33 @@ +/** + * 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.web.service; + +import com.alibaba.fastjson.JSONArray; + +/** + * Kafka acl data interface. + * + * @author jeff. + * + * + */ +public interface AclService { + + /** Get acl data interface. */ + public JSONArray getAcls(String clusterAlias); +} diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java new file mode 100644 index 00000000..331a852b --- /dev/null +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java @@ -0,0 +1,52 @@ +/** + * 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.web.service.impl; + +import org.smartloli.kafka.eagle.core.factory.KafkaFactory; +import org.smartloli.kafka.eagle.core.factory.KafkaService; +import org.smartloli.kafka.eagle.web.service.AclService; +import org.springframework.stereotype.Service; + +import com.alibaba.fastjson.JSONArray; + +/** + * Kafka acl data interface, and set up the return data set. + * + * @author jeff. + * + */ +@Service +public class AclServiceImpl implements AclService { + + /** Kafka service interface. */ + private KafkaService kafkaService = new KafkaFactory().create(); + + /** Get acl data from kafka cluster. */ + public JSONArray getAcls(String clusterAlias) { + JSONArray activeTopics = kafkaService.getKafkaAcl(clusterAlias); + + return activeTopics; + } + +} + + + + + + diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp new file mode 100644 index 00000000..421a3578 --- /dev/null +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp @@ -0,0 +1,118 @@ +<%@ page language="java" contentType="text/html; charset=UTF-8" + pageEncoding="UTF-8" %> +<%@ taglib uri="http://java.sun.com/jsp/jstl/core" prefix="c" %> + + + + + + + + + + + + Consumers - KafkaEagle + + + + + + + + +
+
+ +
+
+
+
+

Consumers

+
    +
  1. ACl
    2. +
  2. List
    3. +
+
+ + List all Acls. +
+ +
+
+
+
+ List +
+
+
+ + + + + + + + + + + +
resourceTypenameprincipalhostoperationpermissionType
+
+
+
+
+
+ + + +
+
+ +
+
+ + + + + + + diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp index 51c8f1ef..b5c5a600 100644 --- a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp @@ -67,6 +67,13 @@
Consumers + +
+ +
+ ACL + +
Performance
" + topic + ""); + + var offset = 0; + + function offsetDetail() { + $("#offset_topic_info").append("
PartitionLogSizeOffsetLagOwnerCreatedModify
"); + if (offset > 0) { + $("#div_children" + (offset - 1)).remove(); + } + $("#result_children" + offset).dataTable({ + "searching" : false, + "bSort" : false, + "bLengthChange" : false, + "bProcessing" : true, + "bServerSide" : true, + "fnServerData" : retrieveData, + "sAjaxSource" : "/consumer/offset/group/topic/ajax", + "aoColumns" : [ { + "mData" : 'partition' + }, { + "mData" : 'logsize' + }, { + "mData" : 'offset' + }, { + "mData" : 'lag' + }, { + "mData" : 'owner' + }, { + "mData" : 'created' + }, { + "mData" : 'modify' + } ] + }); + + offset++; + } + + function retrieveData(sSource, aoData, fnCallback) { + $.ajax({ + "type" : "get", + "contentType" : "application/json", + "url" : sSource, + "dataType" : "json", + "data" : { + aoData : JSON.stringify(aoData), + group : group, + topic : topic + }, + "success" : function(data) { + fnCallback(data) + } + }); + } + + offsetDetail(); + + // 5s/per to the background service request details of the state of offset . + // setInterval(offsetDetail, 1000 * 30); +}); \ No newline at end of file diff --git a/kafka-eagle-web/src/main/webapp/media/js/main/acl/offset.realtime.js b/kafka-eagle-web/src/main/webapp/media/js/main/acl/offset.realtime.js new file mode 100644 index 00000000..75cd1a6f --- /dev/null +++ b/kafka-eagle-web/src/main/webapp/media/js/main/acl/offset.realtime.js @@ -0,0 +1,258 @@ +$(document).ready(function() { + + try{ + var path = window.location.href; + $("#ke_consumer_offsets_a").attr("href","/consumers/offset/?" + path.split("?")[1]); + }catch (e) { + console.error(e); + } + + function getQueryString(name) { + var reg = new RegExp("(^|&)" + name + "=([^&]*)(&|$)", "i"); + var r = window.location.search.substr(1).match(reg); + var context = ""; + if (r != null) + context = r[2]; + reg = null; + r = null; + return context == null || context == "" || context == "undefined" ? "" : context; + } + + var group = getQueryString("group"); + var topic = getQueryString("topic"); + + $("#topic_lag_name_header").find("strong").text("Consumer Blocking Metrics (" + topic + ")"); + $("#topic_producer_name_header").find("strong").text("Producer Performance Metrics (" + topic + ")"); + $("#topic_consumer_name_header").find("strong").text("Consumer Performance Metrics (" + topic + ")"); + + lagOption = { + backgroundColor : "#fff", + tooltip : { + trigger : 'axis', + axisPointer : { + type : 'cross', + label : { + backgroundColor : '#6a7985' + } + } + }, + legend : { + data : [ 'Lag' ] + }, + xAxis : { + type : 'category', + boundaryGap : false, + data : [] + }, + dataZoom : { + show : true, + start : 30 + }, + grid : { + bottom : "70px", + left : "90px", + right : "90px" + }, + yAxis : { + type : 'value' + }, + series : { + type : 'line', + symbol : "none", + name : "Lag", + smooth : true, + areaStyle : { + opacity : 0.1 + }, + data : [] + } + }; + + producerOption = { + backgroundColor : "#fff", + tooltip : { + trigger : 'axis', + axisPointer : { + type : 'cross', + label : { + backgroundColor : '#6a7985' + } + } + }, + legend : { + data : [ 'Producer' ] + }, + xAxis : { + type : 'category', + boundaryGap : false, + data : [] + }, + dataZoom : { + show : true, + start : 30 + }, + grid : { + bottom : "70px", + left : "90px", + right : "90px" + }, + yAxis : { + type : 'value' + }, + series : { + type : 'line', + symbol : "none", + name : "Producer", + smooth : true, + areaStyle : { + opacity : 0.1 + }, + data : [] + } + }; + + consumerOption = { + backgroundColor : "#fff", + tooltip : { + trigger : 'axis', + axisPointer : { + type : 'cross', + label : { + backgroundColor : '#6a7985' + } + } + }, + legend : { + data : [ 'Consumer' ] + }, + xAxis : { + type : 'category', + boundaryGap : false, + data : [] + }, + dataZoom : { + show : true, + start : 30 + }, + grid : { + bottom : "70px", + left : "90px", + right : "90px" + }, + yAxis : { + type : 'value' + }, + series : { + type : 'line', + symbol : "none", + name : "Consumer", + smooth : true, + areaStyle : { + opacity : 0.1 + }, + data : [] + } + }; + + var lagChart = echarts.init(document.getElementById('lag_chart'), 'macarons'); + var producerChart = echarts.init(document.getElementById('producer_chart'), 'macarons'); + var consumerChart = echarts.init(document.getElementById('consumer_chart'), 'macarons'); + lagChart.setOption(lagOption); + producerChart.setOption(producerOption); + consumerChart.setOption(consumerOption); + + $("#lag_chart").resize(function () { + var opt_lagChart=lagChart.getOption(); + lagChart.clear(); + lagChart.resize({width:$("#lag_chart").css('width')}); + lagChart.setOption(opt_lagChart); + }); + + $("#producer_chart").resize(function () { + var opt_producerChart=producerChart.getOption(); + producerChart.clear(); + producerChart.resize({width:$("#producer_chart").css('width')}); + producerChart.setOption(opt_producerChart); + }); + + $("#consumer_chart").resize(function () { + var opt_consumerChart=consumerChart.getOption(); + consumerChart.clear(); + consumerChart.resize({width:$("#consumer_chart").css('width')}); + consumerChart.setOption(opt_consumerChart); + }); + + var start = moment(); + var end = moment(); + + function cb(start, end) { + $('#reportrange span').html(start.format('YYYY-MM-DD') + ' To ' + end.format('YYYY-MM-DD')); + } + + var reportrange = $('#reportrange').daterangepicker({ + startDate : start, + endDate : end, + ranges : { + 'Today' : [ moment(), moment() ], + 'Yesterday' : [ moment().subtract(1, 'days'), moment() ], + 'Lastest 3 days' : [ moment().subtract(3, 'days'), moment() ], + 'Lastest 7 days' : [ moment().subtract(6, 'days'), moment() ] + } + }, cb); + + cb(start, end); + var stime = reportrange[0].innerText.replace(/-/g, '').split("To")[0].trim(); + var etime = reportrange[0].innerText.replace(/-/g, '').split("To")[1].trim(); + + function offserRealtime(stime, etime) { + $.ajax({ + type : 'get', + dataType : 'json', + url : '/consumer/offset/group/topic/realtime/ajax?group=' + group + '&topic=' + topic + '&stime=' + stime + '&etime=' + etime, + success : function(datas) { + if (datas != null) { + // Area Chart + lagOption.xAxis.data = datas.lag.x; + lagOption.series.data = datas.lag.y; + lagChart.setOption(lagOption); + producerOption.xAxis.data = datas.producer.x; + producerOption.series.data = datas.producer.y; + producerChart.setOption(producerOption); + consumerOption.xAxis.data = datas.consumer.x; + consumerOption.series.data = datas.consumer.y; + consumerChart.setOption(consumerOption); + datas = []; + } + } + }); + } + + function offserRateRealtime() { + $.ajax({ + type : 'get', + dataType : 'json', + url : '/consumer/offset/rate/group/topic/realtime/ajax?group=' + group + '&topic=' + topic, + success : function(datas) { + if (datas != null) { + // Consumer & Producer Rate + $("#producer_rate").text(datas.ins); + $("#consumer_rate").text(datas.outs); + datas = []; + } + } + }); + } + + reportrange.on('apply.daterangepicker', function(ev, picker) { + stime = reportrange[0].innerText.replace(/-/g, '').split("To")[0].trim(); + etime = reportrange[0].innerText.replace(/-/g, '').split("To")[1].trim(); + offserRealtime(stime, etime); + }); + + offserRealtime(stime, etime); + offserRateRealtime(); + setInterval(function() { + offserRealtime(stime, etime) + }, 1000 * 60 * 1); + setInterval(offserRateRealtime, 1000 * 60 * 1); +}); \ No newline at end of file diff --git a/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js b/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js index d262936e..9c3b4214 100644 --- a/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js +++ b/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js @@ -23,6 +23,8 @@ $("#ke_navbar_topic_tasks").addClass("active"); } else if (url.indexOf("/consumers") > -1) { $("#ke_navbar_consumers").addClass("active"); + } else if (url.indexOf("/acls") > -1) { + $("#ke_navbar_acls").addClass("active"); } else if ((url.indexOf("/cluster") > -1)) { $("#collapseCluster").addClass('show') if (url.indexOf("/cluster/info") > -1) { From 901ca6206a966d0e73ec92760bb2335c0d31ccf5 Mon Sep 17 00:00:00 2001 From: jeff Date: Sat, 20 Feb 2021 18:23:54 +0800 Subject: [PATCH 09/12] =?UTF-8?q?topic=20=E8=AF=A6=E6=83=85=E9=A1=B5?= =?UTF-8?q?=E9=9D=A2=E6=98=BE=E7=A4=BAACL=E5=88=97=E8=A1=A8=20--story=3D10?= =?UTF-8?q?78831=20--user=3D=E5=86=AF=E4=BF=8Afj14605=20=E7=81=AB=E8=BD=A6?= =?UTF-8?q?=E7=A5=A8=E5=90=8E=E5=8F=B0=E8=BF=AD=E4=BB=A3=EF=BC=88Q2?= =?UTF-8?q?=EF=BC=89=20https://www.tapd.cn/34180125/s/1157602?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eagle/core/factory/KafkaService.java | 2 + .../eagle/core/factory/KafkaServiceImpl.java | 95 ++++++++++++++----- .../eagle/web/controller/AclController.java | 14 ++- .../kafka/eagle/web/service/AclService.java | 4 + .../web/service/impl/AclServiceImpl.java | 7 ++ .../WEB-INF/views/public/plus/navbar.jsp | 30 ++++-- .../webapp/WEB-INF/views/topic/topic_meta.jsp | 26 +++++ .../webapp/media/js/main/topic/topic.meta.js | 34 +++++++ .../webapp/media/js/public/plus/navbar.js | 3 +- 9 files changed, 181 insertions(+), 34 deletions(-) diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java index 70077c96..e72b67f7 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java @@ -265,5 +265,7 @@ public interface KafkaService { */ public JSONArray getKafkaAcl(String clusterAlias); + public JSONArray getKafkaAclBYTopicName(String clusterAlias, String topicname); + } diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java index 5e38fc6e..b766689c 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java @@ -79,6 +79,7 @@ import java.lang.management.MemoryMXBean; import java.util.*; import java.util.Map.Entry; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; /** @@ -1791,7 +1792,6 @@ public String getUsedMemory(String clusterAlias, String host, int port) { * Get kafka 0.10.x, 1.x, 2.x acl metadata. */ public JSONArray getKafkaAcl(String clusterAlias) { - Properties prop = new Properties(); JSONArray acls = new JSONArray(); prop.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, parseBrokerServer(clusterAlias)); @@ -1811,42 +1811,87 @@ public JSONArray getKafkaAcl(String clusterAlias) { ResourcePatternFilter.ANY, new AccessControlEntryFilter( null, - "*", + null, AclOperation.ANY, AclPermissionType.ANY ) ); + acls = _getACL(adminClient, aclBindingFilter); + } catch (Exception e) { + LOG.error("Get kafka acl has error,msg is " + e.getMessage()); + e.printStackTrace(); + } finally { + adminClient.close(); + } + + return acls; + } + + + /** + * Get kafka 0.10.x, 1.x, 2.x acl metadata. + */ + public JSONArray getKafkaAclBYTopicName(String clusterAlias, String topicname) { + Properties prop = new Properties(); + JSONArray acls = new JSONArray(); + prop.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, parseBrokerServer(clusterAlias)); + + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.sasl.enable")) { + sasl(prop, clusterAlias); + } + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.ssl.enable")) { + ssl(prop, clusterAlias); + } + + AdminClient adminClient = null; + try { + adminClient = AdminClient.create(prop); - java.util.Iterator itorA = adminClient.describeAcls(aclBindingFilter).values().get().iterator(); - while (itorA.hasNext()) { - AclBinding ab = itorA.next(); - AccessControlEntry e = ab.entry(); - JSONObject entry = new JSONObject(); - entry.put("host", e.host()); - entry.put("operation",e.operation().name()); - entry.put("principal", e.principal()); - entry.put("permissionType",e.permissionType().name()); - - ResourcePattern p = ab.pattern(); - JSONObject pattern = new JSONObject(); - pattern.put("name", p.name()); - pattern.put("patternType", p.patternType().name()); - pattern.put("resourceType", p.resourceType().name()); - - JSONObject acl = new JSONObject(); - acl.put("entry", entry); - acl.put("pattern", pattern); - acls.add(acl); - } - + AclBindingFilter aclBindingFilter = new AclBindingFilter( + new ResourcePatternFilter( + ResourceType.TOPIC, + topicname, + PatternType.ANY + ), + AccessControlEntryFilter.ANY + ); + acls = _getACL(adminClient, aclBindingFilter); } catch (Exception e) { - LOG.error("Get kafka consumer has error,msg is " + e.getMessage()); + LOG.error("Get kafka acl has error,msg is " + e.getMessage()); e.printStackTrace(); } finally { adminClient.close(); } return acls; + } + + + + public JSONArray _getACL(AdminClient adminClient, AclBindingFilter aclBindingFilter) throws InterruptedException, ExecutionException { + JSONArray acls = new JSONArray(); + java.util.Iterator itorA = adminClient.describeAcls(aclBindingFilter).values().get().iterator(); + while (itorA.hasNext()) { + AclBinding ab = itorA.next(); + AccessControlEntry e = ab.entry(); + JSONObject entry = new JSONObject(); + entry.put("host", e.host()); + entry.put("operation",e.operation().name()); + entry.put("principal", e.principal()); + entry.put("permissionType",e.permissionType().name()); + + ResourcePattern p = ab.pattern(); + JSONObject pattern = new JSONObject(); + pattern.put("name", p.name()); + pattern.put("patternType", p.patternType().name()); + pattern.put("resourceType", p.resourceType().name()); + + JSONObject acl = new JSONObject(); + acl.put("entry", entry); + acl.put("pattern", pattern); + acls.add(acl); + } + return acls; } } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java index b8eb9fa7..6bd96170 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java @@ -25,6 +25,7 @@ import org.smartloli.kafka.eagle.web.service.AclService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.ResponseBody; @@ -57,12 +58,23 @@ public ModelAndView aclsView(HttpSession session) { /** Get acl data by ajax. */ @RequestMapping(value = "/acls/list", method = RequestMethod.GET) @ResponseBody - public JSONArray consumersGraphAjax(HttpServletResponse response, HttpServletRequest request) { + public JSONArray aclslist(HttpServletResponse response, HttpServletRequest request) { HttpSession session = request.getSession(); String clusterAlias = session.getAttribute(KConstants.SessionAlias.CLUSTER_ALIAS).toString(); JSONArray result = aclService.getAcls(clusterAlias); return result; } + + + @RequestMapping(value = "/acls/topic/{tname}", method = RequestMethod.GET) + @ResponseBody + public JSONArray aclstopic(@PathVariable("tname") String tname, HttpServletResponse response, HttpServletRequest request, HttpSession session) { + String clusterAlias = session.getAttribute(KConstants.SessionAlias.CLUSTER_ALIAS).toString(); + JSONArray result = aclService.getTopicAcls(clusterAlias, tname); + + return result; + } + } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java index adabee37..ab725713 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java @@ -30,4 +30,8 @@ public interface AclService { /** Get acl data interface. */ public JSONArray getAcls(String clusterAlias); + + /** Get acl data interface. */ + public JSONArray getTopicAcls(String clusterAlias, String topicname); + } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java index 331a852b..0e0959cf 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java @@ -43,6 +43,13 @@ public JSONArray getAcls(String clusterAlias) { return activeTopics; } + @Override + public JSONArray getTopicAcls(String clusterAlias, String topicname) { + JSONArray acls = kafkaService.getKafkaAclBYTopicName(clusterAlias, topicname); + + return acls; + } + } diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp index b5c5a600..b1f4147a 100644 --- a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp @@ -45,6 +45,7 @@ class="fas fa-cube fa-sm fa-fw mr-1">Hub
+
- + + - + + + +
- +
- ACL - + Consumers + + +
Performance
+ + + + + \ No newline at end of file diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/add_success.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/add_success.jsp new file mode 100644 index 00000000..5752a409 --- /dev/null +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/add_success.jsp @@ -0,0 +1,40 @@ +<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%> +<%@ taglib uri="http://java.sun.com/jsp/jstl/core" prefix="c"%> + + + + + +Success - KafkaEagle + + + + + +
+
+ +
+
+
+
+

Topic

+
    +
  1. ACL
    2. +
  2. CreateGroup
    3. +
+
+ + ${Submit_Status} View Details + +
+
+
+ +
+
+ + + + + \ No newline at end of file diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/createGroup.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/createGroup.jsp new file mode 100644 index 00000000..4a0f504a --- /dev/null +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/createGroup.jsp @@ -0,0 +1,106 @@ +<%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%> +<%@ taglib uri="http://java.sun.com/jsp/jstl/core" prefix="c"%> + + + + + + + + + + + + +Topic Create - KafkaEagle + + + + + +
+
+ +
+
+
+
+

Topic

+
    +
  1. ACL
    2. +
  2. CreateGroup
    3. +
+
+ + Create a new kafka's topic. +
+ +
+
+
+
+ Topic Property +
+
+
+
+
+
+ + + +
+ +
+ + + +
+ + + +
+
+
+
+
+
+
+
+
+ +
+
+ + + + + + \ No newline at end of file diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp index b1f4147a..2ff98e5a 100644 --- a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp @@ -76,7 +76,9 @@
diff --git a/kafka-eagle-web/src/main/webapp/media/js/main/acl/add.result.js b/kafka-eagle-web/src/main/webapp/media/js/main/acl/add.result.js new file mode 100644 index 00000000..e69de29b diff --git a/kafka-eagle-web/src/main/webapp/media/js/main/acl/createGroup.js b/kafka-eagle-web/src/main/webapp/media/js/main/acl/createGroup.js new file mode 100644 index 00000000..f2f1df2e --- /dev/null +++ b/kafka-eagle-web/src/main/webapp/media/js/main/acl/createGroup.js @@ -0,0 +1,3 @@ +/** + * + */ \ No newline at end of file diff --git a/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js b/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js index 08157290..7b306847 100644 --- a/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js +++ b/kafka-eagle-web/src/main/webapp/media/js/public/plus/navbar.js @@ -24,8 +24,12 @@ } else if (url.indexOf("/consumers") > -1) { $("#ke_navbar_consumers").addClass("active"); } else if (url.indexOf("/acls") > -1) { - $("#acls").addClass('show'); - $("#ke_navbar_acls").addClass("active"); + $("#acls").addClass('show'); + if (url.indexOf("/acls/createGroup") > -1) { + $("#ke_navbar_acls_createGroup").addClass("active"); + } else { + $("#ke_navbar_acls").addClass("active"); + } } else if ((url.indexOf("/cluster") > -1)) { $("#collapseCluster").addClass('show') if (url.indexOf("/cluster/info") > -1) { From 95f344d20b21461f2b85589c6da618d08285c0b3 Mon Sep 17 00:00:00 2001 From: jeff Date: Tue, 23 Feb 2021 17:53:44 +0800 Subject: [PATCH 11/12] =?UTF-8?q?=E5=88=A0=E9=99=A4ACL=E6=93=8D=E4=BD=9C?= =?UTF-8?q?=20--story=3D1078831=20--user=3D=E5=86=AF=E4=BF=8Afj14605=20?= =?UTF-8?q?=E7=81=AB=E8=BD=A6=E7=A5=A8=E5=90=8E=E5=8F=B0=E8=BF=AD=E4=BB=A3?= =?UTF-8?q?=EF=BC=88Q2=EF=BC=89=20https://www.tapd.cn/34180125/s/1157602?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eagle/core/factory/KafkaService.java | 3 + .../eagle/core/factory/KafkaServiceImpl.java | 58 ++++++++++++++++++- .../eagle/web/controller/AclController.java | 11 ++++ .../kafka/eagle/web/service/AclService.java | 3 + .../web/service/impl/AclServiceImpl.java | 8 ++- .../main/webapp/WEB-INF/views/acls/acls.jsp | 3 +- .../WEB-INF/views/public/plus/navbar.jsp | 2 +- .../src/main/webapp/media/js/main/acl/acls.js | 37 ++++++++++++ 8 files changed, 117 insertions(+), 8 deletions(-) diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java index 96d7a94b..8f21edae 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaService.java @@ -21,6 +21,7 @@ import org.smartloli.kafka.eagle.common.protocol.*; import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; import java.util.List; import java.util.Map; @@ -269,5 +270,7 @@ public interface KafkaService { Map createGroup(String clusterAlias, String userName, String groupName); + String deleteAcl(String clusterAlias, JSONObject jsonObject); + } diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java index c285d576..4e9edb23 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java @@ -25,6 +25,8 @@ import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.ConsumerGroupListing; import org.apache.kafka.clients.admin.CreateAclsResult; +import org.apache.kafka.clients.admin.DeleteAclsResult; +import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResults; import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult; import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions; import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult; @@ -37,6 +39,7 @@ import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.acl.AccessControlEntry; @@ -1907,9 +1910,8 @@ public JSONArray getKafkaAclBYTopicName(String clusterAlias, String topicname) { return acls; } - - - + + public JSONArray _getACL(AdminClient adminClient, AclBindingFilter aclBindingFilter) throws InterruptedException, ExecutionException { JSONArray acls = new JSONArray(); java.util.Iterator itorA = adminClient.describeAcls(aclBindingFilter).values().get().iterator(); @@ -1935,5 +1937,55 @@ public JSONArray _getACL(AdminClient adminClient, AclBindingFilter aclBindingFil } return acls; } + + @Override + public String deleteAcl(String clusterAlias, JSONObject jsonObject) { + Properties prop = new Properties(); + + String reString = ""; + prop.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, parseBrokerServer(clusterAlias)); + + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.sasl.enable")) { + sasl(prop, clusterAlias); + } + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.ssl.enable")) { + ssl(prop, clusterAlias); + } + + AdminClient adminClient = null; + try { + adminClient = AdminClient.create(prop); + + ArrayList filters = new ArrayList(); + + JSONObject pattern = jsonObject.getJSONObject("pattern"); + ResourcePatternFilter rpFilter = new ResourcePatternFilter( + ResourceType.valueOf(pattern.getString("resourceType")), + pattern.getString("name"), + PatternType.valueOf(pattern.getString("patternType")) + ); + + JSONObject entry = jsonObject.getJSONObject("entry"); + AccessControlEntryFilter arFilter = new AccessControlEntryFilter( + entry.getString("principal"), + entry.getString("host"), + AclOperation.valueOf(entry.getString("operation")), + AclPermissionType.valueOf(entry.getString("permissionType")) + ); + + AclBindingFilter aclBindingFilter = new AclBindingFilter(rpFilter, arFilter); + filters.add(aclBindingFilter); + DeleteAclsResult result = adminClient.deleteAcls(filters); + KafkaFuture kf = result.values().get(aclBindingFilter); + reString = kf.toString();; + } catch (Exception e) { + LOG.error("Get kafka acl has error,msg is " + e.getMessage()); + reString = e.getMessage(); + } finally { + adminClient.close(); + } + + return reString; + } } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java index 821345e4..0dee90d5 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java @@ -29,12 +29,14 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.servlet.ModelAndView; import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; /** * Kafka acl controller to viewer data. @@ -137,5 +139,14 @@ public ModelAndView failedView(HttpSession session) { mav.setViewName("/acls/add_failed"); return mav; } + + + @RequestMapping(value = "/acls/delete", method = RequestMethod.POST) + @ResponseBody + public String delete(@RequestBody JSONObject ob, + HttpSession session) throws Exception { + String clusterAlias = session.getAttribute(KConstants.SessionAlias.CLUSTER_ALIAS).toString(); + return aclService.delete(clusterAlias, ob); + } } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java index 475ef8a8..32d0fd45 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/AclService.java @@ -20,6 +20,7 @@ import java.util.Map; import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; /** * Kafka acl data interface. @@ -37,5 +38,7 @@ public interface AclService { public JSONArray getTopicAcls(String clusterAlias, String topicname); Map createGroup(String clusterAlias, String userName, String groupName); + + String delete(String clusterAlias, JSONObject jsonObject); } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java index 78ba2c4e..fdfad84b 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/AclServiceImpl.java @@ -25,6 +25,7 @@ import org.springframework.stereotype.Service; import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; /** * Kafka acl data interface, and set up the return data set. @@ -57,12 +58,13 @@ public Map createGroup(String clusterAlias, String userName, Str return kafkaService.createGroup(clusterAlias, userName, "group_" + userName + "_" + groupName); } + @Override + public String delete(String clusterAlias, JSONObject jsonObject) { + return kafkaService.deleteAcl(clusterAlias, jsonObject); + } } - - - diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp index 421a3578..0c0f3b41 100644 --- a/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/acls/acls.jsp @@ -52,7 +52,7 @@
- List all Acls. + List all ACL.
@@ -72,6 +72,7 @@ host operation permissionType + Operate diff --git a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp index 2ff98e5a..939c2c28 100644 --- a/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp +++ b/kafka-eagle-web/src/main/webapp/WEB-INF/views/public/plus/navbar.jsp @@ -78,7 +78,7 @@ List CreateGroup + class="fas fa-edit fa-sm fa-fw mr-1">AddGroup
diff --git a/kafka-eagle-web/src/main/webapp/media/js/main/acl/acls.js b/kafka-eagle-web/src/main/webapp/media/js/main/acl/acls.js index 24011b2e..e33983d4 100644 --- a/kafka-eagle-web/src/main/webapp/media/js/main/acl/acls.js +++ b/kafka-eagle-web/src/main/webapp/media/js/main/acl/acls.js @@ -1,3 +1,30 @@ +function deleteM(data, that) { + if (confirm('确定要删除吗') != true) return; + + var row = JSON.parse(decodeURI(data)); + + var url = '/acls/delete'; + $(that).attr('disabled',"true"); + $.ajax({ + url: url, + type: 'POST', + processData: false, + contentType: "application/json", + data: JSON.stringify(row), + success: function(response) { + $(that).removeAttr("disabled"); + alert(response); + location.reload(); + }, + error: function(a,b) { + $(that).removeAttr("disabled"); + alert("失败！"); + return false; + } + }); +} + + $(document).ready(function() { try { function showTables(rows) { @@ -8,6 +35,16 @@ $(document).ready(function() { {"data": "entry.host" }, {"data": "entry.operation" }, {"data": "entry.permissionType" }, + { + "orderable": false, + mRender: function(data, type, row) { + if (type == 'display') { + var sJson = encodeURI(JSON.stringify(row)); + return '删除'; + } + return ""; + } + }, ]; var table = $('#result').DataTable({ From bc034550c2589dd77fa40e5dff38f6dadcc2bd7d Mon Sep 17 00:00:00 2001 From: jeff Date: Fri, 26 Feb 2021 18:03:52 +0800 Subject: [PATCH 12/12] =?UTF-8?q?=E5=8F=96ZK=E4=B8=AD=E4=BF=9D=E5=AD=98?= =?UTF-8?q?=E7=9A=84=E7=94=A8=E6=88=B7=E4=BF=A1=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../eagle/core/factory/v2/ConfigFactory.java | 34 + .../eagle/core/factory/v2/ConfigProvider.java | 30 + .../eagle/core/factory/v2/ConfigService.java | 92 +++ .../core/factory/v2/ConfigServiceImpl.java | 764 ++++++++++++++++++ .../eagle/web/controller/AclController.java | 14 +- .../web/service/impl/TopicServiceImpl.java | 3 + 6 files changed, 936 insertions(+), 1 deletion(-) create mode 100644 kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigFactory.java create mode 100644 kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigProvider.java create mode 100644 kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigService.java create mode 100644 kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigServiceImpl.java diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigFactory.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigFactory.java new file mode 100644 index 00000000..a1d2a23e --- /dev/null +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigFactory.java @@ -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. + */ +package org.smartloli.kafka.eagle.core.factory.v2; + +/** +* Topic factory implements {@link BrokerProvider} service. +* +* @author jeff. +* +* Created by Jun 13, 2019 +*/ +public class ConfigFactory implements ConfigProvider{ + + @Override + public ConfigService create() { + return new ConfigServiceImpl(); + } + +} diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigProvider.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigProvider.java new file mode 100644 index 00000000..07b60f22 --- /dev/null +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigProvider.java @@ -0,0 +1,30 @@ +/** + * 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.core.factory.v2; + +/** +* Topic provider {@link BrokerService} visitor enter. +* +* @author jeff. +* +*/ +public interface ConfigProvider { + + public ConfigService create(); + +} diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigService.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigService.java new file mode 100644 index 00000000..40494779 --- /dev/null +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigService.java @@ -0,0 +1,92 @@ +/** + * 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.core.factory.v2; + +import java.util.List; +import java.util.Map; + +import org.smartloli.kafka.eagle.common.protocol.MetadataInfo; +import org.smartloli.kafka.eagle.common.protocol.PartitionsInfo; + +/** + * Consumer group, topic and topic page or partition interface.Kafka api 2.x + * version. + * + * @author jeff. + * + */ +public interface ConfigService { + + /** Check topic from zookeeper metadata. */ + public boolean findKafkaTopic(String clusterAlias, String topic); + + /** Get topic list. */ + public List topicList(String clusterAlias); + + /** Get select topic list. */ + public String topicListParams(String clusterAlias, String search); + + /** Get kafka broker numbers. */ + public long brokerNumbers(String clusterAlias); + + /** Get topic number from zookeeper. */ + public long topicNumbers(String clusterAlias); + + /** Get topic number with match name from zookeeper. */ + public long topicNumbers(String clusterAlias, String topic); + + /** Get partition number from zookeeper. */ + public long partitionNumbers(String clusterAlias, String topic); + + /** Scan topic page display. */ + public List topicRecords(String clusterAlias, Map params); + + /** Scan topic meta page display. */ + public List topicMetadataRecords(String clusterAlias, String topic, Map params); + + /** Get topic producer logsize total. */ + public long getTopicLogSizeTotal(String clusterAlias, String topic); + + /** Get topic real logsize records. */ + public long getTopicRealLogSize(String clusterAlias, String topic); + + /** Get topic producer send logsize records. */ + public long getTopicProducerLogSize(String clusterAlias, String topic); + + /** Add topic partitions. */ + public Map createTopicPartitions(String clusterAlias, String topic, int totalCount); + + /** Get broker spread by topic. */ + public int getBrokerSpreadByTopic(String clusterAlias, String topic); + + /** Get broker skewed by topic. */ + public int getBrokerSkewedByTopic(String clusterAlias, String topic); + + /** Get broker leader skewed by topic. */ + public int getBrokerLeaderSkewedByTopic(String clusterAlias, String topic); + + /** Get broker id list. */ + public List getBrokerIdList(String clusterAlias); + + /** + * Get kafka broker numbers from zookeeper. + * @return + */ + long getUserList(String clusterAlias); + +} diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigServiceImpl.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigServiceImpl.java new file mode 100644 index 00000000..ccb5a5c8 --- /dev/null +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/ConfigServiceImpl.java @@ -0,0 +1,764 @@ +/** + * 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.core.factory.v2; + +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import com.alibaba.fastjson.TypeReference; +import com.google.common.base.Strings; +import kafka.zk.KafkaZkClient; +import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.NewPartitions; +import org.apache.zookeeper.data.Stat; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.smartloli.kafka.eagle.common.protocol.MetadataInfo; +import org.smartloli.kafka.eagle.common.protocol.PartitionsInfo; +import org.smartloli.kafka.eagle.common.util.*; +import org.smartloli.kafka.eagle.common.util.KConstants.Kafka; +import org.smartloli.kafka.eagle.core.factory.KafkaFactory; +import org.smartloli.kafka.eagle.core.factory.KafkaService; +import scala.Option; +import scala.Tuple2; +import scala.collection.JavaConversions; +import scala.collection.Seq; + +import java.util.*; +import java.util.Map.Entry; + +/** + * Implements {@link ConfigService} all method. + * + * @author jeff. + */ +public class ConfigServiceImpl implements ConfigService { + + private final String CONFIG_USERS_PATH = "/config/users"; + + + private final String BROKER_IDS_PATH = "/brokers/ids"; + private final String BROKER_TOPICS_PATH = "/brokers/topics"; + private final String TOPIC_ISR = "/brokers/topics/%s/partitions/%s/state"; + private final Logger LOG = LoggerFactory.getLogger(ConfigServiceImpl.class); + + + /** + * Get kafka broker numbers from zookeeper. + * @return + */ + @Override + public long getUserList(String clusterAlias) { + long count = 0; + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(CONFIG_USERS_PATH)) { + Seq path = zkc.getChildren(CONFIG_USERS_PATH); + List l = JavaConversions.seqAsJavaList(path); + count = l.size(); + } + } catch (Exception e) { + LOG.error("Get kafka broker numbers has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + + return count; + } + + + /** + * Instance Kafka Zookeeper client pool. + */ + private KafkaZKPoolUtils kafkaZKPool = KafkaZKPoolUtils.getInstance(); + + /** + * Kafka service interface. + */ + private KafkaService kafkaService = new KafkaFactory().create(); + + /** + * Statistics topic total used as page. + */ + public long topicNumbers(String clusterAlias) { + return topicList(clusterAlias).size(); + } + + /** + * Exclude kafka topic(__consumer_offsets) and blacklist topic. + */ + private void excludeTopic(String clusterAlias, List topics) { + if (topics.contains(Kafka.CONSUMER_OFFSET_TOPIC)) { + topics.remove(Kafka.CONSUMER_OFFSET_TOPIC); + } + String[] blackListTopics = SystemConfigUtils.getPropertyArray(clusterAlias + ".kafka.eagle.blacklist.topics", ","); + if (!StrUtils.isNull(blackListTopics)) { + for (String blackListTopic : blackListTopics) { + if (topics.contains(blackListTopic)) { + topics.remove(blackListTopic); + } + } + } + } + + /** + * Get search topic list numbers. + */ + public long topicNumbers(String clusterAlias, String topic) { + long count = 0L; + List topics = topicList(clusterAlias); + for (String name : topics) { + if (topic != null && name.contains(topic)) { + count++; + } + } + return count; + } + + /** + * Statistics topic partitions total used as page. + */ + public long partitionNumbers(String clusterAlias, String topic) { + long count = 0L; + if (Kafka.CONSUMER_OFFSET_TOPIC.equals(topic)) { + return count; + } + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic + "/partitions")) { + Seq subBrokerTopicsPaths = zkc.getChildren(BROKER_TOPICS_PATH + "/" + topic + "/partitions"); + count = JavaConversions.seqAsJavaList(subBrokerTopicsPaths).size(); + } + } catch (Exception e) { + LOG.error("Get topic partition numbers has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + return count; + } + + /** + * Get the number of page records for topic. + */ + public List topicRecords(String clusterAlias, Map params) { + List targets = new ArrayList(); + List topics = topicList(clusterAlias); + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + int start = Integer.parseInt(params.get("start").toString()); + int length = Integer.parseInt(params.get("length").toString()); + if (params.containsKey("search") && params.get("search").toString().length() > 0) { + String search = params.get("search").toString(); + int offset = 0; + int id = start + 1; + for (String topic : topics) { + if (search != null && topic.contains(search)) { + if (offset < (start + length) && offset >= start) { + try { + if (zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic)) { + Tuple2, Stat> tuple = zkc.getDataAndStat(BROKER_TOPICS_PATH + "/" + topic); + PartitionsInfo partition = new PartitionsInfo(); + partition.setId(id++); + partition.setCreated(CalendarUtils.convertUnixTime2Date(tuple._2.getCtime())); + partition.setModify(CalendarUtils.convertUnixTime2Date(tuple._2.getMtime())); + partition.setTopic(topic); + String tupleString = new String(tuple._1.get()); + JSONObject partitionObject = JSON.parseObject(tupleString).getJSONObject("partitions"); + partition.setPartitionNumbers(partitionObject.size()); + partition.setPartitions(partitionObject.keySet()); + targets.add(partition); + } + } catch (Exception ex) { + ex.printStackTrace(); + LOG.error("Scan topic search from zookeeper has error, msg is " + ex.getMessage()); + } + } + offset++; + } + } + } else { + int offset = 0; + int id = start + 1; + for (String topic : topics) { + if (offset < (start + length) && offset >= start) { + try { + if (zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic)) { + Tuple2, Stat> tuple = zkc.getDataAndStat(BROKER_TOPICS_PATH + "/" + topic); + PartitionsInfo partition = new PartitionsInfo(); + partition.setId(id++); + partition.setCreated(CalendarUtils.convertUnixTime2Date(tuple._2.getCtime())); + partition.setModify(CalendarUtils.convertUnixTime2Date(tuple._2.getMtime())); + partition.setTopic(topic); + String tupleString = new String(tuple._1.get()); + JSONObject partitionObject = JSON.parseObject(tupleString).getJSONObject("partitions"); + partition.setPartitionNumbers(partitionObject.size()); + partition.setPartitions(partitionObject.keySet()); + targets.add(partition); + } + } catch (Exception ex) { + ex.printStackTrace(); + LOG.error("Scan topic page from zookeeper has error, msg is " + ex.getMessage()); + } + } + offset++; + } + } + } catch (Exception e) { + LOG.error("Get topic records has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return targets; + } + + /** + * Get broker spread by topic. + */ + public int getBrokerSpreadByTopic(String clusterAlias, String topic) { + int spread = 0; + try { + List topicMetas = topicMetadata(clusterAlias, topic); + Set brokerSizes = new HashSet<>(); + for (MetadataInfo meta : topicMetas) { + List replicasIntegers = new ArrayList<>(); + try { + replicasIntegers = JSON.parseObject(meta.getReplicas(), new TypeReference>() { + }); + } catch (Exception e) { + e.printStackTrace(); + LOG.error("Parse string to int list has error, msg is " + e.getCause().getMessage()); + } + brokerSizes.addAll(replicasIntegers); + } + int brokerSize = kafkaService.getAllBrokersInfo(clusterAlias).size(); + spread = brokerSizes.size() * 100 / brokerSize; + } catch (Exception e) { + e.printStackTrace(); + LOG.error("Get topic skewed info has error, msg is ", e); + } + return spread; + } + + /** + * Get broker skewed by topic. + */ + public int getBrokerSkewedByTopic(String clusterAlias, String topic) { + int skewed = 0; + try { + List topicMetas = topicMetadata(clusterAlias, topic); + int partitionAndReplicaTopics = 0; + Set brokerSizes = new HashSet<>(); + Map brokers = new HashMap<>(); + for (MetadataInfo meta : topicMetas) { + List replicasIntegers = new ArrayList<>(); + try { + replicasIntegers = JSON.parseObject(meta.getReplicas(), new TypeReference>() { + }); + } catch (Exception e) { + e.printStackTrace(); + LOG.error("Parse string to int list has error, msg is " + e.getCause().getMessage()); + } + brokerSizes.addAll(replicasIntegers); + partitionAndReplicaTopics += replicasIntegers.size(); + for (Integer brokerId : replicasIntegers) { + if (brokers.containsKey(brokerId)) { + int value = brokers.get(brokerId); + brokers.put(brokerId, value + 1); + } else { + brokers.put(brokerId, 1); + } + } + } + int brokerSize = brokerSizes.size(); + int normalSkewedValue = MathUtils.ceil(brokerSize, partitionAndReplicaTopics); + int brokerSkewSize = 0; + for (Entry entry : brokers.entrySet()) { + if (entry.getValue() > normalSkewedValue) { + brokerSkewSize++; + } + } + skewed = brokerSkewSize * 100 / brokerSize; + } catch (Exception e) { + e.printStackTrace(); + LOG.error("Get topic skewed info has error, msg is ", e); + } + return skewed; + } + + /** + * Get broker skewed leader by topic. + */ + public int getBrokerLeaderSkewedByTopic(String clusterAlias, String topic) { + int leaderSkewed = 0; + try { + List topicMetas = topicMetadata(clusterAlias, topic); + Map brokerLeaders = new HashMap<>(); + Set brokerSizes = new HashSet<>(); + for (MetadataInfo meta : topicMetas) { + List replicasIntegers = new ArrayList<>(); + try { + replicasIntegers = JSON.parseObject(meta.getReplicas(), new TypeReference>() { + }); + } catch (Exception e) { + e.printStackTrace(); + LOG.error("Parse string to int list has error, msg is " + e.getCause().getMessage()); + } + brokerSizes.addAll(replicasIntegers); + if (brokerLeaders.containsKey(meta.getLeader())) { + int value = brokerLeaders.get(meta.getLeader()); + brokerLeaders.put(meta.getLeader(), value + 1); + } else { + brokerLeaders.put(meta.getLeader(), 1); + } + } + int brokerSize = brokerSizes.size(); + int brokerSkewLeaderNormal = MathUtils.ceil(brokerSize, topicMetas.size()); + int brokerSkewLeaderSize = 0; + for (Entry entry : brokerLeaders.entrySet()) { + if (entry.getValue() > brokerSkewLeaderNormal) { + brokerSkewLeaderSize++; + } + } + leaderSkewed = brokerSkewLeaderSize * 100 / brokerSize; + } catch (Exception e) { + e.printStackTrace(); + LOG.error("Get topic skewed info has error, msg is ", e); + } + return leaderSkewed; + } + + /** + * Check topic from zookeeper metadata. + */ + public boolean findKafkaTopic(String clusterAlias, String topic) { + return topicList(clusterAlias).contains(topic); + } + + /** + * Get kafka broker numbers from zookeeper. + */ + public long brokerNumbers(String clusterAlias) { + long count = 0; + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_IDS_PATH)) { + Seq subBrokerIdsPaths = zkc.getChildren(BROKER_IDS_PATH); + count = JavaConversions.seqAsJavaList(subBrokerIdsPaths).size(); + } + } catch (Exception e) { + LOG.error("Get kafka broker numbers has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return count; + } + + /** + * Get topic list include cgroups from zookeeper. + */ + public List topicList(String clusterAlias) { + List topics = new ArrayList<>(); + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.sasl.cgroup.enable")) { + topics = SystemConfigUtils.getPropertyArrayList(clusterAlias + ".kafka.eagle.sasl.cgroup.topics", ","); + } else if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.ssl.cgroup.enable")) { + topics = SystemConfigUtils.getPropertyArrayList(clusterAlias + ".kafka.eagle.ssl.cgroup.topics", ","); + } else { + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_TOPICS_PATH)) { + Seq subBrokerTopicsPaths = zkc.getChildren(BROKER_TOPICS_PATH); + topics = JavaConversions.seqAsJavaList(subBrokerTopicsPaths); + excludeTopic(clusterAlias, topics); + } + } catch (Exception e) { + ErrorUtils.print(this.getClass()).error("Get topic list has error, msg is ", e); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + } + return topics; + } + + /** + * Get select topic list from zookeeper. + */ + public String topicListParams(String clusterAlias, String search) { + JSONArray targets = new JSONArray(); + int limit = 15; + + List topics = topicList(clusterAlias); + try { + if (Strings.isNullOrEmpty(search)) { + int id = 1; + for (String topic : topics) { + if (id <= limit) { + JSONObject object = new JSONObject(); + object.put("id", id); + object.put("name", topic); + targets.add(object); + id++; + } + } + } else { + int id = 1; + for (String topic : topics) { + if (topic.contains(search)) { + if (id <= limit) { + JSONObject object = new JSONObject(); + object.put("id", id); + object.put("name", topic); + targets.add(object); + id++; + } + } + } + } + } catch (Exception e) { + LOG.error("Get topic list has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } + return targets.toJSONString(); + } + + /** + * Scan topic meta page display from zookeeper and kafka. + */ + public List topicMetadataRecords(String clusterAlias, String topic, Map params) { + List targets = new ArrayList<>(); + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_TOPICS_PATH)) { + List topics = topicList(clusterAlias); + if (topics.contains(topic)) { + int start = Integer.parseInt(params.get("start").toString()); + int length = Integer.parseInt(params.get("length").toString()); + int offset = 0; + Tuple2, Stat> tuple = zkc.getDataAndStat(BROKER_TOPICS_PATH + "/" + topic); + String tupleString = new String(tuple._1.get()); + JSONObject partitionObject = JSON.parseObject(tupleString).getJSONObject("partitions"); + Set partitionSet = new TreeSet<>(); + for (String partitionId : partitionObject.keySet()) { + partitionSet.add(Integer.valueOf(partitionId)); + } + Set partitionSortSet = new TreeSet<>(new Comparator() { + @Override + public int compare(Integer o1, Integer o2) { + int diff = o1 - o2;// asc + if (diff > 0) { + return 1; + } else if (diff < 0) { + return -1; + } + return 0; + } + }); + partitionSortSet.addAll(partitionSet); + for (int partition : partitionSortSet) { + if (offset < (start + length) && offset >= start) { + String path = String.format(TOPIC_ISR, topic, partition); + Tuple2, Stat> tuple2 = zkc.getDataAndStat(path); + String tupleString2 = new String(tuple2._1.get()); + JSONObject topicMetadata = JSON.parseObject(tupleString2); + MetadataInfo metadate = new MetadataInfo(); + metadate.setIsr(topicMetadata.getString("isr")); + metadate.setLeader(topicMetadata.getInteger("leader")); + metadate.setPartitionId(partition); + metadate.setReplicas(kafkaService.getReplicasIsr(clusterAlias, topic, partition)); + long logSize = 0L; + if ("kafka".equals(SystemConfigUtils.getProperty(clusterAlias + ".kafka.eagle.offset.storage"))) { + logSize = kafkaService.getKafkaRealLogSize(clusterAlias, topic, partition); + } else { + logSize = kafkaService.getRealLogSize(clusterAlias, topic, partition); + } + List isrIntegers = new ArrayList<>(); + List replicasIntegers = new ArrayList<>(); + try { + isrIntegers = JSON.parseObject(metadate.getIsr(), new TypeReference>() { + }); + replicasIntegers = JSON.parseObject(metadate.getReplicas(), new TypeReference>() { + }); + } catch (Exception e) { + e.printStackTrace(); + LOG.error("Parse string to int list has error, msg is ", e); + } + if (isrIntegers.size() != replicasIntegers.size()) { + // replicas lost + metadate.setUnderReplicated(true); + } else { + // replicas normal + metadate.setUnderReplicated(false); + } + if (replicasIntegers != null && replicasIntegers.size() > 0 && replicasIntegers.get(0) == metadate.getLeader()) { + // partition preferred leader + metadate.setPreferredLeader(true); + } else { + // partition occurs preferred leader exception + metadate.setPreferredLeader(false); + } + metadate.setLogSize(logSize); + targets.add(metadate); + } + offset++; + } + } + } + } catch (Exception e) { + LOG.error("Get topic metadata records has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return targets; + } + + private List topicMetadata(String clusterAlias, String topic) { + List targets = new ArrayList<>(); + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_TOPICS_PATH)) { + List topics = topicList(clusterAlias); + if (topics.contains(topic)) { + Tuple2, Stat> tuple = zkc.getDataAndStat(BROKER_TOPICS_PATH + "/" + topic); + String tupleString = new String(tuple._1.get()); + JSONObject partitionObject = JSON.parseObject(tupleString).getJSONObject("partitions"); + for (String partition : partitionObject.keySet()) { + String path = String.format(TOPIC_ISR, topic, Integer.valueOf(partition)); + Tuple2, Stat> tuple2 = zkc.getDataAndStat(path); + String tupleString2 = new String(tuple2._1.get()); + JSONObject topicMetadata = JSON.parseObject(tupleString2); + MetadataInfo metadate = new MetadataInfo(); + metadate.setIsr(topicMetadata.getString("isr")); + metadate.setLeader(topicMetadata.getInteger("leader")); + metadate.setPartitionId(Integer.valueOf(partition)); + metadate.setReplicas(kafkaService.getReplicasIsr(clusterAlias, topic, Integer.valueOf(partition))); + targets.add(metadate); + } + } + } + } catch (Exception e) { + LOG.error("Get topic metadata records has error, msg is ", e); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return targets; + } + + /** + * Get topic producer logsize total. + */ + public long getTopicLogSizeTotal(String clusterAlias, String topic) { + long logSize = 0L; + if (Kafka.CONSUMER_OFFSET_TOPIC.equals(topic)) { + return logSize; + } + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic)) { + Tuple2, Stat> tuple = zkc.getDataAndStat(BROKER_TOPICS_PATH + "/" + topic); + String tupleString = new String(tuple._1.get()); + JSONObject partitionObject = JSON.parseObject(tupleString).getJSONObject("partitions"); + Set partitions = new HashSet<>(); + for (String partition : partitionObject.keySet()) { + try { + partitions.add(Integer.valueOf(partition)); + } catch (Exception e) { + LOG.error("Convert partition string to integer has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } + } + if ("kafka".equals(SystemConfigUtils.getProperty(clusterAlias + ".kafka.eagle.offset.storage"))) { + logSize = kafkaService.getKafkaRealLogSize(clusterAlias, topic, partitions); + } else { + logSize = kafkaService.getLogSize(clusterAlias, topic, partitions); + } + } + } catch (Exception e) { + LOG.error("Get topic logsize total has error, msg is ", e); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return logSize; + } + + /** + * Get topic real logsize records. + */ + public long getTopicRealLogSize(String clusterAlias, String topic) { + long logSize = 0L; + if (Kafka.CONSUMER_OFFSET_TOPIC.equals(topic)) { + return logSize; + } + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic)) { + Tuple2, Stat> tuple = zkc.getDataAndStat(BROKER_TOPICS_PATH + "/" + topic); + String tupleString = new String(tuple._1.get()); + JSONObject partitionObject = JSON.parseObject(tupleString).getJSONObject("partitions"); + Set partitions = new HashSet<>(); + for (String partition : partitionObject.keySet()) { + try { + partitions.add(Integer.valueOf(partition)); + } catch (Exception e) { + LOG.error("Convert partition string to integer has error, msg is ", e); + } + } + if ("kafka".equals(SystemConfigUtils.getProperty(clusterAlias + ".kafka.eagle.offset.storage"))) { + logSize = kafkaService.getKafkaRealLogSize(clusterAlias, topic, partitions); + } else { + logSize = kafkaService.getRealLogSize(clusterAlias, topic, partitions); + } + } + } catch (Exception e) { + ErrorUtils.print(this.getClass()).error("Get topic real logsize has error, msg is ", e); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return logSize; + } + + /** + * Get topic producer send logsize records. + */ + public long getTopicProducerLogSize(String clusterAlias, String topic) { + long logSize = 0L; + if (Kafka.CONSUMER_OFFSET_TOPIC.equals(topic)) { + return logSize; + } + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic)) { + Tuple2, Stat> tuple = zkc.getDataAndStat(BROKER_TOPICS_PATH + "/" + topic); + String tupleString = new String(tuple._1.get()); + JSONObject partitionObject = JSON.parseObject(tupleString).getJSONObject("partitions"); + Set partitions = new HashSet<>(); + for (String partition : partitionObject.keySet()) { + try { + partitions.add(Integer.valueOf(partition)); + } catch (Exception e) { + LOG.error("Convert partition string to integer has error, msg is " + e.getCause().getMessage()); + e.printStackTrace(); + } + } + if ("kafka".equals(SystemConfigUtils.getProperty(clusterAlias + ".kafka.eagle.offset.storage"))) { + logSize = kafkaService.getKafkaProducerLogSize(clusterAlias, topic, partitions); + } else { + logSize = kafkaService.getLogSize(clusterAlias, topic, partitions); + } + } + } catch (Exception e) { + LOG.error("Get topic real logsize has error, msg is ", e); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return logSize; + } + + /** + * Add topic partitions. + */ + public Map createTopicPartitions(String clusterAlias, String topic, int totalCount) { + Map targets = new HashMap(); + int existPartitions = (int) partitionNumbers(clusterAlias, topic); + Properties prop = new Properties(); + prop.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafkaService.getKafkaBrokerServer(clusterAlias)); + + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.sasl.enable")) { + kafkaService.sasl(prop, clusterAlias); + } + if (SystemConfigUtils.getBooleanProperty(clusterAlias + ".kafka.eagle.ssl.enable")) { + kafkaService.ssl(prop, clusterAlias); + } + + AdminClient adminClient = null; + try { + adminClient = AdminClient.create(prop); + Map newPartitions = new HashMap(); + newPartitions.put(topic, NewPartitions.increaseTo(existPartitions + totalCount)); + adminClient.createPartitions(newPartitions); + targets.put("status", "success"); + targets.put("info", "Add topic[" + topic + "], before partition[" + existPartitions + "], after partition[" + (existPartitions + totalCount) + "] has successed."); + } catch (Exception e) { + LOG.info("Add kafka topic partitions has error, msg is " + e.getMessage()); + e.printStackTrace(); + targets.put("status", "failed"); + targets.put("info", "Add kafka topic partitions has error, msg is " + e.getMessage()); + } finally { + adminClient.close(); + } + return targets; + } + + /** + * Get broker id list. + */ + public List getBrokerIdList(String clusterAlias) { + List brokerIds = new ArrayList<>(); + KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); + try { + if (zkc.pathExists(BROKER_IDS_PATH)) { + Seq subBrokerIdsPaths = zkc.getChildren(BROKER_IDS_PATH); + for (String id : JavaConversions.seqAsJavaList(subBrokerIdsPaths)) { + brokerIds.add(Integer.parseInt(id)); + } + } + } catch (Exception e) { + LOG.error("Get kafka broker id has error, msg is ", e); + } finally { + if (zkc != null) { + kafkaZKPool.release(clusterAlias, zkc); + zkc = null; + } + } + return brokerIds; + } + +} diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java index 0dee90d5..f57004e0 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/controller/AclController.java @@ -25,6 +25,8 @@ import org.apache.shiro.authz.annotation.RequiresPermissions; import org.smartloli.kafka.eagle.common.util.KConstants; +import org.smartloli.kafka.eagle.core.factory.v2.ConfigFactory; +import org.smartloli.kafka.eagle.core.factory.v2.ConfigService; import org.smartloli.kafka.eagle.web.service.AclService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; @@ -50,11 +52,21 @@ public class AclController { /** Kafka acl service interface. */ @Autowired private AclService aclService; + + + /** + * Broker service interface. + */ + private static ConfigService configService = new ConfigFactory().create(); + + /** acls viewer. */ @RequestMapping(value = "/acls", method = RequestMethod.GET) public ModelAndView aclsView(HttpSession session) { ModelAndView mav = new ModelAndView(); + + mav.setViewName("/acls/acls"); return mav; } @@ -67,7 +79,7 @@ public JSONArray aclslist(HttpServletResponse response, HttpServletRequest reque HttpSession session = request.getSession(); String clusterAlias = session.getAttribute(KConstants.SessionAlias.CLUSTER_ALIAS).toString(); JSONArray result = aclService.getAcls(clusterAlias); - + configService.getUserList(clusterAlias); return result; } diff --git a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/TopicServiceImpl.java b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/TopicServiceImpl.java index f9e9f1c5..3503ddc8 100644 --- a/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/TopicServiceImpl.java +++ b/kafka-eagle-web/src/main/java/org/smartloli/kafka/eagle/web/service/impl/TopicServiceImpl.java @@ -45,6 +45,8 @@ import org.smartloli.kafka.eagle.core.factory.hub.KafkaHubService; import org.smartloli.kafka.eagle.core.factory.v2.BrokerFactory; import org.smartloli.kafka.eagle.core.factory.v2.BrokerService; +import org.smartloli.kafka.eagle.core.factory.v2.ConfigFactory; +import org.smartloli.kafka.eagle.core.factory.v2.ConfigService; import org.smartloli.kafka.eagle.core.metrics.KafkaMetricsFactory; import org.smartloli.kafka.eagle.core.metrics.KafkaMetricsService; import org.smartloli.kafka.eagle.core.sql.execute.KafkaSqlParser; @@ -88,6 +90,7 @@ public class TopicServiceImpl implements TopicService { * Broker service interface. */ private static BrokerService brokerService = new BrokerFactory().create(); + /** * Mx4j service interface.