Improve volume upload summary logs

Author: lentitude2tk

sdk-bulkwriter/src/main/java/io/milvus/bulkwriter/VolumeFileManager.java

@@ -44,6 +44,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.*;
@@ -79,20 +80,29 @@ public CompletableFuture<UploadFilesResult> uploadFilesAsync(UploadFilesRequest
         String localDirOrFilePath = request.getSourceFilePath();
         Pair<List<String>, Long> localPathPair = FileUtils.processLocalPath(localDirOrFilePath);
         String volumePath = convertDirPath(request.getTargetVolumePath());
-
-        UploadContext uploadContext = new UploadContext(refreshVolumeAndClient(volumePath));
-        VolumeSession initialSession = uploadContext.currentSession();
-        initValidator(localPathPair, initialSession.applyVolumeResponse);
         int uploadConcurrency = Math.max(1, request.getUploadConcurrency());
         int maxRetries = Math.max(1, request.getMaxRetries());
         long retryIntervalMillis = Math.max(0L, request.getRetryIntervalMillis());
         long partSizeBytes = Math.max(0L, request.getPartSizeBytes());
-        ExecutorService uploadExecutor = Executors.newFixedThreadPool(uploadConcurrency);
-
         long totalBytes = localPathPair.getValue();
         long totalFilesCount = localPathPair.getKey().size();
-        UploadProgressTracker progressTracker = new UploadProgressTracker(totalBytes, totalFilesCount, request.getProgressListener());
         long startTime = System.currentTimeMillis();
+        logger.info("Starting volume upload: sourcePath:{}, volumeName:{}, volumePath:{}, totalFileCount:{}, totalFileSize:{} bytes ({}), uploadConcurrency:{}, maxRetries:{}, retryInterval:{}, partSize:{}, startTime:{}",
+                localDirOrFilePath, volumeName, volumePath, totalFilesCount, totalBytes, formatBytes(totalBytes),
+                uploadConcurrency, maxRetries, formatDurationMillis(retryIntervalMillis),
+                formatPartSize(partSizeBytes), Instant.ofEpochMilli(startTime));
+
+        UploadContext uploadContext;
+        try {
+            uploadContext = new UploadContext(refreshVolumeAndClient(volumePath));
+            VolumeSession initialSession = uploadContext.currentSession();
+            initValidator(localPathPair, initialSession.applyVolumeResponse);
+        } catch (RuntimeException e) {
+            logUploadFailed(localDirOrFilePath, volumePath, startTime);
+            throw e;
+        }
+        ExecutorService uploadExecutor = Executors.newFixedThreadPool(uploadConcurrency);
+        UploadProgressTracker progressTracker = new UploadProgressTracker(totalBytes, totalFilesCount, request.getProgressListener());
 
         return CompletableFuture.allOf(localPathPair.getKey().stream()
                         .map(localFilePath -> CompletableFuture.runAsync(() -> {
@@ -114,13 +124,19 @@ public CompletableFuture<UploadFilesResult> uploadFilesAsync(UploadFilesRequest
                         }, uploadExecutor)).toArray(CompletableFuture[]::new))
                 .whenComplete((v, t) -> {
                     uploadExecutor.shutdown();
+                    if (t != null) {
+                        logUploadFailed(localDirOrFilePath, volumePath, startTime);
+                    }
                 })
                 .thenApply(v -> {
                     progressTracker.finishUpload();
-                    long totalElapsed = (System.currentTimeMillis() - startTime) / 1000;
+                    long endTime = System.currentTimeMillis();
+                    long totalElapsed = endTime - startTime;
                     VolumeSession session = uploadContext.currentSession();
-                    logger.info("all files in {} has been async uploaded to volume, volumeName:{}, volumePath:{}, totalFileCount:{}, totalFileSize:{}, cost times:{} s",
-                            localDirOrFilePath, session.applyVolumeResponse.getVolumeName(), volumePath, localPathPair.getKey().size(), localPathPair.getValue(), totalElapsed);
+                    logger.info("Volume upload completed: sourcePath:{}, volumeName:{}, volumePath:{}, totalFileCount:{}, totalFileSize:{} bytes ({}), endTime:{}, totalElapsed:{}",
+                            localDirOrFilePath, session.applyVolumeResponse.getVolumeName(), volumePath,
+                            localPathPair.getKey().size(), localPathPair.getValue(), formatBytes(localPathPair.getValue()),
+                            Instant.ofEpochMilli(endTime), formatDurationMillis(totalElapsed));
                     return UploadFilesResult.builder()
                             .volumeName(session.applyVolumeResponse.getVolumeName())
                             .path(volumePath)
@@ -141,6 +157,51 @@ public UploadFilesResult uploadFiles(UploadFilesRequest request) throws Executio
     public void shutdownGracefully() {
     }
 
+    private void logUploadFailed(String localDirOrFilePath, String volumePath, long startTime) {
+        long endTime = System.currentTimeMillis();
+        logger.warn("Volume upload failed: sourcePath:{}, volumeName:{}, volumePath:{}, endTime:{}, totalElapsed:{}",
+                localDirOrFilePath, volumeName, volumePath,
+                Instant.ofEpochMilli(endTime), formatDurationMillis(endTime - startTime));
+    }
+
+    private static String formatPartSize(long partSizeBytes) {
+        if (partSizeBytes <= 0L) {
+            return "auto";
+        }
+        return partSizeBytes + " bytes (" + formatBytes(partSizeBytes) + ")";
+    }
+
+    private static String formatBytes(long bytes) {
+        if (bytes < 1024L) {
+            return bytes + " B";
+        }
+        double value = bytes;
+        String[] units = new String[]{"B", "KiB", "MiB", "GiB", "TiB", "PiB"};
+        int unitIndex = 0;
+        while (value >= 1024.0 && unitIndex < units.length - 1) {
+            value /= 1024.0;
+            unitIndex++;
+        }
+        return String.format(Locale.ROOT, "%.2f %s", value, units[unitIndex]);
+    }
+
+    private static String formatDurationMillis(long durationMillis) {
+        if (durationMillis < 0L) {
+            return "unknown";
+        }
+        long totalSeconds = (durationMillis + 999L) / 1000L;
+        long hours = totalSeconds / 3600L;
+        long minutes = (totalSeconds % 3600L) / 60L;
+        long seconds = totalSeconds % 60L;
+        if (hours > 0L) {
+            return String.format(Locale.ROOT, "%dh %02dm %02ds", hours, minutes, seconds);
+        }
+        if (minutes > 0L) {
+            return String.format(Locale.ROOT, "%dm %02ds", minutes, seconds);
+        }
+        return String.format(Locale.ROOT, "%ds", seconds);
+    }
+
     private void initValidator(Pair<List<String>, Long> localPathPair, ApplyVolumeResponse applyVolumeResponse) {
         Long maxContentLength = applyVolumeResponse.getCondition().getMaxContentLength();
         Long uploadFileContentLength = localPathPair.getValue();
@@ -459,23 +520,6 @@ private long estimatedRemainingTimeMillis(long currentUploadedBytes, long speedB
             return (long) Math.ceil(remainingBytes * 1000.0 / speedBps);
         }
 
-        private String formatDurationMillis(long durationMillis) {
-            if (durationMillis < 0L) {
-                return "unknown";
-            }
-            long totalSeconds = (durationMillis + 999L) / 1000L;
-            long hours = totalSeconds / 3600L;
-            long minutes = (totalSeconds % 3600L) / 60L;
-            long seconds = totalSeconds % 60L;
-            if (hours > 0L) {
-                return String.format("%dh %02dm %02ds", hours, minutes, seconds);
-            }
-            if (minutes > 0L) {
-                return String.format("%dm %02ds", minutes, seconds);
-            }
-            return String.format("%ds", seconds);
-        }
-
         private UploadProgress progressIfNeeded(String currentFile, long currentFileUploadedBytes, long currentFileTotalBytes) {
             long now = System.currentTimeMillis();
             double percent = percent();