Added endpoints for cluster metrics and broker metrics

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/ClusterService.java

@@ -5,6 +5,7 @@ import com.provectus.kafka.ui.cluster.model.ClustersStorage;
 import com.provectus.kafka.ui.cluster.model.ConsumerPosition;
 import com.provectus.kafka.ui.cluster.model.KafkaCluster;
 import com.provectus.kafka.ui.cluster.util.ClusterUtil;
+import com.provectus.kafka.ui.cluster.util.JmxClusterUtil;
 import com.provectus.kafka.ui.kafka.KafkaService;
 import com.provectus.kafka.ui.model.*;
 import lombok.RequiredArgsConstructor;
@@ -12,14 +13,12 @@ import lombok.SneakyThrows;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.math.BigDecimal;
 import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -157,27 +156,27 @@ public class ClusterService {
 
     }
 
-    public Mono<JmxMetric> getJmxMetric(String clusterName, Integer nodeId, JmxMetric metric) {
+    public Mono<JmxMetric> getJmxMetric(String clusterName, Integer nodeId, String canonicalName) {
         return clustersStorage.getClusterByName(clusterName)
                 .map(c -> kafkaService.getOrCreateAdminClient(c)
                         .flatMap(a -> ClusterUtil.toMono(a.getAdminClient().describeCluster().nodes())
                                     .map(n -> n.stream().filter(s -> s.id() == nodeId).findFirst().orElseThrow().host()))
-                        .map(host -> kafkaService.getJmxMetric(c, c.getJmxPort(), host, metric))).orElseThrow();
+                        .map(host -> kafkaService.getJmxMetric(c, c.getJmxPort(), host, canonicalName))).orElseThrow();
     }
 
-    public Mono<JmxMetric> getClusterJmxMetric(String clusterName, JmxMetric metric) {
+    public Mono<JmxMetric> getClusterJmxMetric(String clusterName, String canonicalName) {
         return clustersStorage.getClusterByName(clusterName)
                 .map(c -> kafkaService.getOrCreateAdminClient(c)
                     .flatMap(eac -> ClusterUtil.toMono(eac.getAdminClient().describeCluster().nodes()))
                     .flatMapIterable(n -> n.stream().flatMap(node -> Stream.of(node.host())).collect(Collectors.toList()))
-                    .map(host -> kafkaService.getJmxMetric(c, c.getJmxPort(), host, metric))
+                    .map(host -> kafkaService.getJmxMetric(c, c.getJmxPort(), host, canonicalName))
                     .collectList()
-                    .map(s -> s.stream().filter(metric1 -> metric1.getCanonicalName().equals(metric.getCanonicalName()))
+                    .map(s -> s.stream().filter(metric1 -> JmxClusterUtil.metricNamesEquals(metric1.getCanonicalName(), canonicalName))
                             .collect(Collectors.toList())
                             .stream().reduce((jmx, jmx1) -> {
                                 if (jmx.getCanonicalName().equals(jmx1.getCanonicalName())) {
                                     jmx.getValue().keySet().forEach(k -> jmx1.getValue().compute(k, (k1, v1) ->
-                                        ((BigDecimal) v1).add(((BigDecimal) jmx1.getValue().get(k)))));
+                                        JmxClusterUtil.metricValueReduce(v1, jmx1.getValue().get(k))));
                                 }
                             return jmx;
                             })

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

@@ -10,11 +10,9 @@ import org.springframework.stereotype.Component;
 import javax.management.*;
 import javax.management.remote.JMXConnector;
 import java.io.IOException;
+import java.math.BigDecimal;
 import java.net.MalformedURLException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 
 @Component
@@ -30,7 +28,7 @@ public class JmxClusterUtil {
     public List<InternalJmxMetric> getJmxMetricsNames(int jmxPort, String jmxHost) {
         String jmxUrl = JMX_URL + jmxHost + ":" + jmxPort + "/" + JMX_SERVICE_TYPE;
         List<InternalJmxMetric> result = new ArrayList<>();
-        JMXConnector srv;
+        JMXConnector srv = null;
         try {
             srv = pool.borrowObject(jmxUrl);
             MBeanServerConnection msc = srv.getMBeanServerConnection();
@@ -43,10 +41,13 @@ public class JmxClusterUtil {
                 internalMetric.canonicalName(j.getCanonicalName());
                 result.add(internalMetric.build());
             });
+            pool.returnObject(jmxUrl, srv);
         } catch (IOException ioe) {
             log.error("Cannot get jmxMetricsNames, {}", jmxUrl, ioe);
+            closeConnectionExceptionally(jmxUrl, srv);
         } catch (Exception e) {
             log.error("Cannot get JmxConnection from pool, {}", jmxUrl, e);
+            closeConnectionExceptionally(jmxUrl, srv);
         }
         return result;
     }
@@ -96,8 +97,34 @@ public class JmxClusterUtil {
     }
 
     public static String getParamFromName(String param, String name) {
+        if (!name.contains(param)) {
+            return null;
+        }
         int paramValueBeginIndex = name.indexOf(param) + param.length() + 1;
         int paramValueEndIndex = name.indexOf(',', paramValueBeginIndex);
         return paramValueEndIndex != -1 ? name.substring(paramValueBeginIndex, paramValueEndIndex) : name.substring(paramValueBeginIndex);
     }
+
+    public static boolean metricNamesEquals (String metric1Name, String metric2Name) {
+        boolean result = false;
+        try {
+            result = Objects.equals(getParamFromName("name", metric1Name), getParamFromName("name", metric2Name))
+                    &&
+                    Objects.equals(getParamFromName("type", metric1Name), getParamFromName("type", metric2Name))
+                    &&
+                    getParamFromName("topic", metric1Name) == null ||
+                            Objects.equals(getParamFromName("topic", metric1Name), getParamFromName("topic", metric2Name));
+        } catch (NullPointerException npe) {
+            log.error("Cannot compare {} with {}, npe caught", metric1Name, metric2Name);
+        }
+        return result;
+    }
+
+    public static Object metricValueReduce(Object value1, Object value2) {
+        if (value1 instanceof Number) {
+            return new BigDecimal(value1.toString()).add(new BigDecimal(value2.toString()));
+        } else {
+            return value1;
+        }
+    }
 }

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

@@ -346,13 +346,13 @@ public class KafkaService {
             );
     }
 
-    public JmxMetric getJmxMetric (KafkaCluster cluster, int jmxPort, String host, JmxMetric metric) {
+    public JmxMetric getJmxMetric (KafkaCluster cluster, int jmxPort, String host, String canonicalName) {
         var jmxMetric = cluster.getMetrics().getJmxMetricsNames().stream().filter(c -> {
             var foundTopic = false;
-            var found = JmxClusterUtil.getParamFromName("name", metric.getCanonicalName()).equals(c.getName())
-                        && JmxClusterUtil.getParamFromName("type", metric.getCanonicalName()).equals(c.getType());
+            var found = JmxClusterUtil.getParamFromName("name", canonicalName).equals(c.getName())
+                        && JmxClusterUtil.getParamFromName("type", canonicalName).equals(c.getType());
             if (found && c.getTopic() != null) {
-                foundTopic = c.getTopic().equals(JmxClusterUtil.getParamFromName("topic", metric.getCanonicalName()));
+                foundTopic = c.getTopic().equals(JmxClusterUtil.getParamFromName("topic", canonicalName));
             }
             return found && foundTopic;
         }).findFirst().orElseThrow();

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/MetricsRestController.java

@@ -99,17 +99,16 @@ public class MetricsRestController implements ApiClustersApi {
     }
 
     @Override
-    public Mono<ResponseEntity<JmxMetric>> getBrokerJmxMetric(String clusterName, Integer host, Mono<JmxMetric> metric, ServerWebExchange exchange){
-        return metric
-                .flatMap(m -> clusterService.getJmxMetric(clusterName, host, m)
-                        .map(ResponseEntity::ok));
+    public Mono<ResponseEntity<JmxMetric>> getBrokerJmxMetric(String clusterName, Integer host, String canonicalName, ServerWebExchange exchange){
+        return
+               clusterService.getJmxMetric(clusterName, host, canonicalName)
+                        .map(ResponseEntity::ok);
     }
 
     @Override
-    public Mono<ResponseEntity<JmxMetric>> getClusterJmxMetric(String clusterName, Mono<JmxMetric> metric, ServerWebExchange exchange){
-        return metric
-                .flatMap(m -> clusterService.getClusterJmxMetric(clusterName, m)
-                        .map(ResponseEntity::ok));
+    public Mono<ResponseEntity<JmxMetric>> getClusterJmxMetric(String clusterName, String canonicalName, ServerWebExchange exchange){
+        return clusterService.getClusterJmxMetric(clusterName, canonicalName)
+                        .map(ResponseEntity::ok);
     }
 
     private Mono<ConsumerPosition> parseConsumerPosition(SeekType seekType, List<String> seekTo) {

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

@@ -264,8 +264,8 @@ paths:
               schema:
                 $ref: '#/components/schemas/ConsumerGroupDetails'
 
-  /api/clusters/{clusterName}/metrics/brokers/{brokerId}:
-    post:
+  /api/clusters/{clusterName}/brokers/{brokerId}/metrics/{canonicalName}:
+    get:
       tags:
         - /api/clusters
       summary: get specific JmxMetric for broker
@@ -281,11 +281,11 @@ paths:
           required: true
           schema:
             type: integer
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: '#/components/schemas/JmxMetric'
+        - name: canonicalName
+          in: path
+          required: true
+          schema:
+            type: string
       responses:
         200:
           description: OK
@@ -295,7 +295,7 @@ paths:
                 $ref: '#/components/schemas/JmxMetric'
 
   /api/clusters/{clusterName}/metrics:
-    post:
+    get:
       tags:
         - /api/clusters
       summary: get specific JmxMetric for cluster
@@ -306,11 +306,11 @@ paths:
           required: true
           schema:
             type: string
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: '#/components/schemas/JmxMetric'
+        - name: canonicalName
+          in: path
+          required: true
+          schema:
+            type: string
       responses:
         200:
           description: OK