[log] add server side checksum calculation time (#2523)

Author: jingy-li

clients/da-vinci-client/src/main/java/com/linkedin/davinci/blobtransfer/NettyP2PBlobTransferManager.java

@@ -215,11 +215,9 @@ private void processPeersSequentially(
 
         // Attempt to fetch the blob from the current peer asynchronously
         LOGGER.info(
-            "Attempting to connect to host: {} for store {} version {} partition {} table format {}",
+            "Attempting to connect to host: {} for replica {} table format {}",
             chosenHost,
-            storeName,
-            version,
-            partition,
+            replicaId,
             tableFormat);
 
         CompletionStage<InputStream> perHostTransferFuture =
@@ -315,10 +313,8 @@ private void updateBlobTransferFileReceiveStats(double transferTime, String stor
       aggVersionedBlobTransferStats.recordBlobTransferFileReceiveThroughput(storeName, version, throughput);
     } catch (Exception e) {
       LOGGER.error(
-          "Failed to update updateBlobTransferFileReceiveStats for store {} version {} partition {}",
-          storeName,
-          version,
-          partition,
+          "Failed to update updateBlobTransferFileReceiveStats for replica {}",
+          Utils.getReplicaId(Version.composeKafkaTopic(storeName, version), partition),
           e);
     }
   }
@@ -334,25 +330,18 @@ private void updateBlobTransferFileReceiveStats(double transferTime, String stor
   private List<String> getConnectableHosts(List<String> discoverPeers, String storeName, int version, int partition) {
     // Extract unique hosts from the discovered peers
     Set<String> uniquePeers = discoverPeers.stream().map(peer -> peer.split("_")[0]).collect(Collectors.toSet());
+    String replicaId = Utils.getReplicaId(Version.composeKafkaTopic(storeName, version), partition);
 
-    LOGGER.info(
-        "Discovered {} unique peers store {} version {} partition {}, peers are {}",
-        uniquePeers.size(),
-        storeName,
-        version,
-        partition,
-        uniquePeers);
+    LOGGER.info("Discovered {} unique peers for replica {}, peers are {}", uniquePeers.size(), replicaId, uniquePeers);
 
     // Get the connectable hosts for this store, version, and partition
     Set<String> connectablePeers =
         nettyClient.getConnectableHosts((HashSet<String>) uniquePeers, storeName, version, partition);
 
     LOGGER.info(
-        "Total {} unique connectable peers for store {} version {} partition {}, peers are {}",
+        "Total {} unique connectable peers for replica {}, peers are {}",
         connectablePeers.size(),
-        storeName,
-        version,
-        partition,
+        replicaId,
         connectablePeers);
 
     // Change to list and shuffle the list

clients/da-vinci-client/src/main/java/com/linkedin/davinci/blobtransfer/client/NettyFileTransferClient.java

@@ -161,6 +161,7 @@ public Set<String> getConnectableHosts(
       int version,
       int partition) {
     List<CompletableFuture<String>> futures = new ArrayList<>();
+    String replicaId = Utils.getReplicaId(storeName, version, partition);
 
     // 1. Purge the host connectivity records that are stale
     purgeStaleConnectivityRecords(unconnectableHostsToTimestamp);
@@ -200,23 +201,12 @@ public Set<String> getConnectableHosts(
             });
             return host;
           } else {
-            LOGGER.warn(
-                "Failed to connect to host: {} for store {} version {} partition {}",
-                host,
-                storeName,
-                version,
-                partition);
+            LOGGER.warn("Failed to connect to host: {} for replica {}", host, replicaId);
             unconnectableHostsToTimestamp.put(host, System.currentTimeMillis());
             return null;
           }
         } catch (Exception e) {
-          LOGGER.warn(
-              "Failed to connect to host: {} for store {} version {} partition {}",
-              host,
-              storeName,
-              version,
-              partition,
-              e);
+          LOGGER.warn("Failed to connect to host: {} for replica {}", host, replicaId, e);
           unconnectableHostsToTimestamp.put(host, System.currentTimeMillis());
           return null;
         }
@@ -311,9 +301,8 @@ public CompletionStage<InputStream> get(
             .completeExceptionally(
                 new VenicePeersConnectionException(
                     "The host " + host
-                        + " channel already have P2PFileTransferClientHandler/P2PMetadataTransferHandler for "
-                        + storeName + " version " + version + " partition " + partition + " table format "
-                        + requestedTableFormat));
+                        + " channel already have P2PFileTransferClientHandler/P2PMetadataTransferHandler for replica "
+                        + replicaId + " table format " + requestedTableFormat));
         return perHostTransferFuture;
       }
 
@@ -358,10 +347,8 @@ public CompletionStage<InputStream> get(
       connectTimeoutScheduler.schedule(() -> {
         if (!perHostTransferFuture.toCompletableFuture().isDone()) {
           String errorMsg = String.format(
-              "Request timed out for store %s version %d partition %d table format %s from host %s after %d minutes",
-              storeName,
-              version,
-              partition,
+              "Request timed out for replica %s table format %s from host %s after %d minutes",
+              replicaId,
               requestedTableFormat,
               host,
               blobReceiveTimeoutInMin);

clients/da-vinci-client/src/main/java/com/linkedin/davinci/blobtransfer/server/P2PFileTransferServerHandler.java

@@ -280,7 +280,14 @@ private void sendFile(
     RandomAccessFile raf = new RandomAccessFile(file, "r");
     ChannelFuture sendFileFuture;
     long length = raf.length();
+
+    long checksumStartTime = System.currentTimeMillis();
     String fileChecksum = BlobTransferUtils.generateFileChecksum(file.toPath());
+    LOGGER.info(
+        "Checksum calculation for file: {} for replica {} took {} ms.",
+        file.getName(),
+        replicaInfo,
+        System.currentTimeMillis() - checksumStartTime);
 
     HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
     response.headers().set(HttpHeaderNames.CONTENT_LENGTH, length);