add latencies to logs

Author: zhongkechen

examples/src/main/resources/log4j2.xml

@@ -7,6 +7,8 @@
     </Appenders>
 
     <Loggers>
+        <Logger name="com.amazonaws.lambda.durable" level="INFO"/>
+
         <!-- AWS SDK logs at WARN -->
         <Logger name="software.amazon.awssdk" level="WARN" />
 

sdk/src/main/java/com/amazonaws/lambda/durable/client/LambdaDurableFunctionsClient.java

@@ -4,8 +4,6 @@
 
 import java.util.List;
 import java.util.Objects;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.lambda.LambdaClient;
 import software.amazon.awssdk.services.lambda.model.CheckpointDurableExecutionRequest;
 import software.amazon.awssdk.services.lambda.model.CheckpointDurableExecutionResponse;
@@ -15,7 +13,6 @@
 
 public class LambdaDurableFunctionsClient implements DurableExecutionClient {
 
-    private static final Logger logger = LoggerFactory.getLogger(LambdaDurableFunctionsClient.class);
     private final LambdaClient lambdaClient;
 
     /**
@@ -35,7 +32,6 @@ public CheckpointDurableExecutionResponse checkpoint(String arn, String token, L
                 .checkpointToken(token)
                 .updates(updates)
                 .build();
-        logger.debug("Calling durable backend with {} updates: {}", updates.size(), request);
 
         return lambdaClient.checkpointDurableExecution(request);
     }

sdk/src/main/java/com/amazonaws/lambda/durable/execution/CheckpointBatcher.java

@@ -110,9 +110,13 @@ public List<Operation> fetchAllPages(CheckpointUpdatedExecutionState checkpointU
         }
         var nextMarker = checkpointUpdatedExecutionState.nextMarker();
         while (nextMarker != null && !nextMarker.isEmpty()) {
+            var startTime = System.nanoTime();
             var response = config.getDurableExecutionClient()
                     .getExecutionState(durableExecutionArn, checkpointToken, nextMarker);
-            logger.debug("Durable API getExecutionState called: {}.", response);
+            logger.debug(
+                    "Durable getExecutionState API called (latency={}ns): {}.",
+                    System.nanoTime() - startTime,
+                    response);
             operations.addAll(response.operations());
             nextMarker = response.nextMarker();
         }
@@ -130,9 +134,10 @@ private void doBatchAction(List<OperationUpdate> updates) {
                 return;
             }
 
-            logger.debug("Calling durable API checkpointDurableExecution with {} updates", request.size());
+            var startTime = System.nanoTime();
+            logger.debug("Calling durable checkpoint API with {} updates: {}", updates.size(), request);
             var response = config.getDurableExecutionClient().checkpoint(durableExecutionArn, checkpointToken, request);
-            logger.debug("Durable API checkpointDurableExecution called: {}.", response);
+            logger.debug("Durable checkpoint API called (latency={}ns): {}.", System.nanoTime() - startTime, response);
 
             // Notify callback of completion
             // TODO: sam local backend returns no new execution state when called with zero
@@ -144,16 +149,26 @@ private void doBatchAction(List<OperationUpdate> updates) {
                 // fetch all pages of operations
                 var operations = fetchAllPages(response.newExecutionState());
 
+                var processStartTime = System.nanoTime();
+                int completedFutures = 0;
+                logger.debug(
+                        "Processing {} operations. ({} pending pollers)", operations.size(), pollingFutures.size());
                 // call the callback
                 callback.accept(operations);
 
                 // complete the registered pollingFutures
                 for (var operation : operations) {
                     var pollers = pollingFutures.remove(operation.id());
                     if (pollers != null) {
+                        completedFutures += pollers.size();
                         pollers.forEach(poller -> poller.complete(operation));
                     }
                 }
+                logger.debug(
+                        "{} operations processed and {} pollers completed (latency={}ns). ",
+                        operations.size(),
+                        completedFutures,
+                        System.nanoTime() - processStartTime);
             }
         }
     }