[COMMON] Implementation of MetricSeriesBuilder

common/src/main/java/org/astraea/common/metrics/MetricSeriesBuilder.java

@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.common.metrics;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiFunction;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.astraea.common.admin.ClusterBean;
+import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.admin.NodeInfo;
+import org.astraea.common.admin.Replica;
+import org.astraea.common.admin.TopicPartition;
+import org.astraea.common.function.Bi3Function;
+
+/**
+ * A utility for generating a series of metric objects, where the measured metric value might be
+ * highly correlated to specific variables. For example broker id, calendar time, or unknown noise.
+ * This class offers a way to construct large-scale metric sources of a fake cluster, which will be
+ * useful for testing and experiment purposes.
+ */
+public interface MetricSeriesBuilder {
+
+  static MetricSeriesBuilder builder() {
+    return new MetricSeriesBuilderImpl();
+  }
+
+  MetricSeriesBuilder cluster(ClusterInfo clusterInfo);
+
+  MetricSeriesBuilder timeRange(LocalDateTime firstMetricTime, Duration duration);
+
+  MetricSeriesBuilder sampleInterval(Duration interval);
+
+  MetricSeriesBuilder seriesByBroker(
+      BiFunction<LocalDateTime, Integer, Stream<? extends HasBeanObject>> seriesGenerator);
+
+  MetricSeriesBuilder seriesByBrokerTopic(
+      Bi3Function<LocalDateTime, Integer, String, HasBeanObject> seriesGenerator);
+
+  MetricSeriesBuilder seriesByBrokerPartition(
+      Bi3Function<LocalDateTime, Integer, TopicPartition, HasBeanObject> seriesGenerator);
+
+  MetricSeriesBuilder seriesByBrokerReplica(
+      Bi3Function<LocalDateTime, Integer, Replica, HasBeanObject> seriesGenerator);
+
+  ClusterBean build();
+
+  final class MetricSeriesBuilderImpl implements MetricSeriesBuilder {
+
+    private final List<Supplier<Map<Integer, Stream<? extends HasBeanObject>>>> series =
+        new ArrayList<>();
+
+    private ClusterInfo clusterInfo;
+    private LocalDateTime timeStart;
+    private Duration timeRange;
+    private Duration sampleInterval = Duration.ofSeconds(1);
+
+    @Override
+    public MetricSeriesBuilder cluster(ClusterInfo clusterInfo) {
+      this.clusterInfo = Objects.requireNonNull(clusterInfo);
+      return this;
+    }
+
+    @Override
+    public MetricSeriesBuilder timeRange(LocalDateTime firstMetricTime, Duration duration) {
+      this.timeStart = firstMetricTime;
+      this.timeRange = duration;
+      return this;
+    }
+
+    @Override
+    public MetricSeriesBuilder sampleInterval(Duration interval) {
+      if (interval.isNegative() || interval.isZero())
+        throw new IllegalArgumentException("The sample interval must be positive");
+      this.sampleInterval = interval;
+      return this;
+    }
+
+    @Override
+    public MetricSeriesBuilder seriesByBroker(
+        BiFunction<LocalDateTime, Integer, Stream<? extends HasBeanObject>> seriesGenerator) {
+      final var cluster = clusterInfo;
+      final var start = timeStart;
+      final var end = timeStart.plus(timeRange);
+      final var interval = sampleInterval;
+      this.series.add(
+          () ->
+              Stream.iterate(
+                      start, (t) -> t.isBefore(end) || t.isEqual(end), (t) -> t.plus(interval))
+                  .flatMap(
+                      time ->
+                          cluster.nodes().stream()
+                              .map(
+                                  node ->
+                                      Map.entry(node.id(), seriesGenerator.apply(time, node.id()))))
+                  .collect(
+                      Collectors.toUnmodifiableMap(
+                          Map.Entry::getKey, Map.Entry::getValue, Stream::concat)));
+      return this;
+    }
+
+    @Override
+    public MetricSeriesBuilder seriesByBrokerTopic(
+        Bi3Function<LocalDateTime, Integer, String, HasBeanObject> seriesGenerator) {
+      final var cluster = clusterInfo;
+      final var start = timeStart;
+      final var end = timeStart.plus(timeRange);
+      final var interval = sampleInterval;
+      this.series.add(
+          () ->
+              cluster.nodes().stream()
+                  .collect(
+                      Collectors.toUnmodifiableMap(
+                          NodeInfo::id,
+                          node ->
+                              Stream.iterate(
+                                      start,
+                                      (t) -> t.isBefore(end) || t.isEqual(end),
+                                      (t) -> t.plus(interval))
+                                  .flatMap(
+                                      time ->
+                                          cluster
+                                              .replicaStream(node.id())
+                                              .map(Replica::topic)
+                                              .distinct()
+                                              .map(
+                                                  topic ->
+                                                      seriesGenerator.apply(
+                                                          time, node.id(), topic))))));
+      return this;
+    }
+
+    @Override
+    public MetricSeriesBuilder seriesByBrokerPartition(
+        Bi3Function<LocalDateTime, Integer, TopicPartition, HasBeanObject> seriesGenerator) {
+      final var cluster = clusterInfo;
+      final var start = timeStart;
+      final var end = timeStart.plus(timeRange);
+      final var interval = sampleInterval;
+      this.series.add(
+          () ->
+              cluster.nodes().stream()
+                  .collect(
+                      Collectors.toUnmodifiableMap(
+                          NodeInfo::id,
+                          node ->
+                              Stream.iterate(
+                                      start,
+                                      (t) -> t.isBefore(end) || t.isEqual(end),
+                                      (t) -> t.plus(interval))
+                                  .flatMap(
+                                      time ->
+                                          cluster
+                                              .replicaStream(node.id())
+                                              .map(Replica::topicPartition)
+                                              .map(
+                                                  partition ->
+                                                      seriesGenerator.apply(
+                                                          time, node.id(), partition))))));
+      return this;
+    }
+
+    @Override
+    public MetricSeriesBuilder seriesByBrokerReplica(
+        Bi3Function<LocalDateTime, Integer, Replica, HasBeanObject> seriesGenerator) {
+      final var cluster = clusterInfo;
+      final var start = timeStart;
+      final var end = timeStart.plus(timeRange);
+      final var interval = sampleInterval;
+      this.series.add(
+          () ->
+              cluster.nodes().stream()
+                  .collect(
+                      Collectors.toUnmodifiableMap(
+                          NodeInfo::id,
+                          node ->
+                              Stream.iterate(
+                                      start,
+                                      (t) -> t.isBefore(end) || t.isEqual(end),
+                                      (t) -> t.plus(interval))
+                                  .flatMap(
+                                      time ->
+                                          cluster
+                                              .replicaStream(node.id())
+                                              .map(
+                                                  replica ->
+                                                      seriesGenerator.apply(
+                                                          time, node.id(), replica))))));
+      return this;
+    }
+
+    @Override
+    public ClusterBean build() {
+      Map<Integer, Collection<HasBeanObject>> allMetrics =
+          this.series.stream()
+              .map(Supplier::get)
+              .flatMap(metrics -> metrics.entrySet().stream())
+              .collect(
+                  Collectors.groupingBy(
+                      Map.Entry::getKey,
+                      Collectors.flatMapping(
+                          Map.Entry::getValue, Collectors.toCollection(ArrayList::new))));
+      return ClusterBean.of(allMetrics);
+    }
+  }
+}

common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java

@@ -18,6 +18,7 @@
 
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import org.astraea.common.EnumInfo;
 import org.astraea.common.metrics.BeanObject;
@@ -159,6 +160,48 @@ public List<Gauge> fetch(MBeanClient mBeanClient) {
           .collect(Collectors.toUnmodifiableList());
     }
 
+    public Builder builder() {
+      return new Builder(this);
+    }
+
+    public static class Builder {
+      private final LogMetrics.Log metric;
+      private String topic;
+      private int partition;
+      private long value;
+
+      public Builder(Log metric) {
+        this.metric = metric;
+      }
+
+      public Builder topic(String topic) {
+        this.topic = topic;
+        return this;
+      }
+
+      public Builder partition(int partition) {
+        this.partition = partition;
+        return this;
+      }
+
+      public Builder logSize(long value) {
+        this.value = value;
+        return this;
+      }
+
+      public Gauge build() {
+        return new Gauge(
+            new BeanObject(
+                LogMetrics.DOMAIN_NAME,
+                Map.ofEntries(
+                    Map.entry("type", LOG_TYPE),
+                    Map.entry("name", metric.metricName()),
+                    Map.entry("topic", topic),
+                    Map.entry("partition", String.valueOf(partition))),
+                Map.of("Value", value)));
+      }
+    }
+
     public static class Gauge implements HasGauge<Long> {
       private final BeanObject beanObject;
 

common/src/main/java/org/astraea/common/metrics/broker/ServerMetrics.java

@@ -17,9 +17,12 @@
 package org.astraea.common.metrics.broker;
 
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.astraea.common.EnumInfo;
 import org.astraea.common.metrics.AppInfo;
@@ -395,6 +398,73 @@ public List<Topic.Meter> fetch(MBeanClient mBeanClient) {
           .collect(Collectors.toList());
     }
 
+    public Builder builder() {
+      return new Builder(this);
+    }
+
+    public static class Builder {
+
+      private final ServerMetrics.Topic metric;
+      private String topic;
+      private long time;
+      private final Map<String, Object> attributes = new HashMap<>();
+
+      public Builder(Topic metric) {
+        this.metric = metric;
+      }
+
+      public Builder topic(String topic) {
+        this.topic = topic;
+        return this;
+      }
+
+      public Builder time(long time) {
+        this.time = time;
+        return this;
+      }
+
+      public Builder meanRate(double value) {
+        this.attributes.put("MeanRate", value);
+        return this;
+      }
+
+      public Builder oneMinuteRate(double value) {
+        this.attributes.put("OneMinuteRate", value);
+        return this;
+      }
+
+      public Builder fiveMinuteRate(double value) {
+        this.attributes.put("FiveMinuteRate", value);
+        return this;
+      }
+
+      public Builder fifteenMinuteRate(double value) {
+        this.attributes.put("FifteenMinuteRate", value);
+        return this;
+      }
+
+      public Builder rateUnit(TimeUnit timeUnit) {
+        this.attributes.put("RateUnit", timeUnit);
+        return this;
+      }
+
+      public Builder count(long count) {
+        this.attributes.put("Count", count);
+        return this;
+      }
+
+      public Meter build() {
+        return new Meter(
+            new BeanObject(
+                ServerMetrics.DOMAIN_NAME,
+                Map.ofEntries(
+                    Map.entry("type", "BrokerTopicMetrics"),
+                    Map.entry("topic", topic),
+                    Map.entry("name", metric.metricName())),
+                Map.copyOf(attributes)));
+      }
+    }
+
     public static class Meter implements HasMeter {
 
       private final BeanObject beanObject;