BE: Fixes #445 Added connect stats

Author: germanosin

api/src/main/java/io/kafbat/ui/mapper/ClusterMapper.java

@@ -12,6 +12,10 @@
 import io.kafbat.ui.model.ConfigSourceDTO;
 import io.kafbat.ui.model.ConfigSynonymDTO;
 import io.kafbat.ui.model.ConnectDTO;
+import io.kafbat.ui.model.ConnectorDTO;
+import io.kafbat.ui.model.ConnectorStateDTO;
+import io.kafbat.ui.model.ConnectorStatusDTO;
+import io.kafbat.ui.model.ConnectorTaskStatusDTO;
 import io.kafbat.ui.model.InternalBroker;
 import io.kafbat.ui.model.InternalBrokerConfig;
 import io.kafbat.ui.model.InternalBrokerDiskUsage;
@@ -27,13 +31,16 @@
 import io.kafbat.ui.model.Metrics;
 import io.kafbat.ui.model.PartitionDTO;
 import io.kafbat.ui.model.ReplicaDTO;
+import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.TopicConfigDTO;
 import io.kafbat.ui.model.TopicDTO;
 import io.kafbat.ui.model.TopicDetailsDTO;
 import io.kafbat.ui.model.TopicProducerStateDTO;
+import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import io.kafbat.ui.service.metrics.RawMetric;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.ProducerState;
 import org.apache.kafka.common.acl.AccessControlEntry;
@@ -107,7 +114,38 @@ default ConfigSynonymDTO toConfigSynonym(ConfigEntry.ConfigSynonym config) {
 
   ReplicaDTO toReplica(InternalReplica replica);
 
-  ConnectDTO toKafkaConnect(ClustersProperties.ConnectCluster connect);
+  default ConnectDTO toKafkaConnect(ClustersProperties.ConnectCluster connect, List<InternalConnectorInfo> connectors) {
+    int connectorCount = connectors.size();
+    int failedConnectors = 0;
+    int tasksCount = 0;
+    int failedTasksCount = 0;
+
+    for (InternalConnectorInfo connector : connectors) {
+      Optional<ConnectorDTO> internalConnector = Optional.ofNullable(connector.getConnector());
+
+      failedConnectors += internalConnector
+          .map(ConnectorDTO::getStatus)
+          .map(ConnectorStatusDTO::getState)
+          .filter(ConnectorStateDTO.FAILED::equals)
+          .map(s -> 1).orElse(0);
+
+      tasksCount += internalConnector.map(c -> c.getTasks().size()).orElse(0);
+
+      for (TaskDTO task : connector.getTasks()) {
+        if (task.getStatus() != null && ConnectorTaskStatusDTO.FAILED.equals(task.getStatus().getState())) {
+          failedTasksCount += tasksCount;
+        }
+      }
+    }
+
+    return new ConnectDTO()
+        .address(connect.getAddress())
+        .name(connect.getName())
+        .connectorsCount(connectorCount)
+        .failedConnectorsCount(failedConnectors)
+        .tasksCount(tasksCount)
+        .failedTasksCount(failedTasksCount);
+  }
 
   List<ClusterDTO.FeaturesEnum> toFeaturesEnum(List<ClusterFeature> features);
 

api/src/main/java/io/kafbat/ui/mapper/KafkaConnectMapper.java

@@ -11,7 +11,7 @@
 import io.kafbat.ui.model.FullConnectorInfoDTO;
 import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.TaskStatusDTO;
-import io.kafbat.ui.model.connect.InternalConnectInfo;
+import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import java.util.List;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
@@ -38,7 +38,7 @@ ConnectorPluginConfigValidationResponseDTO fromClient(
       io.kafbat.ui.connect.model.ConnectorPluginConfigValidationResponse
           connectorPluginConfigValidationResponse);
 
-  default FullConnectorInfoDTO fullConnectorInfo(InternalConnectInfo connectInfo) {
+  default FullConnectorInfoDTO fullConnectorInfo(InternalConnectorInfo connectInfo) {
     ConnectorDTO connector = connectInfo.getConnector();
     List<TaskDTO> tasks = connectInfo.getTasks();
     int failedTasksCount = (int) tasks.stream()

api/src/main/java/io/kafbat/ui/model/connect/InternalConnectorInfo.java

@@ -9,7 +9,7 @@
 
 @Data
 @Builder(toBuilder = true)
-public class InternalConnectInfo {
+public class InternalConnectorInfo {
   private final ConnectorDTO connector;
   private final Map<String, Object> config;
   private final List<TaskDTO> tasks;

api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java

@@ -21,7 +21,7 @@
 import io.kafbat.ui.model.KafkaCluster;
 import io.kafbat.ui.model.NewConnectorDTO;
 import io.kafbat.ui.model.TaskDTO;
-import io.kafbat.ui.model.connect.InternalConnectInfo;
+import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import io.kafbat.ui.util.ReactiveFailover;
 import java.util.List;
 import java.util.Map;
@@ -46,11 +46,24 @@ public class KafkaConnectService {
   private final KafkaConfigSanitizer kafkaConfigSanitizer;
 
   public Flux<ConnectDTO> getConnects(KafkaCluster cluster) {
-    return Flux.fromIterable(
-        Optional.ofNullable(cluster.getOriginalProperties().getKafkaConnect())
-            .map(lst -> lst.stream().map(clusterMapper::toKafkaConnect).toList())
-            .orElse(List.of())
-    );
+    return Optional.ofNullable(cluster.getOriginalProperties().getKafkaConnect())
+        .map(connects -> Flux.fromIterable(connects).flatMap(connect ->
+            getConnectorNamesWithErrorsSuppress(cluster, connect.getName()).flatMap(connectorName ->
+                Mono.zip(
+                    getConnector(cluster, connect.getName(), connectorName),
+                    getConnectorTasks(cluster, connect.getName(), connectorName).collectList()
+                ).map(tuple ->
+                    InternalConnectorInfo.builder()
+                        .connector(tuple.getT1())
+                        .config(null)
+                        .tasks(tuple.getT2())
+                        .topics(null)
+                        .build()
+                )
+            ).collectList().map(connectors ->
+                clusterMapper.toKafkaConnect(connect, connectors)
+            )
+        )).orElse(Flux.fromIterable(List.of()));
   }
 
   public Flux<FullConnectorInfoDTO> getAllConnectors(final KafkaCluster cluster,
@@ -65,7 +78,7 @@ public Flux<FullConnectorInfoDTO> getAllConnectors(final KafkaCluster cluster,
                         getConnectorTasks(cluster, connect.getName(), connectorName).collectList(),
                         getConnectorTopics(cluster, connect.getName(), connectorName)
                     ).map(tuple ->
-                        InternalConnectInfo.builder()
+                        InternalConnectorInfo.builder()
                             .connector(tuple.getT1())
                             .config(tuple.getT2())
                             .tasks(tuple.getT3())

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

@@ -3434,6 +3434,14 @@ components:
           type: string
         address:
           type: string
+        connectors_count:
+          type: integer
+        failed_connectors_count:
+          type: integer
+        tasks_count:
+          type: integer
+        failed_tasks_count:
+          type: integer
       required:
         - name