Repository: skiptests/astraea
Pull request #417 (2022-06-22 15:56:31)

Web APIs for all beans
一個簡單的程式碼來暴露beans，後續會再補上測試和更多query


Patch:
diff --git a/app/src/main/java/org/astraea/app/web/BeansHandler.java b/app/src/main/java/org/astraea/app/web/BeansHandler.java
new file mode 100644
index 00000000..62546792
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/web/BeansHandler.java
@@ -0,0 +1,114 @@
+/*
+ * 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.astraea.app.web;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.metrics.jmx.BeanObject;
+import org.astraea.app.metrics.jmx.BeanQuery;
+import org.astraea.app.metrics.jmx.MBeanClient;
+
+public class BeansHandler implements Handler {
+  static final String DOMAIN_NAME_KEY = "domain";
+  private final List<MBeanClient> clients;
+
+  BeansHandler(Admin admin, Function<String, Integer> jmxPorts) {
+    clients =
+        admin.nodes().stream()
+            .map(n -> MBeanClient.jndi(n.host(), jmxPorts.apply(n.host())))
+            .collect(Collectors.toUnmodifiableList());
+  }
+
+  @Override
+  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+    return new NodeBeans(
+        clients.stream()
+            .map(
+                c ->
+                    new NodeBean(
+                        c.host(),
+                        c
+                            .queryBeans(
+                                BeanQuery.builder(queries.getOrDefault(DOMAIN_NAME_KEY, "*"))
+                                    .usePropertyListPattern()
+                                    .build())
+                            .stream()
+                            .map(Bean::new)
+                            .collect(Collectors.toUnmodifiableList())))
+            .collect(Collectors.toUnmodifiableList()));
+  }
+
+  static class Property implements JsonObject {
+    final String key;
+    final String value;
+
+    Property(String key, String value) {
+      this.key = key;
+      this.value = value;
+    }
+  }
+
+  static class Attribute implements JsonObject {
+    final String key;
+    final String value;
+
+    Attribute(String key, String value) {
+      this.key = key;
+      this.value = value;
+    }
+  }
+
+  static class Bean implements JsonObject {
+    final String domainName;
+    final List<Property> properties;
+    final List<Attribute> attributes;
+
+    Bean(BeanObject obj) {
+      this.domainName = obj.domainName();
+      this.properties =
+          obj.getProperties().entrySet().stream()
+              .map(e -> new Property(e.getKey(), e.getValue()))
+              .collect(Collectors.toUnmodifiableList());
+      this.attributes =
+          obj.getAttributes().entrySet().stream()
+              .map(e -> new Attribute(e.getKey(), e.getValue().toString()))
+              .collect(Collectors.toUnmodifiableList());
+    }
+  }
+
+  static class NodeBean implements JsonObject {
+    final String host;
+    final List<Bean> beans;
+
+    NodeBean(String host, List<Bean> beans) {
+      this.host = host;
+      this.beans = beans;
+    }
+  }
+
+  static class NodeBeans implements JsonObject {
+    final List<NodeBean> nodeBeans;
+
+    NodeBeans(List<NodeBean> nodeBeans) {
+      this.nodeBeans = nodeBeans;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/web/WebService.java b/app/src/main/java/org/astraea/app/web/WebService.java
index 8f3004b0..977489c6 100644
--- a/app/src/main/java/org/astraea/app/web/WebService.java
+++ b/app/src/main/java/org/astraea/app/web/WebService.java
@@ -20,8 +20,13 @@ import com.beust.jcommander.Parameter;
 import com.sun.net.httpserver.HttpServer;
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.function.Function;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.argument.NonNegativeIntegerField;
+import org.astraea.app.argument.StringMapField;
 
 public class WebService {
 
@@ -38,6 +43,8 @@ public class WebService {
     server.createContext("/quotas", new QuotaHandler(Admin.of(arg.configs())));
     server.createContext("/pipelines", new PipelineHandler(Admin.of(arg.configs())));
     server.createContext("/transactions", new TransactionHandler(Admin.of(arg.configs())));
+    if (arg.needJmx())
+      server.createContext("/beans", new BeansHandler(Admin.of(arg.configs()), arg.jmxPorts()));
     server.start();
   }
 
@@ -48,5 +55,31 @@ public class WebService {
         validateWith = NonNegativeIntegerField.class,
         converter = NonNegativeIntegerField.class)
     int port = 8001;
+
+    @Parameter(
+        names = {"--jmx.port"},
+        description = "Integer: the port to query JMX for each server",
+        validateWith = NonNegativeIntegerField.class,
+        converter = NonNegativeIntegerField.class)
+    int jmxPort = -1;
+
+    @Parameter(
+        names = {"--jmx.ports"},
+        description = "Map: the jmx port for each node. For example: 192.168.50.2=19999",
+        validateWith = StringMapField.class,
+        converter = StringMapField.class)
+    Map<String, String> jmxPorts = Map.of();
+
+    boolean needJmx() {
+      return jmxPort > 0 || !jmxPorts.isEmpty();
+    }
+
+    Function<String, Integer> jmxPorts() {
+      return name ->
+          Optional.of(jmxPorts.getOrDefault(name, String.valueOf(jmxPort)))
+              .map(Integer::valueOf)
+              .filter(i -> i > 0)
+              .orElseThrow(() -> new NoSuchElementException(name + " has no jmx port"));
+    }
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java b/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
new file mode 100644
index 00000000..4f04b9db
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.astraea.app.web;
+
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.common.Utils;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class BeanHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testArgument() {
+    var arg = new WebService.Argument();
+    Assertions.assertFalse(arg.needJmx());
+    arg.jmxPort = 1000;
+    Assertions.assertTrue(arg.needJmx());
+    arg.jmxPort = -1;
+    arg.jmxPorts = Map.of("a", "100");
+    Assertions.assertTrue(arg.needJmx());
+    Assertions.assertEquals(100, arg.jmxPorts().apply("a"));
+    Assertions.assertThrows(NoSuchElementException.class, () -> arg.jmxPorts().apply("b"));
+
+    arg.jmxPort = 999;
+    Assertions.assertEquals(999, arg.jmxPorts().apply("b"));
+  }
+
+  @Test
+  void testBeans() throws InterruptedException {
+    var topic = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new BeansHandler(admin, name -> jmxServiceURL().getPort());
+      var response =
+          Assertions.assertInstanceOf(
+              BeansHandler.NodeBeans.class, handler.get(Optional.empty(), Map.of()));
+      Assertions.assertNotEquals(0, response.nodeBeans.size());
+    }
+  }
+}