[METRICS] Append worker metrics

common/src/main/java/org/astraea/common/metrics/client/HasCoordinatorMetrics.java

@@ -0,0 +1,107 @@
+/*
+ * 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.common.metrics.client;
+
+import org.astraea.common.metrics.HasBeanObject;
+
+@FunctionalInterface
+public interface HasCoordinatorMetrics extends HasBeanObject {
+
+  default String clientId() {
+    return beanObject().properties().get("client-id");
+  }
+
+  default double failedRebalanceRatePerHour() {
+    return (double) beanObject().attributes().get("failed-rebalance-rate-per-hour");
+  }
+
+  default double failedRebalanceTotal() {
+    return (double) beanObject().attributes().get("failed-rebalance-total");
+  }
+
+  default double heartbeatRate() {
+    return (double) beanObject().attributes().get("heartbeat-rate");
+  }
+
+  default double heartbeatResponseTimeMax() {
+    return (double) beanObject().attributes().get("heartbeat-response-time-max");
+  }
+
+  default double heartbeatTotal() {
+    return (double) beanObject().attributes().get("heartbeat-total");
+  }
+
+  default double joinRate() {
+    return (double) beanObject().attributes().get("join-rate");
+  }
+
+  default double joinTimeAvg() {
+    return (double) beanObject().attributes().get("join-time-avg");
+  }
+
+  default double joinTimeMax() {
+    return (double) beanObject().attributes().get("join-time-max");
+  }
+
+  default double joinTotal() {
+    return (double) beanObject().attributes().get("join-total");
+  }
+
+  default double lastHeartbeatSecondsAgo() {
+    return (double) beanObject().attributes().get("last-heartbeat-seconds-ago");
+  }
+
+  default double lastRebalanceSecondsAgo() {
+    return (double) beanObject().attributes().get("last-rebalance-seconds-ago");
+  }
+
+  default double rebalanceLatencyAvg() {
+    return (double) beanObject().attributes().get("rebalance-latency-avg");
+  }
+
+  default double rebalanceLatencyMax() {
+    return (double) beanObject().attributes().get("rebalance-latency-max");
+  }
+
+  default double rebalanceLatencyTotal() {
+    return (double) beanObject().attributes().get("rebalance-latency-total");
+  }
+
+  default double rebalanceRatePerHour() {
+    return (double) beanObject().attributes().get("rebalance-rate-per-hour");
+  }
+
+  default double rebalanceTotal() {
+    return (double) beanObject().attributes().get("rebalance-total");
+  }
+
+  default double syncRate() {
+    return (double) beanObject().attributes().get("sync-rate");
+  }
+
+  default double syncTimeAvg() {
+    return (double) beanObject().attributes().get("sync-time-avg");
+  }
+
+  default double syncTimeMax() {
+    return (double) beanObject().attributes().get("sync-time-max");
+  }
+
+  default double syncTotal() {
+    return (double) beanObject().attributes().get("sync-total");
+  }
+}

common/src/main/java/org/astraea/common/metrics/client/HasNodeMetrics.java

@@ -19,6 +19,7 @@
 import java.util.function.Function;
 import org.astraea.common.metrics.HasBeanObject;
 
+@FunctionalInterface
 public interface HasNodeMetrics extends HasBeanObject {
   Function<String, Integer> BROKER_ID_FETCHER =
       node -> Integer.parseInt(node.substring(node.indexOf("-") + 1));

common/src/main/java/org/astraea/common/metrics/client/admin/HasAdminMetrics.java → common/src/main/java/org/astraea/common/metrics/client/HasSelectorMetrics.java

@@ -14,11 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.common.metrics.client.admin;
+package org.astraea.common.metrics.client;
 
 import org.astraea.common.metrics.HasBeanObject;
 
-public interface HasAdminMetrics extends HasBeanObject {
+@FunctionalInterface
+public interface HasSelectorMetrics extends HasBeanObject {
 
   default String clientId() {
     return beanObject().properties().get("client-id");
@@ -68,6 +69,10 @@ default double incomingByteTotal() {
     return (double) beanObject().attributes().get("incoming-byte-total");
   }
 
+  default double ioRatio() {
+    return (double) beanObject().attributes().get("io-ratio");
+  }
+
   default double ioTimeNsAvg() {
     return (double) beanObject().attributes().get("io-time-ns-avg");
   }
@@ -76,6 +81,10 @@ default double ioTimeNsTotal() {
     return (double) beanObject().attributes().get("io-time-ns-total");
   }
 
+  default double ioWaitRatio() {
+    return (double) beanObject().attributes().get("io-wait-ratio");
+  }
+
   default double ioWaitTimeNsAvg() {
     return (double) beanObject().attributes().get("io-wait-time-ns-avg");
   }
@@ -84,6 +93,14 @@ default double ioWaitTimeNsTotal() {
     return (double) beanObject().attributes().get("io-wait-time-ns-total");
   }
 
+  default double ioWaitTimeTotal() {
+    return (double) beanObject().attributes().get("io-waittime-total");
+  }
+
+  default double ioTimeTotal() {
+    return (double) beanObject().attributes().get("iotime-total");
+  }
+
   default double networkIoRate() {
     return (double) beanObject().attributes().get("network-io-rate");
   }

common/src/main/java/org/astraea/common/metrics/client/admin/AdminMetrics.java

@@ -21,6 +21,7 @@
 import org.astraea.common.metrics.BeanQuery;
 import org.astraea.common.metrics.MBeanClient;
 import org.astraea.common.metrics.client.HasNodeMetrics;
+import org.astraea.common.metrics.client.HasSelectorMetrics;
 
 public class AdminMetrics {
   /**
@@ -43,7 +44,7 @@ public static Collection<HasNodeMetrics> nodes(MBeanClient mBeanClient) {
         .collect(Collectors.toUnmodifiableList());
   }
 
-  public static Collection<HasAdminMetrics> of(MBeanClient mBeanClient) {
+  public static Collection<HasSelectorMetrics> of(MBeanClient mBeanClient) {
     return mBeanClient
         .beans(
             BeanQuery.builder()
@@ -52,7 +53,7 @@ public static Collection<HasAdminMetrics> of(MBeanClient mBeanClient) {
                 .property("client-id", "*")
                 .build())
         .stream()
-        .map(b -> (HasAdminMetrics) () -> b)
+        .map(b -> (HasSelectorMetrics) () -> b)
         .collect(Collectors.toUnmodifiableList());
   }
 }

common/src/main/java/org/astraea/common/metrics/client/consumer/HasConsumerCoordinatorMetrics.java

@@ -16,13 +16,9 @@
  */
 package org.astraea.common.metrics.client.consumer;
 
-import org.astraea.common.metrics.HasBeanObject;
+import org.astraea.common.metrics.client.HasCoordinatorMetrics;
 
-public interface HasConsumerCoordinatorMetrics extends HasBeanObject {
-
-  default String clientId() {
-    return beanObject().properties().get("client-id");
-  }
+public interface HasConsumerCoordinatorMetrics extends HasCoordinatorMetrics {
 
   default double assignedPartitions() {
     return (double) beanObject().attributes().get("assigned-partitions");
@@ -44,42 +40,6 @@ default double commitTotal() {
     return (double) beanObject().attributes().get("commit-total");
   }
 
-  default double failedRebalanceRatePerHour() {
-    return (double) beanObject().attributes().get("failed-rebalance-rate-per-hour");
-  }
-
-  default double failedRebalanceTotal() {
-    return (double) beanObject().attributes().get("failed-rebalance-total");
-  }
-
-  default double heartbeatRate() {
-    return (double) beanObject().attributes().get("heartbeat-rate");
-  }
-
-  default double joinRate() {
-    return (double) beanObject().attributes().get("join-rate");
-  }
-
-  default double joinTimeAvg() {
-    return (double) beanObject().attributes().get("join-time-avg");
-  }
-
-  default double joinTimeMax() {
-    return (double) beanObject().attributes().get("join-time-max");
-  }
-
-  default double joinTotal() {
-    return (double) beanObject().attributes().get("join-total");
-  }
-
-  default double lastHeartbeatSecondsAgo() {
-    return (double) beanObject().attributes().get("last-heartbeat-seconds-ago");
-  }
-
-  default double lastRebalanceSecondsAgo() {
-    return (double) beanObject().attributes().get("last-rebalance-seconds-ago");
-  }
-
   default double partitionAssignedLatencyAvg() {
     return (double) beanObject().attributes().get("partition-assigned-latency-avg");
   }
@@ -103,40 +63,4 @@ default double partitionRevokedLatencyAvg() {
   default double partitionRevokedLatencyMax() {
     return (double) beanObject().attributes().get("partition-revoked-latency-max");
   }
-
-  default double rebalanceLatencyAvg() {
-    return (double) beanObject().attributes().get("rebalance-latency-avg");
-  }
-
-  default double rebalanceLatencyMax() {
-    return (double) beanObject().attributes().get("rebalance-latency-max");
-  }
-
-  default double rebalanceLatencyTotal() {
-    return (double) beanObject().attributes().get("rebalance-latency-total");
-  }
-
-  default double rebalanceRatePerHour() {
-    return (double) beanObject().attributes().get("rebalance-rate-per-hour");
-  }
-
-  default double rebalanceTotal() {
-    return (double) beanObject().attributes().get("rebalance-total");
-  }
-
-  default double syncRate() {
-    return (double) beanObject().attributes().get("sync-rate");
-  }
-
-  default double syncTimeAvg() {
-    return (double) beanObject().attributes().get("sync-time-avg");
-  }
-
-  default double syncTimeMax() {
-    return (double) beanObject().attributes().get("sync-time-max");
-  }
-
-  default double syncTotal() {
-    return (double) beanObject().attributes().get("sync-total");
-  }
 }