Implement comprehensive 3-layer progress tracking system

ThreadSafeProgressCollector utility: Thread-safe progress data collection from background threads Concurrent queues for progress updates and verbose messages Safe PowerShell method calls only from main thread Automatic completion handling and queue processing MultiLayerProgressReporter for 3-layer progress: Layer 1: Collection progress (overall files with ETA) Layer 2: File progress (current file with size info) Layer 3: Transfer progress (bytes with speed calculation) Hierarchical progress reporting with parent-child relationships Real-time speed and ETA calculations Enhanced New-MinIOObject cmdlet integration: UploadSingleFileWithProgress method with progress integration Multi-file upload with comprehensive progress tracking Background thread safety for progress callbacks Verbose message queuing and processing Graceful error handling with progress completion Progress tracking features: Real-time transfer speed calculation Estimated time remaining (ETA) for collections Hierarchical progress bars in PowerShell Thread-safe verbose logging Automatic progress completion on success/failure Architecture improvements: Separation of concerns: progress collection vs reporting Background thread compatibility Memory-efficient concurrent collections Comprehensive error handling and cleanup This provides enterprise-grade progress tracking for large file operations with detailed visibility into collection, file, and transfer-level progress.
2026-07-26 06:48:13 +00:00 · 2025-07-11 17:35:17 -04:00
parent e94cb35c0c
commit 6f693d7236
3 changed files with 454 additions and 24 deletions
@@ -302,7 +302,7 @@ namespace PSMinIO.Cmdlets
        }

        /// <summary>
-        /// Uploads a collection of files
+        /// Uploads a collection of files with multi-layer progress tracking
        /// </summary>
        /// <param name="files">Files to upload</param>
        private void UploadFileCollection(FileInfo[] files)
@@ -324,38 +324,48 @@ namespace PSMinIO.Cmdlets

            WriteVerboseMessage("Uploading {0} files to bucket '{1}'", validFiles.Length, BucketName);

-            // Calculate total size for overall progress
-            var totalSize = validFiles.Sum(f => f.Length);
-            var totalProcessed = 0L;
-
+            // Create multi-layer progress reporter
+            var progressReporter = new MultiLayerProgressReporter(this, validFiles.Length, "Uploading Files");
            var uploadedObjects = new List<MinIOUploadResult>();

-            for (int i = 0; i < validFiles.Length; i++)
+            try
            {
-                var fileInfo = validFiles[i];
-                var objectName = GetObjectName(fileInfo);
-
-                try
+                for (int i = 0; i < validFiles.Length; i++)
                {
-                    WriteVerboseMessage("Uploading file {0}/{1}: {2} -> {3}",
-                        i + 1, validFiles.Length, fileInfo.Name, objectName);
+                    var fileInfo = validFiles[i];
+                    var objectName = GetObjectName(fileInfo);

-                    var uploadResult = UploadSingleFile(fileInfo, objectName);
-                    if (uploadResult != null)
+                    try
                    {
-                        uploadedObjects.Add(uploadResult);
-                        totalProcessed += fileInfo.Length;
+                        // Start new file in progress reporter
+                        progressReporter.StartNewFile(fileInfo.Name, fileInfo.Length);
+                        progressReporter.QueueVerboseMessage("Uploading file {0}/{1}: {2} -> {3}",
+                            i + 1, validFiles.Length, fileInfo.Name, objectName);
+
+                        var uploadResult = UploadSingleFileWithProgress(fileInfo, objectName, progressReporter);
+                        if (uploadResult != null)
+                        {
+                            uploadedObjects.Add(uploadResult);
+                            progressReporter.CompleteCurrentFile();
+                            progressReporter.QueueVerboseMessage("Successfully uploaded: {0}", fileInfo.Name);
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+                        var errorRecord = new ErrorRecord(ex, "FileUploadFailed", ErrorCategory.WriteError, fileInfo);
+                        WriteError(errorRecord);
+                        progressReporter.QueueVerboseMessage("Failed to upload {0}: {1}", fileInfo.Name, ex.Message);
                    }

-                    WriteVerboseMessage("Successfully uploaded: {0}", fileInfo.Name);
-                }
-                catch (Exception ex)
-                {
-                    var errorRecord = new ErrorRecord(ex, "FileUploadFailed", ErrorCategory.WriteError, fileInfo);
-                    WriteError(errorRecord);
-                    WriteVerboseMessage("Failed to upload {0}: {1}", fileInfo.Name, ex.Message);
+                    // Process queued progress updates
+                    progressReporter.ProcessQueuedUpdates();
                }
            }
+            finally
+            {
+                // Complete all progress reporting
+                progressReporter.Complete();
+            }

            // Always return uploaded objects if PassThru is specified
            if (PassThru.IsPresent)
@@ -501,7 +511,73 @@ namespace PSMinIO.Cmdlets
        }

        /// <summary>
-        /// Uploads a single file and returns the result
+        /// Uploads a single file with progress tracking and returns the result
+        /// </summary>
+        /// <param name="fileInfo">File to upload</param>
+        /// <param name="objectName">Object name in bucket</param>
+        /// <param name="progressReporter">Multi-layer progress reporter</param>
+        /// <returns>Upload result</returns>
+        private MinIOUploadResult? UploadSingleFileWithProgress(FileInfo fileInfo, string objectName, MultiLayerProgressReporter progressReporter)
+        {
+            // Create upload result
+            var uploadResult = new MinIOUploadResult(BucketName, objectName, string.Empty, fileInfo.Length)
+            {
+                SourceFilePath = fileInfo.FullName
+            };
+            uploadResult.MarkStarted();
+
+            try
+            {
+                // Determine content type
+                var fileContentType = ContentType ?? GetContentType(fileInfo.Extension);
+                uploadResult.ContentType = fileContentType;
+
+                // Convert metadata
+                Dictionary<string, string>? metadata = null;
+                if (Metadata != null && Metadata.Count > 0)
+                {
+                    metadata = new Dictionary<string, string>();
+                    foreach (var key in Metadata.Keys)
+                    {
+                        if (key != null && Metadata[key] != null)
+                        {
+                            metadata[key.ToString()!] = Metadata[key]!.ToString()!;
+                        }
+                    }
+                }
+
+                // Upload the file with progress tracking
+                string etag;
+                using (var fileStream = fileInfo.OpenRead())
+                {
+                    etag = S3Client.PutObject(
+                        BucketName,
+                        objectName,
+                        fileStream,
+                        fileContentType,
+                        metadata,
+                        bytesTransferred =>
+                        {
+                            // Update progress reporter from background thread
+                            progressReporter.UpdateTransferProgress(bytesTransferred);
+                            uploadResult.BytesTransferred = bytesTransferred;
+                        });
+                }
+
+                uploadResult.ETag = etag;
+                uploadResult.MarkCompleted();
+
+                return uploadResult;
+            }
+            catch (Exception ex)
+            {
+                uploadResult.MarkFailed(ex.Message);
+                throw;
+            }
+        }
+
+        /// <summary>
+        /// Uploads a single file and returns the result (legacy method for backward compatibility)
        /// </summary>
        /// <param name="fileInfo">File to upload</param>
        /// <param name="objectName">Object name in bucket</param>
@@ -0,0 +1,199 @@
+using System;
+using System.Management.Automation;
+
+namespace PSMinIO.Utils
+{
+    /// <summary>
+    /// Multi-layer progress reporter for 3-level progress tracking:
+    /// Collection progress (overall files) > File progress (current file) > Transfer progress (bytes)
+    /// </summary>
+    public class MultiLayerProgressReporter
+    {
+        private readonly PSCmdlet _cmdlet;
+        private readonly ThreadSafeProgressCollector _progressCollector;
+        
+        // Activity IDs for the three layers
+        private const int CollectionActivityId = 1;
+        private const int FileActivityId = 2;
+        private const int TransferActivityId = 3;
+
+        // Collection-level tracking
+        private int _totalFiles;
+        private int _currentFileIndex;
+        private string _collectionOperation = "Processing Files";
+
+        // File-level tracking
+        private string _currentFileName = string.Empty;
+        private long _currentFileSize;
+        private long _currentFileBytesTransferred;
+
+        // Transfer-level tracking
+        private long _totalBytesTransferred;
+        private DateTime _startTime;
+
+        public MultiLayerProgressReporter(PSCmdlet cmdlet, int totalFiles, string operation = "Processing Files")
+        {
+            _cmdlet = cmdlet ?? throw new ArgumentNullException(nameof(cmdlet));
+            _progressCollector = new ThreadSafeProgressCollector(cmdlet);
+            _totalFiles = totalFiles;
+            _collectionOperation = operation;
+            _startTime = DateTime.UtcNow;
+
+            // Initialize collection progress
+            UpdateCollectionProgress();
+        }
+
+        /// <summary>
+        /// Starts processing a new file
+        /// </summary>
+        /// <param name="fileName">Name of the file being processed</param>
+        /// <param name="fileSize">Size of the file in bytes</param>
+        public void StartNewFile(string fileName, long fileSize)
+        {
+            _currentFileIndex++;
+            _currentFileName = fileName;
+            _currentFileSize = fileSize;
+            _currentFileBytesTransferred = 0;
+
+            // Update collection progress
+            UpdateCollectionProgress();
+
+            // Initialize file progress
+            UpdateFileProgress();
+
+            // Initialize transfer progress
+            UpdateTransferProgress();
+
+            // Process any queued updates
+            _progressCollector.ProcessQueuedUpdates();
+        }
+
+        /// <summary>
+        /// Updates transfer progress for the current file
+        /// </summary>
+        /// <param name="bytesTransferred">Bytes transferred for current file</param>
+        public void UpdateTransferProgress(long bytesTransferred)
+        {
+            _currentFileBytesTransferred = bytesTransferred;
+            UpdateTransferProgress();
+        }
+
+        /// <summary>
+        /// Completes the current file and updates overall progress
+        /// </summary>
+        public void CompleteCurrentFile()
+        {
+            _totalBytesTransferred += _currentFileSize;
+
+            // Complete transfer progress
+            _progressCollector.QueueProgressCompletion(TransferActivityId, "Transfer Progress", FileActivityId);
+
+            // Complete file progress
+            _progressCollector.QueueProgressCompletion(FileActivityId, "File Progress", CollectionActivityId);
+
+            // Update collection progress
+            UpdateCollectionProgress();
+
+            // Process updates
+            _progressCollector.ProcessQueuedUpdates();
+        }
+
+        /// <summary>
+        /// Completes all progress reporting
+        /// </summary>
+        public void Complete()
+        {
+            // Complete collection progress
+            _progressCollector.QueueProgressCompletion(CollectionActivityId, _collectionOperation);
+
+            // Process final updates and complete
+            _progressCollector.Complete();
+        }
+
+        /// <summary>
+        /// Processes any queued progress updates
+        /// </summary>
+        public void ProcessQueuedUpdates()
+        {
+            _progressCollector.ProcessQueuedUpdates();
+        }
+
+        /// <summary>
+        /// Updates collection-level progress
+        /// </summary>
+        private void UpdateCollectionProgress()
+        {
+            var percentage = _totalFiles > 0 ? (int)((double)_currentFileIndex / _totalFiles * 100) : 0;
+            var status = $"Processing file {_currentFileIndex} of {_totalFiles}";
+
+            if (_currentFileIndex > 0)
+            {
+                var elapsed = DateTime.UtcNow - _startTime;
+                var avgTimePerFile = elapsed.TotalSeconds / _currentFileIndex;
+                var remainingFiles = _totalFiles - _currentFileIndex;
+                var estimatedTimeRemaining = TimeSpan.FromSeconds(avgTimePerFile * remainingFiles);
+
+                status += $" (ETA: {SizeFormatter.FormatDuration(estimatedTimeRemaining)})";
+            }
+
+            _progressCollector.QueueProgressUpdate(CollectionActivityId, _collectionOperation, status, percentage);
+        }
+
+        /// <summary>
+        /// Updates file-level progress
+        /// </summary>
+        private void UpdateFileProgress()
+        {
+            var percentage = _currentFileSize > 0 ? (int)((double)_currentFileBytesTransferred / _currentFileSize * 100) : 0;
+            var status = $"Processing: {_currentFileName} ({SizeFormatter.FormatBytes(_currentFileBytesTransferred)}/{SizeFormatter.FormatBytes(_currentFileSize)})";
+
+            _progressCollector.QueueProgressUpdate(FileActivityId, "File Progress", status, percentage, CollectionActivityId);
+        }
+
+        /// <summary>
+        /// Updates transfer-level progress
+        /// </summary>
+        private void UpdateTransferProgress()
+        {
+            var percentage = _currentFileSize > 0 ? (int)((double)_currentFileBytesTransferred / _currentFileSize * 100) : 0;
+            var status = $"Transferring: {SizeFormatter.FormatBytes(_currentFileBytesTransferred)}/{SizeFormatter.FormatBytes(_currentFileSize)}";
+
+            if (_currentFileBytesTransferred > 0)
+            {
+                var elapsed = DateTime.UtcNow - _startTime;
+                if (elapsed.TotalSeconds > 0)
+                {
+                    var speed = _currentFileBytesTransferred / elapsed.TotalSeconds;
+                    status += $" at {SizeFormatter.FormatSpeed(speed)}";
+                }
+            }
+
+            _progressCollector.QueueProgressUpdate(TransferActivityId, "Transfer Progress", status, percentage, FileActivityId);
+        }
+
+        /// <summary>
+        /// Queues a verbose message
+        /// </summary>
+        /// <param name="message">Message to log</param>
+        /// <param name="args">Message arguments</param>
+        public void QueueVerboseMessage(string message, params object[] args)
+        {
+            _progressCollector.QueueVerboseMessage(message, args);
+        }
+
+        /// <summary>
+        /// Gets the current collection progress percentage
+        /// </summary>
+        public int CollectionProgressPercentage => _totalFiles > 0 ? (int)((double)_currentFileIndex / _totalFiles * 100) : 0;
+
+        /// <summary>
+        /// Gets the current file progress percentage
+        /// </summary>
+        public int FileProgressPercentage => _currentFileSize > 0 ? (int)((double)_currentFileBytesTransferred / _currentFileSize * 100) : 0;
+
+        /// <summary>
+        /// Gets the total bytes transferred across all files
+        /// </summary>
+        public long TotalBytesTransferred => _totalBytesTransferred + _currentFileBytesTransferred;
+    }
+}
@@ -0,0 +1,155 @@
+using System;
+using System.Collections.Concurrent;
+using System.Management.Automation;
+using System.Threading;
+
+namespace PSMinIO.Utils
+{
+    /// <summary>
+    /// Thread-safe progress data collector that accumulates progress updates from background threads
+    /// and allows the main thread to safely report them to PowerShell
+    /// </summary>
+    public class ThreadSafeProgressCollector
+    {
+        private readonly PSCmdlet _cmdlet;
+        private readonly ConcurrentQueue<ProgressUpdate> _progressQueue = new();
+        private readonly ConcurrentQueue<VerboseMessage> _verboseQueue = new();
+        private readonly object _lockObject = new();
+        private volatile bool _isCompleted = false;
+
+        public ThreadSafeProgressCollector(PSCmdlet cmdlet)
+        {
+            _cmdlet = cmdlet ?? throw new ArgumentNullException(nameof(cmdlet));
+        }
+
+        /// <summary>
+        /// Queues a progress update from a background thread
+        /// </summary>
+        public void QueueProgressUpdate(int activityId, string activity, string statusDescription, int percentComplete, int parentActivityId = -1)
+        {
+            if (_isCompleted) return;
+
+            _progressQueue.Enqueue(new ProgressUpdate
+            {
+                ActivityId = activityId,
+                Activity = activity,
+                StatusDescription = statusDescription,
+                PercentComplete = percentComplete,
+                ParentActivityId = parentActivityId,
+                Timestamp = DateTime.UtcNow
+            });
+        }
+
+        /// <summary>
+        /// Queues a progress completion from a background thread
+        /// </summary>
+        public void QueueProgressCompletion(int activityId, string activity, int parentActivityId = -1)
+        {
+            if (_isCompleted) return;
+
+            _progressQueue.Enqueue(new ProgressUpdate
+            {
+                ActivityId = activityId,
+                Activity = activity,
+                StatusDescription = "Completed",
+                PercentComplete = 100,
+                ParentActivityId = parentActivityId,
+                IsCompleted = true,
+                Timestamp = DateTime.UtcNow
+            });
+        }
+
+        /// <summary>
+        /// Queues a verbose message from a background thread
+        /// </summary>
+        public void QueueVerboseMessage(string message, params object[] args)
+        {
+            if (_isCompleted) return;
+
+            _verboseQueue.Enqueue(new VerboseMessage
+            {
+                Message = args.Length > 0 ? string.Format(message, args) : message,
+                Timestamp = DateTime.UtcNow
+            });
+        }
+
+        /// <summary>
+        /// Processes all queued updates from the main thread (safe to call PowerShell methods)
+        /// </summary>
+        public void ProcessQueuedUpdates()
+        {
+            // Process verbose messages first
+            while (_verboseQueue.TryDequeue(out var verboseMessage))
+            {
+                MinIOLogger.WriteVerbose(_cmdlet, verboseMessage.Message);
+            }
+
+            // Process progress updates
+            while (_progressQueue.TryDequeue(out var progressUpdate))
+            {
+                var progressRecord = new ProgressRecord(
+                    progressUpdate.ActivityId,
+                    progressUpdate.Activity,
+                    progressUpdate.StatusDescription)
+                {
+                    PercentComplete = progressUpdate.PercentComplete
+                };
+
+                if (progressUpdate.ParentActivityId >= 0)
+                {
+                    progressRecord.ParentActivityId = progressUpdate.ParentActivityId;
+                }
+
+                if (progressUpdate.IsCompleted)
+                {
+                    progressRecord.RecordType = ProgressRecordType.Completed;
+                }
+
+                _cmdlet.WriteProgress(progressRecord);
+            }
+        }
+
+        /// <summary>
+        /// Marks the collector as completed (no more updates will be accepted)
+        /// </summary>
+        public void Complete()
+        {
+            _isCompleted = true;
+            // Process any remaining updates
+            ProcessQueuedUpdates();
+        }
+
+        /// <summary>
+        /// Gets the number of pending progress updates
+        /// </summary>
+        public int PendingProgressUpdates => _progressQueue.Count;
+
+        /// <summary>
+        /// Gets the number of pending verbose messages
+        /// </summary>
+        public int PendingVerboseMessages => _verboseQueue.Count;
+
+        /// <summary>
+        /// Progress update data structure
+        /// </summary>
+        private class ProgressUpdate
+        {
+            public int ActivityId { get; set; }
+            public string Activity { get; set; } = string.Empty;
+            public string StatusDescription { get; set; } = string.Empty;
+            public int PercentComplete { get; set; }
+            public int ParentActivityId { get; set; } = -1;
+            public bool IsCompleted { get; set; }
+            public DateTime Timestamp { get; set; }
+        }
+
+        /// <summary>
+        /// Verbose message data structure
+        /// </summary>
+        private class VerboseMessage
+        {
+            public string Message { get; set; } = string.Empty;
+            public DateTime Timestamp { get; set; }
+        }
+    }
+}