Expose metrics per cluster & added graphs enabled feature

Author: germanosin

api/src/main/java/io/kafbat/ui/config/ClustersProperties.java

@@ -37,6 +37,8 @@ public class ClustersProperties {
 
   PollingProperties polling = new PollingProperties();
 
+  MetricsStorage defaultMetricsStorage = new MetricsStorage();
+
   @Data
   public static class Cluster {
     @NotBlank(message = "field name for for cluster could not be blank")

api/src/main/java/io/kafbat/ui/controller/PrometheusExposeController.java

@@ -4,6 +4,8 @@
 import io.kafbat.ui.model.KafkaCluster;
 import io.kafbat.ui.service.StatisticsCache;
 import io.kafbat.ui.service.metrics.prometheus.PrometheusExpose;
+import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import org.springframework.http.ResponseEntity;
@@ -18,7 +20,7 @@ public class PrometheusExposeController extends AbstractController implements Pr
   private final StatisticsCache statisticsCache;
 
   @Override
-  public Mono<ResponseEntity<String>> getAllMetrics(ServerWebExchange exchange) {
+  public Mono<ResponseEntity<String>> exposeAllMetrics(ServerWebExchange exchange) {
     return Mono.just(
         PrometheusExpose.exposeAllMetrics(
             clustersStorage.getKafkaClusters()
@@ -29,4 +31,17 @@ public Mono<ResponseEntity<String>> getAllMetrics(ServerWebExchange exchange) {
     );
   }
 
+  @Override
+  public Mono<ResponseEntity<String>> exposeClusterMetrics(String clusterName,
+                                                           ServerWebExchange exchange) {
+    Optional<KafkaCluster> cluster = clustersStorage.getClusterByName(clusterName);
+    if (cluster.isPresent() && cluster.get().isExposeMetricsViaPrometheusEndpoint()) {
+      return Mono.just(PrometheusExpose.exposeAllMetrics(
+          Map.of(clusterName, statisticsCache.get(cluster.get()).getMetrics())
+      ));
+    } else {
+      return Mono.just(ResponseEntity.notFound().build());
+    }
+  }
+
 }

api/src/main/java/io/kafbat/ui/model/ClusterFeature.java

@@ -7,5 +7,6 @@ public enum ClusterFeature {
   TOPIC_DELETION,
   KAFKA_ACL_VIEW,
   KAFKA_ACL_EDIT,
-  CLIENT_QUOTA_MANAGEMENT
+  CLIENT_QUOTA_MANAGEMENT,
+  GRAPHS_ENABLED
 }

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

@@ -36,6 +36,10 @@ public Mono<List<ClusterFeature>> getAvailableFeatures(ReactiveAdminClient admin
       features.add(Mono.just(ClusterFeature.KSQL_DB));
     }
 
+    if (cluster.getPrometheusStorageClient() != null) {
+      features.add(Mono.just(ClusterFeature.GRAPHS_ENABLED));
+    }
+
     if (cluster.getSchemaRegistryClient() != null) {
       features.add(Mono.just(ClusterFeature.SCHEMA_REGISTRY));
     }

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

@@ -87,9 +87,18 @@ public KafkaCluster create(ClustersProperties properties,
     if (ksqlConfigured(clusterProperties)) {
       builder.ksqlClient(ksqlClient(clusterProperties));
     }
+    if (prometheusStorageConfigured(properties.getDefaultMetricsStorage())) {
+      builder.prometheusStorageClient(
+          prometheusStorageClient(properties.getDefaultMetricsStorage(), clusterProperties.getSsl())
+      );
+    }
     if (prometheusStorageConfigured(clusterProperties)) {
-      builder.prometheusStorageClient(prometheusStorageClient(clusterProperties));
+      builder.prometheusStorageClient(prometheusStorageClient(
+          clusterProperties.getMetrics().getStore(),
+          clusterProperties.getSsl())
+      );
     }
+
     builder.originalProperties(clusterProperties);
     return builder.build();
   }
@@ -129,7 +138,8 @@ public Mono<ClusterConfigValidationDTO> validate(ClustersProperties.Cluster clus
             : Mono.<Optional<Map<String, ApplicationPropertyValidationDTO>>>just(Optional.empty()),
 
         prometheusStorageConfigured(clusterProperties)
-            ? validatePrometheusStore(() -> prometheusStorageClient(clusterProperties)).map(Optional::of)
+            ? validatePrometheusStore(() -> prometheusStorageClient(
+                clusterProperties.getMetrics().getStore(), clusterProperties.getSsl())).map(Optional::of)
             : Mono.<Optional<ApplicationPropertyValidationDTO>>just(Optional.empty())
     ).map(tuple -> {
       var validation = new ClusterConfigValidationDTO();
@@ -156,13 +166,14 @@ private Properties convertProperties(Map<String, Object> propertiesMap) {
     return properties;
   }
 
-  private ReactiveFailover<PrometheusClientApi> prometheusStorageClient(ClustersProperties.Cluster cluster) {
+  private ReactiveFailover<PrometheusClientApi> prometheusStorageClient(
+      ClustersProperties.MetricsStorage storage, ClustersProperties.TruststoreConfig ssl) {
     WebClient webClient = new WebClientConfigurator()
-        .configureSsl(cluster.getSsl(), null)
+        .configureSsl(ssl, null)
         .configureBufferSize(webClientMaxBuffSize)
         .build();
     return ReactiveFailover.create(
-        parseUrlList(cluster.getMetrics().getStore().getPrometheus().getUrl()),
+        parseUrlList(storage.getPrometheus().getUrl()),
         url -> new PrometheusClientApi(new io.kafbat.ui.prometheus.ApiClient(webClient).setBasePath(url)),
         ReactiveFailover.CONNECTION_REFUSED_EXCEPTION_FILTER,
         "No live schemaRegistry instances available",
@@ -173,6 +184,12 @@ private ReactiveFailover<PrometheusClientApi> prometheusStorageClient(ClustersPr
   private boolean prometheusStorageConfigured(ClustersProperties.Cluster cluster) {
     return Optional.ofNullable(cluster.getMetrics())
         .flatMap(m -> Optional.ofNullable(m.getStore()))
+        .map(this::prometheusStorageConfigured)
+        .orElse(false);
+  }
+
+  private boolean prometheusStorageConfigured(ClustersProperties.MetricsStorage storage) {
+    return Optional.ofNullable(storage)
         .flatMap(s -> Optional.ofNullable(s.getPrometheus()))
         .map(p -> StringUtils.hasText(p.getUrl()))
         .orElse(false);

api/src/main/java/io/kafbat/ui/service/metrics/scrape/jmx/JmxMetricsFormatter.java

@@ -17,7 +17,7 @@
 public class JmxMetricsFormatter {
 
   // copied from https://github.com/prometheus/jmx_exporter/blob/b6b811b4aae994e812e902b26dd41f29364c0e2b/collector/src/main/java/io/prometheus/jmx/JmxMBeanPropertyCache.java#L15
-  private static final Pattern PROPERTY_PATTERN = Pattern.compile( // NOSONAR
+  private static final Pattern PROPERTY_PATTERN = Pattern.compile(// NOSONAR
       "([^,=:\\*\\?]+)=(\"(?:[^\\\\\"]*(?:\\\\.)?)*\"|[^,=:\"]*)" // NOSONAR
   );
 

api/src/main/java/io/kafbat/ui/service/metrics/scrape/prometheus/PrometheusTextFormatParser.java

@@ -38,7 +38,7 @@ public class PrometheusTextFormatParser {
 
   // Regex to capture metric name, optional labels, value, and optional timestamp.
   // Groups: 1=name, 2=labels (content), 3=value, 4=timestamp
-  private static final Pattern METRIC_LINE_PATTERN = Pattern.compile( // NOSONAR
+  private static final Pattern METRIC_LINE_PATTERN = Pattern.compile(// NOSONAR
       "^([a-zA-Z_:][a-zA-Z0-9_:]*)"            // Metric name
           + "(?:\\{([^}]*)})?"                     // Optional labels (content in group 2)
           + "\\s+"

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

@@ -205,8 +205,28 @@ paths:
     get:
       tags:
         - PrometheusExpose
-      summary: getAllMetrics
-      operationId: getAllMetrics
+      summary: exposeAllMetrics
+      operationId: exposeAllMetrics
+      responses:
+        200:
+          description: OK
+          content:
+            application/text:
+              schema:
+                type: string
+
+  /metrics/{clusterName}:
+    get:
+      tags:
+        - PrometheusExpose
+      summary: exposeClusterMetrics
+      operationId: exposeClusterMetrics
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
       responses:
         200:
           description: OK
@@ -2540,6 +2560,7 @@ components:
               - KAFKA_ACL_VIEW # get ACLs listing
               - KAFKA_ACL_EDIT # create & delete ACLs
               - CLIENT_QUOTA_MANAGEMENT
+              - GRAPHS_ENABLED
       required:
         - id
         - name
@@ -4494,6 +4515,8 @@ components:
                   type: integer
                 internalTopicPrefix:
                   type: string
+                defaultMetricsStorage:
+                  $ref: '#/components/schemas/ClusterMetricsStoreConfig'
                 clusters:
                   type: array
                   items:
@@ -4585,28 +4608,7 @@ components:
                           prometheusExpose:
                             type: boolean
                           store:
-                            type: object
-                            properties:
-                              prometheus:
-                                type: object
-                                properties:
-                                  url:
-                                    type: string
-                                  remoteWrite:
-                                    type: boolean
-                                  pushGatewayUrl:
-                                    type: string
-                                  pushGatewayUsername:
-                                    type: string
-                                  pushGatewayPassword:
-                                    type: string
-                                  pushGatewayJobName:
-                                    type: string
-                              kafka:
-                                type: object
-                                properties:
-                                  topic:
-                                    type: string
+                            $ref: '#/components/schemas/ClusterMetricsStoreConfig'
                       properties:
                         type: object
                         additionalProperties: true
@@ -4688,3 +4690,21 @@ components:
                             type: object
                             additionalProperties:
                               type: string
+    ClusterMetricsStoreConfig:
+      type: object
+      properties:
+        prometheus:
+          type: object
+          properties:
+            url:
+              type: string
+            remoteWrite:
+              type: boolean
+            pushGatewayUrl:
+              type: string
+            pushGatewayUsername:
+              type: string
+            pushGatewayPassword:
+              type: string
+            pushGatewayJobName:
+              type: string