Show real outgoing bytes

app/src/main/java/org/astraea/app/performance/Metrics.java

@@ -0,0 +1,112 @@
+/*
+ * 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.app.performance;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
+
+/** Used to record statistics. This is thread safe. */
+public class Metrics implements BiConsumer<Long, Integer> {
+  private double avgLatency;
+  private long num;
+  private long max;
+  private long min;
+  private long bytes;
+  private long currentBytes;
+  private final Map<MutableMetric<Double>, Long> realBytes = new HashMap<>();
+
+  public Metrics() {
+    avgLatency = 0;
+    num = 0;
+    max = 0;
+    min = Long.MAX_VALUE;
+    bytes = 0;
+    currentBytes = 0;
+  }
+
+  /** Simultaneously add latency and bytes. */
+  @Override
+  public synchronized void accept(Long latency, Integer bytes) {
+    ++num;
+    putLatency(latency);
+    addBytes(bytes);
+  }
+
+  /** Add a new value to latency metric. */
+  private synchronized void putLatency(long latency) {
+    min = Math.min(min, latency);
+    max = Math.max(max, latency);
+    avgLatency += (((double) latency) - avgLatency) / (double) num;
+  }
+  /** Add a new value to bytes. */
+  private synchronized void addBytes(long bytes) {
+    this.currentBytes += bytes;
+    this.bytes += bytes;
+  }
+
+  /** @return Get the number of latency put. */
+  public synchronized long num() {
+    return num;
+  }
+  /** @return Get the maximum of latency put. */
+  public synchronized long max() {
+    return max;
+  }
+  /** @return Get the minimum of latency put. */
+  public synchronized long min() {
+    return min;
+  }
+  /** @return Get the average latency. */
+  public synchronized double avgLatency() {
+    return avgLatency;
+  }
+
+  /** @return total send/received bytes */
+  public synchronized long bytes() {
+    return bytes;
+  }
+
+  public synchronized long clearAndGetCurrentBytes() {
+    var ans = currentBytes;
+    currentBytes = 0;
+    return ans;
+  }
+
+  public void putRealBytesMetric(MutableMetric<Double> realBytesMetric) {
+    this.realBytes.put(realBytesMetric, 0L);
+  }
+
+  /**
+   * Get current real bytes since last call. Warning: Real bytes is recorded in `realBytes`. Please
+   * #setRealBytesMetric before get #currentRealBytes
+   */
+  public synchronized long currentRealBytes() {
+    AtomicLong all = new AtomicLong(0L);
+
+    // Get sum of all instances and update
+    realBytes.replaceAll(
+        (metric, last) -> {
+          var totalRealBytes = (long) metric.metricValue().doubleValue();
+          all.addAndGet(totalRealBytes - last);
+          return totalRealBytes;
+        });
+
+    return all.get();
+  }
+}

app/src/main/java/org/astraea/app/performance/MutableMetric.java

@@ -0,0 +1,58 @@
+/*
+ * 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.app.performance;
+
+import java.util.function.Supplier;
+
+/**
+ * A metric that the value may change over time. The changes may be caused by other thread, or
+ * change when metricValue() is called.
+ */
+public interface MutableMetric<T> {
+  /** Default return type is double. */
+  static MutableMetric<Double> of(org.apache.kafka.common.Metric kafkaMetric) {
+    return new MutableMetric<>() {
+      @Override
+      public String metricName() {
+        return kafkaMetric.metricName().name();
+      }
+
+      @Override
+      public Double metricValue() {
+        return Double.parseDouble(kafkaMetric.metricValue().toString());
+      }
+    };
+  }
+
+  static <T> MutableMetric<T> create(String name, Supplier<T> metricValue) {
+    return new MutableMetric<>() {
+      @Override
+      public String metricName() {
+        return name;
+      }
+
+      @Override
+      public T metricValue() {
+        return metricValue.get();
+      }
+    };
+  }
+
+  String metricName();
+
+  T metricValue();
+}

app/src/main/java/org/astraea/app/performance/Performance.java

@@ -47,6 +47,9 @@
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Isolation;
+import org.astraea.app.metrics.client.consumer.ConsumerMetrics;
+import org.astraea.app.metrics.client.producer.ProducerMetrics;
+import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.producer.Producer;
 
 /**
@@ -148,7 +151,14 @@ public static String execute(final Argument param) throws InterruptedException,
             .map(ConsumerThread::report)
             .collect(Collectors.toUnmodifiableList());
 
-    var tracker = TrackerThread.create(producerReports, consumerReports, param.exeTime);
+    // The receivers (that is, producerReceiver and consumerReceiver) will fetch mbean once per
+    // second. The interval should be less than 1 second. Let the receiver fetch new data.
+    var beanCollector = BeanCollector.builder().interval(Duration.ofMillis(500)).build();
+    var producerReceiver = beanCollector.register().local().fetcher(ProducerMetrics::nodes).build();
+    var consumerReceiver = beanCollector.register().local().fetcher(ConsumerMetrics::nodes).build();
+    var tracker =
+        TrackerThread.create(
+            producerReports, consumerReports, producerReceiver, consumerReceiver, param.exeTime);
 
     Optional<Runnable> fileWriter =
         param.CSVPath == null
@@ -191,6 +201,8 @@ public static String execute(final Argument param) throws InterruptedException,
 
     consumerThreads.forEach(AbstractThread::waitForDone);
     tracker.waitForDone();
+    producerReceiver.close();
+    consumerReceiver.close();
     return param.topic;
   }
 

app/src/main/java/org/astraea/app/performance/TrackerThread.java

@@ -18,20 +18,29 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
+import java.util.function.ToDoubleFunction;
 import org.astraea.app.common.DataSize;
 import org.astraea.app.common.Utils;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.client.HasNodeMetrics;
+import org.astraea.app.metrics.collector.Receiver;
 
 /** Print out the given metrics. */
 public interface TrackerThread extends AbstractThread {
 
   static TrackerThread create(
-      List<Report> producerReports, List<Report> consumerReports, ExeTime exeTime) {
+      List<Report> producerReports,
+      List<Report> consumerReports,
+      Receiver producerReceiver,
+      Receiver consumerReceiver,
+      ExeTime exeTime) {
     var start = System.currentTimeMillis() - Duration.ofSeconds(1).toMillis();
 
     Function<Result, Boolean> logProducers =
@@ -56,6 +65,11 @@ static TrackerThread create(
               "  average throughput: %.3f MB/second%n", result.averageBytes(duration));
           System.out.printf(
               "  current throughput: %s/second%n", DataSize.Byte.of(result.totalCurrentBytes()));
+          System.out.printf(
+              "  total real out-going: %s/second%n",
+              DataSize.Byte.of(
+                  sumOfAttribute(producerReceiver.current(), HasNodeMetrics::outgoingByteTotal)
+                      .longValue()));
           System.out.println("  publish max latency: " + result.maxLatency + " ms");
           System.out.println("  publish mim latency: " + result.minLatency + " ms");
           for (int i = 0; i < result.bytes.size(); ++i) {
@@ -90,6 +104,11 @@ static TrackerThread create(
               "  average throughput: %.3f MB/second%n", result.averageBytes(duration));
           System.out.printf(
               "  current throughput: %s/second%n", DataSize.Byte.of(result.totalCurrentBytes()));
+          System.out.printf(
+              "  total real in-coming: %s/second%n",
+              DataSize.Byte.of(
+                  sumOfAttribute(consumerReceiver.current(), HasNodeMetrics::incomingByteTotal)
+                      .longValue()));
           System.out.println("  end-to-end max latency: " + result.maxLatency + " ms");
           System.out.println("  end-to-end mim latency: " + result.minLatency + " ms");
           for (int i = 0; i < result.bytes.size(); ++i) {
@@ -159,6 +178,20 @@ public void close() {
     };
   }
 
+  /**
+   * @param mbeans mBeans fetched by the receivers
+   * @return sum of given attribute of all beans which is instance of HasNodeMetrics.
+   */
+  static Double sumOfAttribute(
+      Collection<HasBeanObject> mbeans, ToDoubleFunction<HasNodeMetrics> targetAttribute) {
+    return mbeans.stream()
+        .filter(mbean -> mbean instanceof HasNodeMetrics)
+        .map(mbean -> (HasNodeMetrics) mbean)
+        .mapToDouble(targetAttribute)
+        .filter(d -> !Double.isNaN(d))
+        .sum();
+  }
+
   static double avg(Duration duration, long value) {
     return duration.toSeconds() <= 0
         ? 0

app/src/main/java/org/astraea/app/producer/Builder.java

@@ -34,6 +34,7 @@
 import org.apache.kafka.common.errors.OutOfOrderSequenceException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.astraea.app.admin.Compression;
+import org.astraea.app.performance.MutableMetric;
 
 public class Builder<Key, Value> {
   private final Map<String, Object> configs = new HashMap<>();
@@ -148,6 +149,17 @@ private BaseProducer(org.apache.kafka.clients.producer.Producer<Key, Value> kafk
       this.kafkaProducer = kafkaProducer;
     }
 
+    @Override
+    public MutableMetric<Double> getMetric(String metricName) {
+      var kafkaMetric =
+          kafkaProducer.metrics().entrySet().stream()
+              .filter(e -> e.getKey().name().equals(metricName))
+              .findFirst()
+              .orElseThrow()
+              .getValue();
+      return MutableMetric.of(kafkaMetric);
+    }
+
     @Override
     public void flush() {
       kafkaProducer.flush();

app/src/main/java/org/astraea/app/producer/Producer.java

@@ -19,6 +19,7 @@
 import java.util.Collection;
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
+import org.astraea.app.performance.MutableMetric;
 
 /** An interface for sending records. */
 public interface Producer<Key, Value> extends AutoCloseable {
@@ -38,6 +39,11 @@ public interface Producer<Key, Value> extends AutoCloseable {
 
   void close();
 
+  /** Get a kafkaMetric by name. */
+  default MutableMetric<Double> getMetric(String metricName) {
+    return MutableMetric.create("null", () -> 0.0);
+  }
+
   /** @return true if the producer supports transactional. */
   default boolean transactional() {
     return transactionId().isPresent();

app/src/test/java/org/astraea/app/performance/PerformanceTest.java

@@ -18,7 +18,6 @@
 
 import com.beust.jcommander.ParameterException;
 import java.util.concurrent.CountDownLatch;
-import java.util.function.BiConsumer;
 import org.astraea.app.argument.Argument;
 import org.astraea.app.consumer.Isolation;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -33,8 +32,6 @@ void testTransactionalProducer() {
     String[] arguments1 = {
       "--bootstrap.servers", bootstrapServers(), "--topic", topic, "--transaction.size", "2"
     };
-    var latch = new CountDownLatch(1);
-    BiConsumer<Long, Integer> observer = (x, y) -> latch.countDown();
     var argument = Argument.parse(new Performance.Argument(), arguments1);
     try (var producer = argument.createProducer()) {
       Assertions.assertTrue(producer.transactional());
@@ -48,7 +45,13 @@ void testProducerExecutor() throws InterruptedException {
       "--bootstrap.servers", bootstrapServers(), "--topic", topic, "--compression", "gzip"
     };
     var latch = new CountDownLatch(1);
-    BiConsumer<Long, Integer> observer = (x, y) -> latch.countDown();
+    Metrics observer =
+        new Metrics() {
+          @Override
+          public void accept(Long x, Integer y) {
+            latch.countDown();
+          }
+        };
     var argument = Argument.parse(new Performance.Argument(), arguments1);
     try (var producer = argument.createProducer()) {
       Assertions.assertFalse(producer.transactional());