Merge pull request #870: [proxima-direct-ingest-server] add latency m…

…etrics to IngestService
datadrivencz · Jan 23, 2024 · 8150cd8 · 8150cd8
2 parents 220cb84 + 525c708
commit 8150cd8
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 6 deletions.
diff --git a/direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/IngestServer.java b/direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/IngestServer.java
@@ -134,6 +134,7 @@ static boolean ingestRequest(
       String uuid,
       Consumer<Rpc.Status> responseConsumer) {
 
+    long start = System.currentTimeMillis();
     AttributeDescriptor<?> attributeDesc = ingest.getAttributeDescriptor();
 
     OnlineAttributeWriter writer = getWriterForAttributeInTransform(direct, attributeDesc);
@@ -156,11 +157,12 @@ static boolean ingestRequest(
     }
 
     Metrics.COMMIT_LOG_APPEND.increment();
-    // write the ingest into the commit log and confirm to the client
+    // write the element into the commit log and confirm to the client
     log.debug("Writing {} to commit log {}", ingest, writer.getUri());
     writer.write(
         ingest,
         (s, exc) -> {
+          Metrics.INGEST_LATENCY.increment(System.currentTimeMillis() - start);
           if (s) {
             responseConsumer.accept(ok(uuid));
           } else {

diff --git a/direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/IngestService.java b/direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/IngestService.java
@@ -357,6 +357,7 @@ private StreamElement validateAndConvertToStreamElement(
     if (Strings.isNullOrEmpty(request.getKey())
         || Strings.isNullOrEmpty(request.getEntity())
         || Strings.isNullOrEmpty(request.getAttribute())) {
+
       consumer.accept(status(request.getUuid(), 400, "Missing required fields in input message"));
       return null;
     }
@@ -398,7 +399,6 @@ private StreamElement validateAndConvertToStreamElement(
 
   @Override
   public void ingest(Ingest request, StreamObserver<Status> responseObserver) {
-
     Metrics.INGEST_SINGLE.increment();
     processSingleIngest(
         request,
@@ -410,17 +410,14 @@ public void ingest(Ingest request, StreamObserver<Status> responseObserver) {
 
   @Override
   public StreamObserver<Ingest> ingestSingle(StreamObserver<Status> responseObserver) {
-
     return new IngestObserver(responseObserver);
   }
 
   @Override
   public StreamObserver<IngestBulk> ingestBulk(StreamObserver<StatusBulk> responseObserver) {
-
     // the responseObserver doesn't have to be synchronized in this
     // case, because the communication with the observer is done
     // in single flush thread
-
     return new IngestBulkObserver(responseObserver);
   }
 
@@ -431,13 +428,17 @@ public void commit(
 
     log.debug("Committing transaction {}", request.getTransactionId());
     try {
+      long start = System.currentTimeMillis();
       transactionContext
           .get(request.getTransactionId())
           .commit(
               (succ, exc) -> {
+                Metrics.INGEST_LATENCY.increment(System.currentTimeMillis() - start);
                 if (exc != null) {
                   log.warn(
                       "Error during committing transaction {}", request.getTransactionId(), exc);
+                } else {
+                  log.info("Transaction {} committed", request.getTransactionId());
                 }
                 responseObserver.onNext(
                     TransactionCommitResponse.newBuilder()
@@ -448,7 +449,6 @@ public void commit(
                         .build());
                 responseObserver.onCompleted();
               });
-      log.info("Transaction {} committed", request.getTransactionId());
     } catch (TransactionRejectedException e) {
       log.info("Transaction {} rejected.", request.getTransactionId());
       responseObserver.onNext(

diff --git a/direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/metrics/Metrics.java b/direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/metrics/Metrics.java
@@ -85,6 +85,9 @@ public class Metrics {
 
   public static final GaugeMetric LIVENESS = FACTORY.gauge(GROUP, "liveness");
 
+  public static final ApproxPercentileMetric INGEST_LATENCY =
+      FACTORY.percentile(GROUP, "ingest-latency", 60_000, 1_000);
+
   private static final Map<String, Pair<Boolean, GaugeMetric>> consumerMetrics =
       Collections.synchronizedMap(new HashMap<>());