Merge pull request #73 from jeffchao/master

Use timer for commit latency.
Netflix · Oct 30, 2020 · 0a1d193 · 0a1d193
2 parents 86c776b + b397373
commit 0a1d193
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 7 deletions.
diff --git a/...erg/src/main/java/io/mantisrx/connector/iceberg/sink/committer/IcebergCommitterStage.java b/...erg/src/main/java/io/mantisrx/connector/iceberg/sink/committer/IcebergCommitterStage.java
@@ -166,7 +166,7 @@ public Observable<Map<String, Object>> call(Observable<DataFile> source) {
                             long start = scheduler.now();
                             Map<String, Object> summary = committer.commit(dataFiles);
                             long now = scheduler.now();
-                            metrics.setGauge(CommitterMetrics.COMMIT_LATENCY_MSEC, now - start);
+                            metrics.record(CommitterMetrics.COMMIT_LATENCY_MSEC, now - start, TimeUnit.MILLISECONDS);
                             metrics.setGauge(CommitterMetrics.COMMIT_BATCH_SIZE, dataFiles.size());
                             return summary;
                         } catch (RuntimeException e) {

diff --git a/.../src/main/java/io/mantisrx/connector/iceberg/sink/committer/metrics/CommitterMetrics.java b/.../src/main/java/io/mantisrx/connector/iceberg/sink/committer/metrics/CommitterMetrics.java
@@ -16,10 +16,13 @@
 
 package io.mantisrx.connector.iceberg.sink.committer.metrics;
 
+import java.util.concurrent.TimeUnit;
+
 import io.mantisrx.common.metrics.Counter;
 import io.mantisrx.common.metrics.Gauge;
 import io.mantisrx.common.metrics.Metrics;
 import io.mantisrx.common.metrics.MetricsRegistry;
+import io.mantisrx.common.metrics.Timer;
 
 public class CommitterMetrics {
 
@@ -33,7 +36,7 @@ public class CommitterMetrics {
     private final Counter commitFailureCount;
 
     public static final String COMMIT_LATENCY_MSEC = "commitLatencyMsec";
-    private final Gauge commitLatencyMsec;
+    private final Timer commitLatencyMsec;
 
     public static final String COMMIT_BATCH_SIZE = "commitBatchSize";
     private final Gauge commitBatchSize;
@@ -44,7 +47,7 @@ public CommitterMetrics() {
                 .addCounter(INVOCATION_COUNT)
                 .addCounter(COMMIT_SUCCESS_COUNT)
                 .addCounter(COMMIT_FAILURE_COUNT)
-                .addGauge(COMMIT_LATENCY_MSEC)
+                .addTimer(COMMIT_LATENCY_MSEC)
                 .addGauge(COMMIT_BATCH_SIZE)
                 .build();
 
@@ -53,15 +56,12 @@ public CommitterMetrics() {
         invocationCount = metrics.getCounter(INVOCATION_COUNT);
         commitSuccessCount = metrics.getCounter(COMMIT_SUCCESS_COUNT);
         commitFailureCount = metrics.getCounter(COMMIT_FAILURE_COUNT);
-        commitLatencyMsec = metrics.getGauge(COMMIT_LATENCY_MSEC);
+        commitLatencyMsec = metrics.getTimer(COMMIT_LATENCY_MSEC);
         commitBatchSize = metrics.getGauge(COMMIT_BATCH_SIZE);
     }
 
     public void setGauge(final String metric, final long value) {
         switch (metric) {
-            case COMMIT_LATENCY_MSEC:
-                commitLatencyMsec.set(value);
-                break;
             case COMMIT_BATCH_SIZE:
                 commitBatchSize.set(value);
                 break;
@@ -70,6 +70,16 @@ public void setGauge(final String metric, final long value) {
         }
     }
 
+    public void record(final String metric, final long amount, TimeUnit unit) {
+        switch (metric) {
+            case COMMIT_LATENCY_MSEC:
+                commitLatencyMsec.record(amount, unit);
+                break;
+            default:
+                break;
+        }
+    }
+
     public void increment(final String metric) {
         switch (metric) {
             case INVOCATION_COUNT: