Feature/topology visualizer contract #205

etc/checkstyle/checkstyle.xml

@@ -240,6 +240,7 @@
             <property name="tokens"
                       value="CLASS_DEF, INTERFACE_DEF, ENUM_DEF, ANNOTATION_DEF, ANNOTATION_FIELD_DEF,
                     PARAMETER_DEF, VARIABLE_DEF, METHOD_DEF"/>
+            <property name="allowedAbbreviations" value="DTO"/>
         </module>
         <module name="OverloadMethodsDeclarationOrder"/>
         <module name="VariableDeclarationUsageDistance"/>

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

@@ -5,11 +5,15 @@
 import java.util.Map;
 import java.util.Properties;
 import lombok.Data;
+import lombok.Getter;
+import lombok.Setter;
 import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 
 @Configuration
 @ConfigurationProperties("kafka")
+@EnableConfigurationProperties
 @Data
 public class ClustersProperties {
 
@@ -36,6 +40,7 @@ public static class Cluster {
     Properties properties;
     boolean readOnly = false;
     boolean disableLogDirsCollection = false;
+    List<StreamApplication> streamApplications = new ArrayList<>();
   }
 
   @Data
@@ -49,4 +54,11 @@ public static class SchemaRegistryAuth {
     String username;
     String password;
   }
+
+  @Getter
+  @Setter
+  public static class StreamApplication {
+    private String applicationId;
+    private String topologyUrl;
+  }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/StreamTopologyController.java

@@ -0,0 +1,31 @@
+package com.provectus.kafka.ui.controller;
+
+import com.provectus.kafka.ui.api.StreamTopologiesApi;
+import com.provectus.kafka.ui.model.ProcessorTopologyDTO;
+import com.provectus.kafka.ui.model.StreamApplicationsDTO;
+import com.provectus.kafka.ui.service.StreamTopologyService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequiredArgsConstructor
+public class StreamTopologyController implements StreamTopologiesApi {
+  private final StreamTopologyService topologyService;
+
+  @Override
+  public Mono<ResponseEntity<StreamApplicationsDTO>> getStreamApplications(
+      String clusterName, ServerWebExchange exchange) {
+    return Mono.just(ResponseEntity.ok(topologyService.getTopologyApplications(clusterName)));
+  }
+
+  @Override
+  public Mono<ResponseEntity<ProcessorTopologyDTO>> getStreamTopology(String clusterName,
+                                                                   String applicationId,
+                                                                   ServerWebExchange exchange) {
+    return topologyService.getStreamTopology(clusterName, applicationId)
+        .map(ResponseEntity::ok);
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/ErrorCode.java

@@ -23,7 +23,8 @@ public enum ErrorCode {
   KSQLDB_NOT_FOUND(4011, HttpStatus.NOT_FOUND),
   DIR_NOT_FOUND(4012, HttpStatus.BAD_REQUEST),
   TOPIC_OR_PARTITION_NOT_FOUND(4013, HttpStatus.BAD_REQUEST),
-  INVALID_REQUEST(4014, HttpStatus.BAD_REQUEST);
+  INVALID_REQUEST(4014, HttpStatus.BAD_REQUEST),
+  INVALID_STREAM_TOPOLOGY_STRING(4015, HttpStatus.UNPROCESSABLE_ENTITY);
 
   static {
     // codes uniqueness check

kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/InvalidStreamTopologyString.java

@@ -0,0 +1,11 @@
+package com.provectus.kafka.ui.exception;
+
+public class InvalidStreamTopologyString extends RuntimeException {
+  public InvalidStreamTopologyString() {
+    super("Invalid stream topology string");
+  }
+
+  public InvalidStreamTopologyString(String message) {
+    super(message);
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/StreamTopologyParsingException.java

@@ -0,0 +1,20 @@
+package com.provectus.kafka.ui.exception;
+
+public class StreamTopologyParsingException extends CustomBaseException {
+  public StreamTopologyParsingException() {
+    super("Stream topology string is invalid");
+  }
+
+  public StreamTopologyParsingException(String message) {
+    super(message);
+  }
+
+  public StreamTopologyParsingException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  @Override
+  public ErrorCode getErrorCode() {
+    return ErrorCode.INVALID_STREAM_TOPOLOGY_STRING;
+  }
+}

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

@@ -0,0 +1,90 @@
+package com.provectus.kafka.ui.service;
+
+import static com.provectus.kafka.ui.config.ClustersProperties.StreamApplication;
+
+import com.provectus.kafka.ui.config.ClustersProperties;
+import com.provectus.kafka.ui.exception.ClusterNotFoundException;
+import com.provectus.kafka.ui.exception.InvalidStreamTopologyString;
+import com.provectus.kafka.ui.exception.NotFoundException;
+import com.provectus.kafka.ui.exception.StreamTopologyParsingException;
+import com.provectus.kafka.ui.model.ProcessorTopologyDTO;
+import com.provectus.kafka.ui.model.StreamApplicationsDTO;
+import com.provectus.kafka.ui.service.topology.parser.StreamTopologyParser;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.springframework.stereotype.Service;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+@Log4j2
+@Service
+public class StreamTopologyService {
+  private final Map<String, Map<String, String>> clusterStreamApps;
+  private final WebClient webClient;
+  private final StreamTopologyParser topologyParser;
+
+  public StreamTopologyService(ClustersProperties clustersProperties, WebClient webClient,
+                               StreamTopologyParser topologyParser) {
+    this.clusterStreamApps = getClusterToStreamAppsMap(clustersProperties);
+    this.webClient = webClient;
+    this.topologyParser = topologyParser;
+  }
+
+  public StreamApplicationsDTO getTopologyApplications(String clusterName) {
+    final var streamApplications = new StreamApplicationsDTO();
+    final var applicationIds = Optional.ofNullable(clusterStreamApps.get(clusterName))
+        .map(Map::keySet)
+        .map(ArrayList::new)
+        .orElseThrow(ClusterNotFoundException::new);
+    return streamApplications.applicationIds(applicationIds);
+  }
+
+  public Mono<ProcessorTopologyDTO> getStreamTopology(String clusterName, String applicationId) {
+    return Optional.ofNullable(clusterStreamApps.get(clusterName))
+        .map(apps -> apps.get(applicationId))
+        .map(this::getTopologyString)
+        .map(topologyMono ->
+            topologyMono.map(s -> parseTopologyString(s, clusterName, applicationId))
+        )
+        .orElseThrow(() -> new NotFoundException("Stream application not found"));
+  }
+
+  private ProcessorTopologyDTO parseTopologyString(String topologyString, String clusterName,
+                                                String applicationId) {
+    try {
+      return topologyParser.parse(topologyString);
+    } catch (InvalidStreamTopologyString e) {
+      log.error("cannot parse stream topology", e);
+      throw new StreamTopologyParsingException(String
+          .format("cannot parse stream topology <clusterName %s>, <applicationId %s>",
+              clusterName, applicationId));
+    }
+  }
+
+  private Mono<String> getTopologyString(String topologyUrl) {
+    return webClient.get()
+        .uri(topologyUrl)
+        .retrieve()
+        .bodyToMono(String.class)
+        .doOnError(log::error);
+  }
+
+  private Map<String, Map<String, String>> getClusterToStreamAppsMap(
+      ClustersProperties clustersProperties) {
+    return clustersProperties.getClusters().stream()
+        .map(cluster -> ImmutablePair.of(cluster.getName(), getAppToEndpointMap(cluster)))
+        .collect(Collectors.toMap(ImmutablePair::getKey, ImmutablePair::getValue));
+  }
+
+  private Map<String, String> getAppToEndpointMap(ClustersProperties.Cluster cluster) {
+    return cluster.getStreamApplications().stream()
+        .collect(Collectors.toMap(
+            StreamApplication::getApplicationId,
+            StreamApplication::getTopologyUrl)
+        );
+  }
+}