[ISSUE #299] Add TPS and latency report in benchmark command (#300)

* feat(benchmark): initialize new module: benchmark 1. initialize new module: benchmark * feat(example): add latency metrics report in benchmark 1. add latency metrics report in benchmark
openmessaging · Jul 16, 2023 · 38c2377 · 38c2377
1 parent 93d6ff1
commit 38c2377
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 11 deletions.
diff --git a/example/pom.xml b/example/pom.xml
@@ -24,6 +24,11 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-simple</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>4.2.7</version>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>

diff --git a/...le/src/main/java/io/openmessaging/storage/dledger/example/register/RegisterBenchmark.java b/...le/src/main/java/io/openmessaging/storage/dledger/example/register/RegisterBenchmark.java
@@ -16,9 +16,15 @@
 
 package io.openmessaging.storage.dledger.example.register;
 
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Slf4jReporter;
 import io.openmessaging.storage.dledger.example.register.client.RegisterDLedgerClient;
 import io.openmessaging.storage.dledger.example.register.protocol.RegisterReadResponse;
 import io.openmessaging.storage.dledger.example.register.protocol.RegisterWriteResponse;
+import io.openmessaging.storage.dledger.utils.DLedgerUtils;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.time.StopWatch;
@@ -39,6 +45,13 @@ public class RegisterBenchmark {
 
     private BenchmarkType benchmarkType;
 
+    private Slf4jReporter reporter;
+    private Histogram latency;
+
+    private Meter tps;
+
+    private Counter errorCounter;
+
     public enum BenchmarkType {
         Write,
         Read,
@@ -51,6 +64,12 @@ public RegisterBenchmark(String group, String peers, int clientNum, long operati
         this.clientNum = clientNum;
         this.operationNumPerClient = operationNumPerClient;
         this.benchmarkType = benchmarkType;
+        MetricRegistry registry = new MetricRegistry();
+        reporter = Slf4jReporter.forRegistry(registry).convertRatesTo(TimeUnit.SECONDS)
+            .convertDurationsTo(TimeUnit.MILLISECONDS).build();
+        latency = registry.histogram("request_latency_ms");
+        tps = registry.meter("request_tps");
+        errorCounter = registry.counter("error_request_num");
     }
 
     public void start() throws Exception {
@@ -65,14 +84,13 @@ public void start() throws Exception {
             new Thread() {
                 @Override
                 public void run() {
-                    long success = 0;
-                    long failNum = 0;
                     RegisterDLedgerClient client = new RegisterDLedgerClient(group, peers);
                     client.startup();
                     try {
                         barrier.await();
-                        while (success < operationNumPerClient) {
+                        for (int i = 0; i < operationNumPerClient; i++) {
                             int code = 200;
+                            long start = System.currentTimeMillis();
                             if (benchmarkType == BenchmarkType.Read) {
                                 // Read
                                 RegisterReadResponse resp = client.read(13);
@@ -83,18 +101,16 @@ public void run() {
                                 code = resp.getCode();
                             }
                             if (code == 200) {
-                                success++;
+                                tps.mark();
+                                latency.update(DLedgerUtils.elapsed(start));
                             } else {
-                                failNum++;
+                                errorCounter.inc();
                             }
                         }
                         barrier.await();
-                        client.shutdown();
                     } catch (Exception e) {
                         logger.error("client {} error", operationType, e);
                     } finally {
-                        logger.info("client {} finished, need {} total: {}, success: {}, fail: {}",
-                            operationType, operationType, operationNumPerClient, success, failNum);
                         client.shutdown();
                     }
                 }
@@ -104,9 +120,9 @@ public void run() {
         StopWatch stopWatch = StopWatch.createStarted();
         barrier.await();
         final long cost = stopWatch.getTime(TimeUnit.MILLISECONDS);
-        final long tps = Math.round(totalOperation * 1000 / cost);
-        logger.info("Test type: {}, client num : {}, operation num per client: {}, total operation num: {}, cost: {}, tps: {}",
-            operationType, clientNum, operationNumPerClient, totalOperation, cost, tps);
+        logger.info("Test type: {}, client num : {}, operation num per client: {}, total operation num: {}, cost: {}}",
+            operationType, clientNum, operationNumPerClient, totalOperation, cost);
+        reporter.report();
     }
 
 }