Backend jmx metrics (#64)

* Start doing endpoint for jmx metrics

* Added endpoint for getting jmx metric per broker

* Cluster jmx metrics sum endpoit added

* Added endpoints for cluster metrics and broker metrics

* Cleared some code

* Fixed jmxmetrics names

* Changed to all values in metrics

* Removed redundant imports

* Renamed param constant

* Changed to calculate brokers and clusters metrics in one place

* Removed redundant imports

* Fixed some mistakes

* Replaced multiple method usage into single

* Fixed mulptiple call

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

* Just small fixes

* removed redundant variable

* Renamed method for cluster level metrics

* Fixed after PR and added sum for number cluster metrics by num and persec keywords in canonicalname

* Added metricdto object

* Added list of metrics to enum

* Renames and optimizings

* Renamed jmxmetrics objects param to metrics

Co-authored-by: Roman Nedzvetskiy <[email protected]>

Author: yazebochan

docker/kafka-clusters-only.yaml

@@ -23,39 +23,35 @@ services:
       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka0:9092,PLAINTEXT_HOST://localhost:29091 #,PLAIN://kafka0:29090
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT #,PLAIN:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
+      JMX_PORT: 9997
+      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9997
+
+  kafka01:
+    image: confluentinc/cp-kafka:5.1.0
+    depends_on:
+      - zookeeper0
+    ports:
+      - 29093:29093
+      - 9999:9999
+    environment:
+      KAFKA_BROKER_ID: 2
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2183
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka01:9092,PLAINTEXT_HOST://localhost:29093,PLAIN://kafka0:29090
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAIN:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
       JMX_PORT: 9997
       KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9997
-#
-#  kafka01:
-#    image: confluentinc/cp-kafka:5.1.0
-#    depends_on:
-#      - zookeeper0
-#    ports:
-#      - 29093:29093
-#      - 9999:9999
-#    environment:
-#      KAFKA_BROKER_ID: 2
-#      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2183
-#      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka01:9092,PLAINTEXT_HOST://localhost:29093,PLAIN://kafka0:29090
-#      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,PLAIN:PLAINTEXT
-#      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-#      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
-#      JMX_PORT: 9997
-#      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Dcom.sun.management.jmxremote.rmi.port=9997
 
   kafka-init-topics0:
     image: confluentinc/cp-kafka:5.1.0
     depends_on:
       - kafka0
-      - kafka1
-    command:
-                "kafka-console-producer --broker-list kafka1:9092 --topic secondUsers && \
-                This is message 1 && \
-                This is message 2 && \
-                This is message 3 && \
-                Message 4 && \
-                Message 5"
+    command: "bash -c 'echo Waiting for Kafka to be ready... && \
+                cub kafka-ready -b kafka0:9092 1 20 && \
+                kafka-topics --create --topic users --partitions 2 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2183 && \
+                kafka-topics --create --topic messages --partitions 3 --replication-factor 2 --if-not-exists --zookeeper zookeeper0:2183'"
     environment:
       KAFKA_BROKER_ID: ignored
       KAFKA_ZOOKEEPER_CONNECT: ignored
@@ -101,21 +97,21 @@ services:
     networks:
      - default
 
-#  schemaregistry0:
-#    image: confluentinc/cp-schema-registry:5.1.0
-#    depends_on:
-#      - zookeeper0
-#      - kafka0
-#      - kafka01
-#    ports:
-#      - 8085:8085
-#    environment:
-#      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka0:9092,PLAINTEXT://kafka01:9092
-#      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper0:2183
-#      SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL: PLAINTEXT
-#      SCHEMA_REGISTRY_HOST_NAME: schemaregistry
-#      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8085
-#
-#      SCHEMA_REGISTRY_SCHEMA_REGISTRY_INTER_INSTANCE_PROTOCOL: "http"
-#      SCHEMA_REGISTRY_LOG4J_ROOT_LOGLEVEL: INFO
-#      SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
+  schemaregistry0:
+    image: confluentinc/cp-schema-registry:5.1.0
+    depends_on:
+      - zookeeper0
+      - kafka0
+      - kafka01
+    ports:
+      - 8085:8085
+    environment:
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka0:9092,PLAINTEXT://kafka01:9092
+      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper0:2183
+      SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL: PLAINTEXT
+      SCHEMA_REGISTRY_HOST_NAME: schemaregistry
+      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8085
+
+      SCHEMA_REGISTRY_SCHEMA_REGISTRY_INTER_INSTANCE_PROTOCOL: "http"
+      SCHEMA_REGISTRY_LOG4J_ROOT_LOGLEVEL: INFO
+      SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/config/Config.java

@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.cluster.config;
 
+import com.provectus.kafka.ui.cluster.util.JmxMetricsNames;
 import com.provectus.kafka.ui.cluster.util.JmxPoolFactory;
 import org.apache.commons.pool2.KeyedObjectPool;
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
@@ -9,6 +10,9 @@
 import org.springframework.jmx.export.MBeanExporter;
 
 import javax.management.remote.JMXConnector;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 @Configuration
 public class Config {
@@ -35,4 +39,9 @@ public MBeanExporter exporter()
         exporter.setExcludedBeans("pool");
         return exporter;
     }
+
+    @Bean
+    public List<String> jmxMetricsNames() {
+        return Stream.of(JmxMetricsNames.values()).map(Enum::name).collect(Collectors.toList());
+    }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/mapper/ClusterMapper.java

@@ -9,24 +9,16 @@
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
 
-import java.math.BigDecimal;
-
 @Mapper(componentModel = "spring")
 public interface ClusterMapper {
 
-    KafkaCluster toKafkaCluster(ClustersProperties.Cluster clusterProperties);
-
     @Mapping(target = "brokerCount", source = "metrics.brokerCount")
     @Mapping(target = "onlinePartitionCount", source = "metrics.onlinePartitionCount")
     @Mapping(target = "topicCount", source = "metrics.topicCount")
-    @Mapping(target = "bytesInPerSec", source = "metrics.bytesInPerSec")
-    @Mapping(target = "bytesOutPerSec", source = "metrics.bytesOutPerSec")
+    @Mapping(target = "metrics", source = "metrics.metrics")
     Cluster toCluster(KafkaCluster cluster);
 
-    default BigDecimal map (Number number) {
-        return new BigDecimal(number.toString());
-    }
-
+    KafkaCluster toKafkaCluster(ClustersProperties.Cluster clusterProperties);
     BrokersMetrics toBrokerMetrics(InternalClusterMetrics metrics);
     Topic toTopic(InternalTopic topic);
     TopicDetails toTopicDetails(InternalTopic topic);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/model/InternalBrokerMetrics.java

@@ -1,10 +1,14 @@
 package com.provectus.kafka.ui.cluster.model;
 
+import com.provectus.kafka.ui.model.Metric;
 import lombok.Builder;
 import lombok.Data;
 
+import java.util.List;
+
 @Data
 @Builder(toBuilder = true)
 public class InternalBrokerMetrics {
     private final Long segmentSize;
+    private final List<Metric> jmxMetrics;
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/model/InternalClusterMetrics.java

@@ -1,9 +1,10 @@
 package com.provectus.kafka.ui.cluster.model;
 
+import com.provectus.kafka.ui.model.Metric;
 import lombok.Builder;
 import lombok.Data;
 
-import java.math.BigDecimal;
+import java.util.List;
 import java.util.Map;
 
 
@@ -24,5 +25,6 @@ public class InternalClusterMetrics {
     private final int segmentCount;
     private final long segmentSize;
     private final Map<Integer, InternalBrokerMetrics> internalBrokerMetrics;
+    private final List<Metric> metrics;
     private final int zooKeeperStatus;
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/model/InternalTopic.java

@@ -22,7 +22,6 @@ public class InternalTopic {
     private final int inSyncReplicas;
     private final int replicationFactor;
     private final int underReplicatedPartitions;
-    //TODO: find way to fill
     private final long segmentSize;
     private final int segmentCount;
     private final Map<TopicPartition, Long> partitionSegmentSize;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/model/MetricDto.java

@@ -0,0 +1,14 @@
+package com.provectus.kafka.ui.cluster.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+import java.math.BigDecimal;
+
+@Getter
+@AllArgsConstructor
+public class MetricDto {
+    private String canonicalName;
+    private String metricName;
+    private BigDecimal value;
+}

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

@@ -38,15 +38,20 @@ public List<Cluster> getClusters() {
                 .collect(Collectors.toList());
     }
 
-    public Optional<BrokersMetrics> getBrokersMetrics(String name) {
-        return clustersStorage.getClusterByName(name)
+    public Mono<BrokersMetrics> getBrokersMetrics(String name, Integer id) {
+        return Mono.justOrEmpty(clustersStorage.getClusterByName(name)
                 .map(KafkaCluster::getMetrics)
-                .map(clusterMapper::toBrokerMetrics);
+                .map(s -> {
+                    var brokerMetrics = clusterMapper.toBrokerMetrics(s);
+                    brokerMetrics.setMetrics(s.getInternalBrokerMetrics().get(id).getJmxMetrics());
+                    brokerMetrics.setSegmentZise(Long.valueOf(s.getSegmentSize()).intValue());
+                    return brokerMetrics;
+                }));
     }
 
     public List<Topic> getTopics(String name) {
         return clustersStorage.getClusterByName(name)
-                .map( c ->
+                .map(c ->
                         c.getTopics().values().stream()
                                 .map(clusterMapper::toTopic)
                                 .collect(Collectors.toList())
@@ -127,6 +132,7 @@ public Flux<Broker> getBrokers (String clusterName) {
                     .map(n -> n.stream().map(node -> {
                         Broker broker = new Broker();
                         broker.setId(node.idString());
+                        broker.setHost(node.host());
                         return broker;
                     }).collect(Collectors.toList())))
                 .flatMapMany(Flux::fromIterable);
@@ -154,6 +160,5 @@ public Flux<TopicMessage> getMessages(String clusterName, String topicName, Cons
         return clustersStorage.getClusterByName(clusterName)
                 .map(c -> consumingService.loadMessages(c, topicName, consumerPosition, query, limit))
                 .orElse(Flux.empty());
-
     }
 }

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

@@ -1,7 +1,7 @@
 package com.provectus.kafka.ui.cluster.util;
 
-import com.provectus.kafka.ui.cluster.model.*;
 import com.provectus.kafka.ui.cluster.deserialization.RecordDeserializer;
+import com.provectus.kafka.ui.cluster.model.*;
 import com.provectus.kafka.ui.model.*;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.*;
@@ -28,10 +28,6 @@
 @Slf4j
 public class ClusterUtil {
 
-
-
-
-
     private static final String CLUSTER_VERSION_PARAM_KEY = "inter.broker.protocol.version";
 
     private static final ZoneId UTC_ZONE_ID = ZoneId.of("UTC");
@@ -56,7 +52,7 @@ public static Mono<String> toMono(KafkaFuture<Void> future, String topicName) {
         }));
     }
 
-    public static ConsumerGroup convertToConsumerGroup(ConsumerGroupDescription c, KafkaCluster cluster) {
+    public static ConsumerGroup convertToConsumerGroup(ConsumerGroupDescription c) {
         ConsumerGroup consumerGroup = new ConsumerGroup();
         consumerGroup.setConsumerGroupId(c.groupId());
         consumerGroup.setNumConsumers(c.members().size());