Repository: skiptests/astraea
Pull request #520 (2022-07-28 16:17:35)

Refactor host metrics
as title


Patch:
diff --git a/app/src/main/java/org/astraea/app/common/Utils.java b/app/src/main/java/org/astraea/app/common/Utils.java
index 22d9af55..c8f0fe3b 100644
--- a/app/src/main/java/org/astraea/app/common/Utils.java
+++ b/app/src/main/java/org/astraea/app/common/Utils.java
@@ -23,6 +23,7 @@ import java.nio.file.Files;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
+import java.util.Objects;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.CompletableFuture;
@@ -169,6 +170,19 @@ public final class Utils {
     return value;
   }
 
+  /**
+   * check the content of string
+   *
+   * @param value to check
+   * @return input string if the string is not empty. Otherwise, it throws NPE or
+   *     IllegalArgumentException
+   */
+  public static String requireNonEmpty(String value) {
+    if (Objects.requireNonNull(value).isEmpty())
+      throw new IllegalArgumentException("the value: " + value + " can't be empty");
+    return value;
+  }
+
   /**
    * Check if the time is expired.
    *
diff --git a/app/src/main/java/org/astraea/app/cost/CpuCost.java b/app/src/main/java/org/astraea/app/cost/CpuCost.java
index 0a1023de..dc2e1aa9 100644
--- a/app/src/main/java/org/astraea/app/cost/CpuCost.java
+++ b/app/src/main/java/org/astraea/app/cost/CpuCost.java
@@ -24,8 +24,8 @@ import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
+import org.astraea.app.metrics.platform.HostMetrics;
 import org.astraea.app.metrics.platform.OperatingSystemInfo;
 
 /**
@@ -57,6 +57,6 @@ public class CpuCost implements HasBrokerCost {
 
   @Override
   public Optional<Fetcher> fetcher() {
-    return Optional.of(client -> List.of(KafkaMetrics.Host.operatingSystem(client)));
+    return Optional.of(client -> List.of(HostMetrics.operatingSystem(client)));
   }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/MemoryCost.java b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
index 08bab078..d4d1df2c 100644
--- a/app/src/main/java/org/astraea/app/cost/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
@@ -23,9 +23,9 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.platform.HasJvmMemory;
+import org.astraea.app.metrics.platform.HostMetrics;
 
 public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
 
@@ -57,6 +57,6 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
 
   @Override
   public Optional<Fetcher> fetcher() {
-    return Optional.of(client -> List.of(KafkaMetrics.Host.jvmMemory(client)));
+    return Optional.of(client -> List.of(HostMetrics.jvmMemory(client)));
   }
 }
diff --git a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
index 18ef641b..ba3dc573 100644
--- a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
@@ -24,8 +24,6 @@ import java.util.stream.Collectors;
 import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
 import org.astraea.app.metrics.broker.HasValue;
 import org.astraea.app.metrics.broker.TotalTimeMs;
-import org.astraea.app.metrics.platform.JvmMemory;
-import org.astraea.app.metrics.platform.OperatingSystemInfo;
 import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
 import org.astraea.app.metrics.producer.HasProducerTopicMetrics;
 
@@ -212,33 +210,6 @@ public final class KafkaMetrics {
     }
   }
 
-  public static final class Host {
-
-    private Host() {}
-
-    /**
-     * retrieve broker's operating system info
-     *
-     * @param mBeanClient a {@link MBeanClient} instance connect to specific kafka broker
-     * @return a {@link OperatingSystemInfo} describe broker's os info (arch, processors, memory,
-     *     cpu loading ...)
-     */
-    public static OperatingSystemInfo operatingSystem(MBeanClient mBeanClient) {
-      return new OperatingSystemInfo(
-          mBeanClient.queryBean(
-              BeanQuery.builder()
-                  .domainName("java.lang")
-                  .property("type", "OperatingSystem")
-                  .build()));
-    }
-
-    public static JvmMemory jvmMemory(MBeanClient mBeanClient) {
-      return new JvmMemory(
-          mBeanClient.queryBean(
-              BeanQuery.builder().domainName("java.lang").property("type", "Memory").build()));
-    }
-  }
-
   public static final class Producer {
 
     private Producer() {}
diff --git a/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java b/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java
index 6ea754f1..4b36fa34 100644
--- a/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java
@@ -20,7 +20,6 @@ import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -31,7 +30,6 @@ import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiFunction;
 import org.astraea.app.common.Utils;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.MBeanClient;
 
 public class BeanCollector {
@@ -88,7 +86,7 @@ public class BeanCollector {
       private boolean local = false;
       private String host;
       private int port = -1;
-      private Fetcher fetcher = client -> List.of(KafkaMetrics.Host.jvmMemory(client));
+      private Fetcher fetcher;
 
       @Override
       public Register host(String host) {
@@ -118,7 +116,12 @@ public class BeanCollector {
 
       @Override
       public Receiver build() {
-        var nodeKey = host + ":" + port;
+        if (!local) {
+          Utils.requirePositive(port);
+          Utils.requireNonEmpty(host);
+        }
+        Objects.requireNonNull(fetcher);
+        var nodeKey = local ? "local" : host + ":" + port;
         var node = nodes.computeIfAbsent(nodeKey, ignored -> new Node());
         var receiver =
             new Receiver() {
diff --git a/app/src/main/java/org/astraea/app/metrics/platform/HostMetrics.java b/app/src/main/java/org/astraea/app/metrics/platform/HostMetrics.java
new file mode 100644
index 00000000..1b7937fa
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/metrics/platform/HostMetrics.java
@@ -0,0 +1,40 @@
+/*
+ * 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.metrics.platform;
+
+import org.astraea.app.metrics.BeanQuery;
+import org.astraea.app.metrics.MBeanClient;
+
+public final class HostMetrics {
+
+  public static OperatingSystemInfo operatingSystem(MBeanClient mBeanClient) {
+    return new OperatingSystemInfo(
+        mBeanClient.queryBean(
+            BeanQuery.builder()
+                .domainName("java.lang")
+                .property("type", "OperatingSystem")
+                .build()));
+  }
+
+  public static JvmMemory jvmMemory(MBeanClient mBeanClient) {
+    return new JvmMemory(
+        mBeanClient.queryBean(
+            BeanQuery.builder().domainName("java.lang").property("type", "Memory").build()));
+  }
+
+  private HostMetrics() {}
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/app/common/UtilsTest.java b/app/src/test/java/org/astraea/app/common/UtilsTest.java
index 8b907948..9ff5b886 100644
--- a/app/src/test/java/org/astraea/app/common/UtilsTest.java
+++ b/app/src/test/java/org/astraea/app/common/UtilsTest.java
@@ -76,4 +76,10 @@ public class UtilsTest {
 
     Assertions.assertEquals(Utils.sequence(List.of(future1, future2)).join(), List.of(1, 2));
   }
+
+  @Test
+  void testNonEmpty() {
+    Assertions.assertThrows(IllegalArgumentException.class, () -> Utils.requireNonEmpty(""));
+    Assertions.assertThrows(NullPointerException.class, () -> Utils.requireNonEmpty(null));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
index cc16cf33..a75fc346 100644
--- a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
@@ -21,7 +21,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.condition.OS.LINUX;
-import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
@@ -40,14 +39,11 @@ import org.astraea.app.common.Utils;
 import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
 import org.astraea.app.metrics.broker.LogMetrics;
 import org.astraea.app.metrics.broker.TotalTimeMs;
-import org.astraea.app.metrics.platform.JvmMemory;
-import org.astraea.app.metrics.platform.OperatingSystemInfo;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledOnOs;
 import org.junit.jupiter.api.condition.EnabledOnOs;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
@@ -152,52 +148,6 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     assertThrows(IllegalArgumentException.class, () -> KafkaMetrics.BrokerTopic.of("nothing"));
   }
 
-  @Test
-  void operatingSystem() {
-    OperatingSystemInfo operatingSystemInfo = KafkaMetrics.Host.operatingSystem(mBeanClient);
-    assertDoesNotThrow(operatingSystemInfo::arch);
-    assertDoesNotThrow(operatingSystemInfo::availableProcessors);
-    assertDoesNotThrow(operatingSystemInfo::committedVirtualMemorySize);
-    assertDoesNotThrow(operatingSystemInfo::freePhysicalMemorySize);
-    assertDoesNotThrow(operatingSystemInfo::freeSwapSpaceSize);
-    assertDoesNotThrow(operatingSystemInfo::name);
-    assertDoesNotThrow(operatingSystemInfo::processCpuLoad);
-    assertDoesNotThrow(operatingSystemInfo::processCpuTime);
-    assertDoesNotThrow(operatingSystemInfo::systemCpuLoad);
-    assertDoesNotThrow(operatingSystemInfo::systemLoadAverage);
-    assertDoesNotThrow(operatingSystemInfo::totalPhysicalMemorySize);
-    assertDoesNotThrow(operatingSystemInfo::totalSwapSpaceSize);
-    assertDoesNotThrow(operatingSystemInfo::version);
-  }
-
-  @Test
-  @DisabledOnOs(WINDOWS)
-  void maxFileDescriptorCount() {
-    OperatingSystemInfo operatingSystemInfo = KafkaMetrics.Host.operatingSystem(mBeanClient);
-    assertDoesNotThrow(operatingSystemInfo::maxFileDescriptorCount);
-  }
-
-  @Test
-  @DisabledOnOs(WINDOWS)
-  void openFileDescriptorCount() {
-    OperatingSystemInfo operatingSystemInfo = KafkaMetrics.Host.operatingSystem(mBeanClient);
-    assertDoesNotThrow(operatingSystemInfo::openFileDescriptorCount);
-  }
-
-  @Test
-  void testJvmMemory() {
-    JvmMemory jvmMemory = KafkaMetrics.Host.jvmMemory(mBeanClient);
-
-    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getCommitted());
-    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getMax());
-    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getUsed());
-    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getInit());
-    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getCommitted());
-    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getMax());
-    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getUsed());
-    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getInit());
-  }
-
   @Test
   @EnabledOnOs(LINUX)
   void linuxDiskReadBytes() {
diff --git a/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java b/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
index e82bfc50..58013917 100644
--- a/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
@@ -27,8 +27,8 @@ import java.util.stream.IntStream;
 import org.astraea.app.common.Utils;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.metrics.platform.HostMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -286,7 +286,7 @@ public class BeanCollectorTest {
         collector
             .register()
             .local()
-            .fetcher(client -> List.of(KafkaMetrics.Host.jvmMemory(client)))
+            .fetcher(client -> List.of(HostMetrics.jvmMemory(client)))
             .build()) {
 
       // wait for updating cache
@@ -294,4 +294,19 @@ public class BeanCollectorTest {
       Assertions.assertNotEquals(0, receiver.current().size());
     }
   }
+
+  @Test
+  void testIllegalArgument() {
+    var collector =
+        BeanCollector.builder()
+            .interval(Duration.ofSeconds(1))
+            .clientCreator(clientCreator)
+            .build();
+    Assertions.assertThrows(
+        IllegalArgumentException.class, () -> collector.register().host("aaa").build());
+    Assertions.assertThrows(
+        NullPointerException.class, () -> collector.register().port(111).build());
+    Assertions.assertThrows(
+        NullPointerException.class, () -> collector.register().host("aaa").port(111).build());
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/metrics/platform/HostMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/platform/HostMetricsTest.java
new file mode 100644
index 00000000..106d1057
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/platform/HostMetricsTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.metrics.platform;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.condition.OS.WINDOWS;
+
+import org.astraea.app.metrics.MBeanClient;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+
+public class HostMetricsTest {
+
+  @Test
+  void operatingSystem() {
+    var operatingSystemInfo = HostMetrics.operatingSystem(MBeanClient.local());
+    assertDoesNotThrow(operatingSystemInfo::arch);
+    assertDoesNotThrow(operatingSystemInfo::availableProcessors);
+    assertDoesNotThrow(operatingSystemInfo::committedVirtualMemorySize);
+    assertDoesNotThrow(operatingSystemInfo::freePhysicalMemorySize);
+    assertDoesNotThrow(operatingSystemInfo::freeSwapSpaceSize);
+    assertDoesNotThrow(operatingSystemInfo::name);
+    assertDoesNotThrow(operatingSystemInfo::processCpuLoad);
+    assertDoesNotThrow(operatingSystemInfo::processCpuTime);
+    assertDoesNotThrow(operatingSystemInfo::systemCpuLoad);
+    assertDoesNotThrow(operatingSystemInfo::systemLoadAverage);
+    assertDoesNotThrow(operatingSystemInfo::totalPhysicalMemorySize);
+    assertDoesNotThrow(operatingSystemInfo::totalSwapSpaceSize);
+    assertDoesNotThrow(operatingSystemInfo::version);
+  }
+
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void maxFileDescriptorCount() {
+    OperatingSystemInfo operatingSystemInfo = HostMetrics.operatingSystem(MBeanClient.local());
+    assertDoesNotThrow(operatingSystemInfo::maxFileDescriptorCount);
+  }
+
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void openFileDescriptorCount() {
+    OperatingSystemInfo operatingSystemInfo = HostMetrics.operatingSystem(MBeanClient.local());
+    assertDoesNotThrow(operatingSystemInfo::openFileDescriptorCount);
+  }
+
+  @Test
+  void testJvmMemory() {
+    JvmMemory jvmMemory = HostMetrics.jvmMemory(MBeanClient.local());
+    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getCommitted());
+    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getMax());
+    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getUsed());
+    assertDoesNotThrow(() -> jvmMemory.heapMemoryUsage().getInit());
+    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getCommitted());
+    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getMax());
+    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getUsed());
+    assertDoesNotThrow(() -> jvmMemory.nonHeapMemoryUsage().getInit());
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/service/RequireJmxServerTest.java b/app/src/test/java/org/astraea/app/service/RequireJmxServerTest.java
index b63afb5c..468063a0 100644
--- a/app/src/test/java/org/astraea/app/service/RequireJmxServerTest.java
+++ b/app/src/test/java/org/astraea/app/service/RequireJmxServerTest.java
@@ -18,8 +18,8 @@ package org.astraea.app.service;
 
 import java.util.Set;
 import org.astraea.app.metrics.BeanQuery;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.metrics.platform.HostMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -46,7 +46,7 @@ public class RequireJmxServerTest extends RequireBrokerCluster {
 
   private void testMemory(MBeanClient client) {
     try (client) {
-      var memory = KafkaMetrics.Host.jvmMemory(client);
+      var memory = HostMetrics.jvmMemory(client);
       Assertions.assertNotEquals(0, memory.heapMemoryUsage().getMax());
     }
   }