Removed cluster level metrics, now only broker-level metrics in cluster

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/util/JmxClusterUtil.java

@@ -53,10 +53,10 @@ public class JmxClusterUtil {
         return result;
     }
 
-    private Map<String, Object> getJmxMetric(int jmxPort, String jmxHost, String canonicalName) {
+    private Map<String, BigDecimal> getJmxMetric(int jmxPort, String jmxHost, String canonicalName) {
         String jmxUrl = JMX_URL + jmxHost + ":" + jmxPort + "/" + JMX_SERVICE_TYPE;
 
-        Map<String, Object> resultAttr = new HashMap<>();
+        Map<String, BigDecimal> resultAttr = new HashMap<>();
         JMXConnector srv = null;
         try {
             srv = pool.borrowObject(jmxUrl);
@@ -66,8 +66,10 @@ public class JmxClusterUtil {
             var attrNames = msc.getMBeanInfo(name).getAttributes();
             for (MBeanAttributeInfo attrName : attrNames) {
                 var value = msc.getAttribute(name, attrName.getName());
-                if (value instanceof Number) {
-                    resultAttr.put(attrName.getName(), value);
+                if (value instanceof BigDecimal) {
+                    resultAttr.put(attrName.getName(), (BigDecimal) value);
+                } else if (value instanceof Integer) {
+                    resultAttr.put(attrName.getName(), new BigDecimal((Integer) value));
                 }
             }
             pool.returnObject(jmxUrl, srv);
@@ -98,11 +100,11 @@ public class JmxClusterUtil {
         }
     }
 
-    public static Object metricValueReduce(Object value1, Object value2) {
-        if (value1 instanceof Number) {
+    public static BigDecimal metricValueReduce(Number value1, Number value2) {
+        if (value1 instanceof Integer) {
             return new BigDecimal(value1.toString()).add(new BigDecimal(value2.toString()));
         } else {
-            return value1;
+            return new BigDecimal(value1.longValue()).add(new BigDecimal(value2.longValue()));
         }
     }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/kafka/KafkaService.java

@@ -352,36 +352,22 @@ public class KafkaService {
     }
 
     private Mono<InternalClusterMetrics> fillJmxMetrics (InternalClusterMetrics internalClusterMetrics, String clusterName) {
+        return fillBrokerMetrics(internalClusterMetrics, clusterName);
+    }
+
+    private Mono<InternalClusterMetrics> fillBrokerMetrics(InternalClusterMetrics internalClusterMetrics, String clusterName) {
         return getOrCreateAdminClient(clustersStorage.getClusterByName(clusterName).orElseThrow())
                 .flatMap(ac -> ClusterUtil.toMono(ac.getAdminClient().describeCluster().nodes()))
                 .flatMapIterable(nodes -> nodes)
-                .flatMap(broker -> getOrCreateAdminClient(clustersStorage.getClusterByName(clusterName).orElseThrow())
-                            .flatMap(ac -> ClusterUtil.toMono(ac.getAdminClient().describeCluster().nodes()))
-                    .map(node -> getJmxMetric(clusterName, node.stream().filter(n -> n.id() == broker.id()).findFirst().orElseThrow()))
-                    .map(jmx -> {
-                        var jmxMetric = internalClusterMetrics.getInternalBrokerMetrics().get(broker.id()).toBuilder().jmxMetrics(jmx).build();
-                        var tempBrokerMetrics = internalClusterMetrics.getInternalBrokerMetrics();
-                        tempBrokerMetrics.put(broker.id(), jmxMetric);
-                        return internalClusterMetrics.toBuilder().internalBrokerMetrics(tempBrokerMetrics).build();
-                    })).collectList()
-                .map(s -> s.stream().reduce((s1, s2) -> {
-                    s1.getInternalBrokerMetrics().putAll(s2.getInternalBrokerMetrics().entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
-                    return s1;
-                }).orElseThrow())
-                .map(i -> {
-                    var tempMetrics = new HashMap<>(i.getInternalBrokerMetrics());
-                    var resultMetrics = tempMetrics.values().stream().flatMap(s -> Stream.of(s.getJmxMetrics())).reduce((s1, s2) -> {
-                        s1.forEach(j1 -> {
-                            s2.forEach(j2 -> {
-                                if (j1.getCanonicalName().equals(j2.getCanonicalName())) {
-                                    j1.getValue().keySet().forEach(k -> j2.getValue().compute(k, (k1, v1) ->
-                                            JmxClusterUtil.metricValueReduce(j1, j2.getValue().get(k1))));
-                                }
-                            });
-                        });
-                        return s1;
-                    }).orElseThrow();
-                    return i.toBuilder().jmxMetrics(resultMetrics).build();
+                .map(broker -> {
+                    var jmx = getJmxMetric(clusterName, broker);
+                    Map<Integer, InternalBrokerMetrics> result = new HashMap<>();
+                    result.put(broker.id(), internalClusterMetrics.getInternalBrokerMetrics().get(broker.id()).toBuilder().jmxMetrics(jmx).build());
+                    return result;
+                })
+                .collectList()
+                .map(s -> { var brokerMetrics = ClusterUtil.toSingleMap(s.stream());
+                    return internalClusterMetrics.toBuilder().internalBrokerMetrics(brokerMetrics).build();
                 });
     }
 }

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -531,4 +531,4 @@ components:
         value:
           type: string
           additionalProperties:
-            type: object
+            type: number