wip

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

@@ -1,7 +1,7 @@
 package com.provectus.kafka.ui.model;
 
 import com.provectus.kafka.ui.service.ReactiveAdminClient;
-import com.provectus.kafka.ui.service.metrics.v2.scrape.inferred.ScrapedClusterState;
+import com.provectus.kafka.ui.service.metrics.v2.scrape.ScrapedClusterState;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -19,12 +19,13 @@ public class Statistics {
   List<ClusterFeature> features;
   ReactiveAdminClient.ClusterDescription clusterDescription;
   Metrics metrics;
+  ScrapedClusterState clusterState;
+
+  //TODO: to be removed -->>
   InternalLogDirStats logDirInfo;
   Map<String, TopicDescription> topicDescriptions;
   Map<String, List<ConfigEntry>> topicConfigs;
 
-  ScrapedClusterState clusterState;
-
   public static Statistics empty() {
     return builder()
         .status(ServerStatusDTO.OFFLINE)

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/ScrapedClusterState.java

@@ -0,0 +1,56 @@
+package com.provectus.kafka.ui.service.metrics.v2.scrape;
+
+import com.google.common.collect.Table;
+import com.provectus.kafka.ui.service.ReactiveAdminClient;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import lombok.Value;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.admin.ConsumerGroupDescription;
+import org.apache.kafka.clients.admin.TopicDescription;
+import reactor.core.publisher.Mono;
+
+@Value
+public class ScrapedClusterState {
+
+  record NodeState(SegmentStats segmentStats) {
+  }
+
+  record TopicState(
+      Instant scrapeTime,
+      String name,
+      List<ConfigEntry> configs,
+      TopicDescription description,
+      Map<Integer, Long> offsets,
+      SegmentStats segmentStats,
+      Map<Integer, SegmentStats> partitionsSegmentStats) {
+  }
+
+  record ConsumerGroupState(
+      Instant scrapeTime,
+      String group,
+      ConsumerGroupDescription description,
+      Table<String, Integer, Long> committedOffsets,
+      Map<String, Instant> lastTopicActivity) {
+  }
+
+  record SegmentStats(long segmentSize,
+                      int segmentsCount) {
+  }
+
+  Instant scrapeStartTime;
+  Map<Integer, NodeState> nodesStates;
+  Map<String, TopicState> topicStates;
+  Map<String, ConsumerGroupState> consumerGroupsStates;
+
+  public static ScrapedClusterState empty() {
+    //TODO impl
+    return null;
+  }
+
+  public static Mono<ScrapedClusterState> scrape(ReactiveAdminClient ac) {
+    return null;//TODO impl
+  }
+
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/ScrapedMetrics.java

@@ -3,14 +3,19 @@ package com.provectus.kafka.ui.service.metrics.v2.scrape;
 import io.prometheus.client.Collector.MetricFamilySamples;
 import java.util.Collection;
 
+import java.util.List;
 import java.util.stream.Stream;
 
 public interface ScrapedMetrics {
 
-  Stream<MetricFamilySamples> asStream();
-
   static ScrapedMetrics create(Collection<MetricFamilySamples> lst) {
     return lst::stream;
   }
 
+  static ScrapedMetrics empty() {
+    return create(List.of());
+  }
+
+  Stream<MetricFamilySamples> asStream();
+
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/Scrapping.java

@@ -0,0 +1,8 @@
+package com.provectus.kafka.ui.service.metrics.v2.scrape;
+
+public class Scrapping {
+
+  
+
+
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/inferred/InferredMetrics.java

@@ -2,19 +2,22 @@ package com.provectus.kafka.ui.service.metrics.v2.scrape.inferred;
 
 import static io.prometheus.client.Collector.*;
 
+import com.provectus.kafka.ui.service.metrics.v2.scrape.ScrapedClusterState;
 import com.provectus.kafka.ui.service.metrics.v2.scrape.ScrapedMetrics;
+import java.util.List;
 import java.util.stream.Stream;
 
 public class InferredMetrics implements ScrapedMetrics {
 
-  @Override
-  public Stream<MetricFamilySamples> asStream() {
-    return null;
+  private final List<MetricFamilySamples> metrics;
+
+  public InferredMetrics(List<MetricFamilySamples> metrics) {
+    this.metrics = metrics;
   }
 
-  public ScrapedClusterState clusterState() {
-    //todo: impl
-    return null;
+  @Override
+  public Stream<MetricFamilySamples> asStream() {
+    return metrics.stream();
   }
 
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/inferred/InferredMetricsScraper.java

@@ -0,0 +1,34 @@
+package com.provectus.kafka.ui.service.metrics.v2.scrape.inferred;
+
+import com.provectus.kafka.ui.service.metrics.v2.scrape.ScrapedClusterState;
+import com.provectus.kafka.ui.service.metrics.v2.scrape.Scraper;
+import java.util.List;
+import java.util.function.Supplier;
+import lombok.RequiredArgsConstructor;
+import reactor.core.publisher.Mono;
+
+@RequiredArgsConstructor
+public class InferredMetricsScraper implements Scraper<InferredMetrics> {
+
+  private final Supplier<ScrapedClusterState> currentStateSupplier;
+  private ScrapedClusterState prevState = null;
+
+  @Override
+  public synchronized Mono<InferredMetrics> scrape() {
+    if (prevState == null) {
+      prevState = currentStateSupplier.get();
+      return Mono.empty();
+    }
+    var newState = currentStateSupplier.get();
+    var inferred = infer(prevState, newState);
+    prevState = newState;
+    return Mono.just(inferred);
+  }
+
+  private static InferredMetrics infer(ScrapedClusterState prevState,
+                                       ScrapedClusterState newState) {
+    //TODO: impl
+    return new InferredMetrics(List.of());
+  }
+
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/inferred/InferredMetricsScrapper.java

@@ -1,22 +0,0 @@
-package com.provectus.kafka.ui.service.metrics.v2.scrape.inferred;
-
-import com.provectus.kafka.ui.service.ReactiveAdminClient;
-import com.provectus.kafka.ui.service.metrics.v2.scrape.Scraper;
-import reactor.core.publisher.Mono;
-
-public class InferredMetricsScrapper implements Scraper<InferredMetrics> {
-
-  private final ReactiveAdminClient adminClient;
-
-  private volatile ScrapedClusterState clusterState;
-
-  public InferredMetricsScrapper(ReactiveAdminClient adminClient) {
-    this.adminClient = adminClient;
-  }
-
-  @Override
-  public Mono<InferredMetrics> scrape() {
-    return null;
-  }
-
-}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/inferred/ScrapedClusterState.java

@@ -1,19 +0,0 @@
-package com.provectus.kafka.ui.service.metrics.v2.scrape.inferred;
-
-import com.provectus.kafka.ui.service.metrics.v2.scrape.inferred.states.ConsumerGroupsState;
-import com.provectus.kafka.ui.service.metrics.v2.scrape.inferred.states.TopicsState;
-import java.time.Instant;
-import lombok.Value;
-
-@Value
-public class ScrapedClusterState {
-
-  Instant scrapeStart;
-  TopicsState topicsState;
-  ConsumerGroupsState consumerGroupsState;
-
-  public static ScrapedClusterState empty() {
-    return new ScrapedClusterState(null, null, null);
-  }
-
-}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/inferred/states/ConsumerGroupsState.java

@@ -1,4 +0,0 @@
-package com.provectus.kafka.ui.service.metrics.v2.scrape.inferred.states;
-
-public class ConsumerGroupsState {
-}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/metrics/v2/scrape/inferred/states/TopicsState.java

@@ -1,4 +0,0 @@
-package com.provectus.kafka.ui.service.metrics.v2.scrape.inferred.states;
-
-public class TopicsState {
-}