Switch runtime metrics to operation-weighted progress

HISTORY.md

@@ -6,8 +6,8 @@ Utility to optimize media files for Direct Play in Plex, Emby, Jellyfin, etc.
 
 - Version 3.22:
   - Added always-on runtime metrics published via `System.Diagnostics.Metrics` under the `PlexCleaner.Process` meter, readable with `dotnet-counters` with no extra infrastructure.
-    - Overall progress is weighted by input bytes rather than file count, so a run mixing tiny and huge files reports the actual work completed, and it advances smoothly during a long operation because each in-flight file contributes partial credit from the running tool (the full-file scan, the re-encode, and the deinterlace).
-    - Instruments include the run file and byte totals, in-flight and active-thread counts, completed bytes, the weighted `progress.ratio` and `eta.seconds`, cumulative per-outcome counters (completed, modified, errors, verify-failed, and a per-`State`-flag tally), and the `file.duration` and per-tool `tool.duration` histograms. Metrics are aggregate only, with bounded `state` and `tool` tags and no filename tags.
+    - Overall progress is operation-weighted: each heavy full-file operation (the closed-caption and interlace scans, bitrate analysis, re-encode, deinterlace, and verify) counts the file's size as work to do when it starts and as work done when it finishes, so a run mixing tiny and huge files reports the actual work completed, and the total grows as the non-deterministic per-file path is discovered.
+    - Instruments include the run file total and input byte total, in-flight and active-thread counts, the `work.total`/`work.completed` byte gauges and the `progress.ratio` and `eta.seconds` derived from them, cumulative per-outcome counters (completed, modified, errors, verify-failed, and a per-`State`-flag tally), and the `file.duration` and per-tool `tool.duration` histograms. Metrics are aggregate only, with bounded `state` and `tool` tags and no filename tags.
     - The run-scoped gauges reset at the start of every processing pass, so monitor mode and back-to-back commands each report their own run, while the counters stay cumulative for rate display.
     - Instruments are inert until a listener observes them, so the feature is always on with no configuration flag and negligible idle overhead.
     - The meter is OpenTelemetry and `dotnet-monitor` compatible, and the Docker image ships a `counters` wrapper, so reading the metrics is `docker exec <container> counters`.

PlexCleaner/FfMpegTool.cs

@@ -164,11 +164,14 @@ public VerifyResult VerifyMedia(string fileName)
             // Execute command: ffmpeg can exit 0 yet report stream errors on stderr
             // Classify stderr line by line as it streams to keep memory bounded, e.g. non-monotonic-DTS file emits a warning per packet
             VerifyClassifier.Accumulator classifier = new();
-            if (!ExecuteStreamStdErr(command, classifier.Add, out int exitCode))
+            Metrics.OpStarted();
+            bool executed = ExecuteStreamStdErr(command, classifier.Add, out int exitCode);
+            if (!executed)
             {
                 // Process could not run
                 return VerifyResult.DecodeError;
             }
+            Metrics.OpCompleted();
 
             // A non-zero exit is always a failure, fail closed even if stderr shows only the timestamp warning
             VerifyResult verifyResult = classifier.Result;
@@ -310,24 +313,6 @@ when durationUs > 0
             };
         }
 
-        private bool ExecuteEncodeWithProgress(Command command, string inputName)
-        {
-            Metrics.FileSink? sink = Metrics.CurrentFileSink;
-            return ExecuteStreamStdOut(
-                    command,
-                    line =>
-                    {
-                        double? fraction = ParseProgressFraction(line, sink?.DurationUs ?? 0);
-                        if (fraction.HasValue)
-                        {
-                            Metrics.ReportFileFraction(sink, fraction.Value);
-                        }
-                    },
-                    out int exitCode,
-                    out string standardError
-                ) && (exitCode == 0 || LogFailedResult(exitCode, standardError, inputName));
-        }
-
         public bool ConvertToMkv(
             string inputName,
             SelectMediaProps? selectMediaProps,
@@ -351,10 +336,7 @@ string outputName
             // Build command line
             Command command = GetBuilder()
                 .GlobalOptions(options =>
-                    options
-                        .Default()
-                        .Progress()
-                        .Add(Program.Config.ConvertOptions.FfMpegOptions.Global)
+                    options.Default().Add(Program.Config.ConvertOptions.FfMpegOptions.Global)
                 )
                 .InputOptions(options => options.Default().TestSnippets().InputFile(inputName))
                 .OutputOptions(options =>
@@ -368,7 +350,14 @@ string outputName
                 .Build();
 
             // Execute command
-            return ExecuteEncodeWithProgress(command, inputName);
+            Metrics.OpStarted();
+            bool executed = Execute(command, true, true, out BufferedCommandResult result);
+            if (!executed)
+            {
+                return false;
+            }
+            Metrics.OpCompleted();
+            return result.ExitCode == 0 || LogFailedResult(result, inputName);
         }
 
         public bool ConvertToMkv(string inputName, string outputName)
@@ -379,10 +368,7 @@ public bool ConvertToMkv(string inputName, string outputName)
             // Build command line
             Command command = GetBuilder()
                 .GlobalOptions(options =>
-                    options
-                        .Default()
-                        .Progress()
-                        .Add(Program.Config.ConvertOptions.FfMpegOptions.Global)
+                    options.Default().Add(Program.Config.ConvertOptions.FfMpegOptions.Global)
                 )
                 .InputOptions(options => options.Default().TestSnippets().InputFile(inputName))
                 .OutputOptions(options =>
@@ -398,7 +384,14 @@ public bool ConvertToMkv(string inputName, string outputName)
                 .Build();
 
             // Execute command
-            return ExecuteEncodeWithProgress(command, inputName);
+            Metrics.OpStarted();
+            bool executed = Execute(command, true, true, out BufferedCommandResult result);
+            if (!executed)
+            {
+                return false;
+            }
+            Metrics.OpCompleted();
+            return result.ExitCode == 0 || LogFailedResult(result, inputName);
         }
 
         public bool SetTimestamps(string inputName, string outputName)
@@ -525,10 +518,13 @@ public bool GetIdetText(string fileName, out string text)
                 .Build();
 
             // Execute command
-            if (!Execute(command, true, true, out BufferedCommandResult result))
+            Metrics.OpStarted();
+            bool executed = Execute(command, true, true, out BufferedCommandResult result);
+            if (!executed)
             {
                 return false;
             }
+            Metrics.OpCompleted();
             text = result.StandardError.Trim();
             return result.ExitCode == 0 || LogFailedResult(result, fileName);
         }

PlexCleaner/FfProbeTool.cs

@@ -226,10 +226,13 @@ public bool GetClosedCaptions(string fileName, out bool hasClosedCaptions)
                 .Build();
 
             // Execute command
-            if (!Execute(command, false, true, out BufferedCommandResult result))
+            Metrics.OpStarted();
+            bool executed = Execute(command, false, true, out BufferedCommandResult result);
+            if (!executed)
             {
                 return false;
             }
+            Metrics.OpCompleted();
             if (result.ExitCode != 0)
             {
                 return LogFailedResult(result, fileName);
@@ -313,12 +316,15 @@ bool quickScan
                 .Build();
 
             // Get packet list
-            if (!GetPackets(command, packetFunc, out string error))
+            Metrics.OpStarted();
+            bool got = GetPackets(command, packetFunc, out string error);
+            if (!got)
             {
                 Log.Error("Failed to get analysis packets : {FileName}", fileName);
                 LogErrorOutput(error);
                 return false;
             }
+            Metrics.OpCompleted();
             return true;
         }
 

PlexCleaner/HandBrakeTool.cs

@@ -121,7 +121,7 @@ bool deInterlace
 
             // Build command line
             Command command = GetBuilder()
-                .GlobalOptions(options => options.Default().Json())
+                .GlobalOptions(options => options.Default())
                 .InputOptions(options => options.InputFile(inputName).TestSnippets())
                 .OutputOptions(options =>
                     options
@@ -140,20 +140,14 @@ bool deInterlace
                 .Build();
 
             // Execute command
-            Metrics.FileSink? sink = Metrics.CurrentFileSink;
-            return ExecuteStreamStdOut(
-                    command,
-                    line =>
-                    {
-                        double? fraction = ParseProgressFraction(line);
-                        if (fraction.HasValue)
-                        {
-                            Metrics.ReportFileFraction(sink, fraction.Value);
-                        }
-                    },
-                    out int exitCode,
-                    out string standardError
-                ) && (exitCode == 0 || LogFailedResult(exitCode, standardError, inputName));
+            Metrics.OpStarted();
+            bool executed = Execute(command, true, true, out BufferedCommandResult result);
+            if (!executed)
+            {
+                return false;
+            }
+            Metrics.OpCompleted();
+            return result.ExitCode == 0 || LogFailedResult(result, inputName);
         }
 
         [GeneratedRegex(

PlexCleaner/Metrics.cs

@@ -1,32 +1,10 @@
-using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Diagnostics.Metrics;
 
 namespace PlexCleaner;
 
 internal static class Metrics
 {
-    // Per-file partial credit for the in-flight fold.
-    internal sealed class FileSink
-    {
-        internal long Bytes { get; init; }
-
-        // Written on the worker thread and read on the tool pipe thread, so accessed via Volatile.
-        private long _durationUs;
-
-        internal long DurationUs => Volatile.Read(ref _durationUs);
-
-        internal void SetDurationUs(long durationUs) => Volatile.Write(ref _durationUs, durationUs);
-
-        // Fraction as a 0..10000 permyriad so it reads and writes atomically as an int.
-        private int _permyriad;
-
-        internal double Fraction => Volatile.Read(ref _permyriad) / 10000.0;
-
-        internal void SetFraction(double fraction) =>
-            Volatile.Write(ref _permyriad, (int)Math.Clamp(fraction * 10000.0, 0.0, 10000.0));
-    }
-
     private static readonly Meter s_meter = new("PlexCleaner.Process");
 
     // Cumulative for the process lifetime, not reset between runs.
@@ -66,13 +44,15 @@ internal void SetFraction(double fraction) =>
     // All access is via Interlocked so the parallel loop needs no lock.
     private static long s_runFilesTotal;
     private static long s_runBytesTotal;
-    private static long s_runBytesCompleted;
+
+    // Operation-weighted progress: each heavy full-file operation adds the file size to the work total when it starts and the completed total when it ends.
+    private static long s_runWorkTotal;
+    private static long s_runWorkCompleted;
     private static long s_runInflight;
     private static long s_runStartTimestamp;
 
-    // The current file is exposed via ThreadLocal so a worker-thread call site can capture it for a tool closure.
-    private static readonly ConcurrentDictionary<FileSink, byte> s_runInflightSinks = new();
-    private static readonly ThreadLocal<FileSink?> s_currentFileSink = new();
+    // The current file's size, set on the worker thread so OpStarted and OpCompleted can weight by it.
+    private static readonly ThreadLocal<long> s_currentFileSize = new();
 
     // Each State flag (minus None) with its tag, pre-built once so RecordStates allocates nothing per file.
     private static readonly (
@@ -119,15 +99,21 @@ static Metrics()
             description: "Sum of input sizes in the current run"
         );
         _ = s_meter.CreateObservableGauge(
-            "plexcleaner.bytes.completed",
-            () => Interlocked.Read(ref s_runBytesCompleted),
+            "plexcleaner.work.total",
+            () => Interlocked.Read(ref s_runWorkTotal),
+            unit: "By",
+            description: "Operation work discovered, file size added per heavy operation, grows as the path unfolds"
+        );
+        _ = s_meter.CreateObservableGauge(
+            "plexcleaner.work.completed",
+            () => Interlocked.Read(ref s_runWorkCompleted),
             unit: "By",
-            description: "Bytes of finished files (no partial credit for in-flight files)"
+            description: "Operation work finished, file size added per completed heavy operation"
         );
         _ = s_meter.CreateObservableGauge(
             "plexcleaner.progress.ratio",
             ComputeProgress,
-            description: "Byte-weighted overall progress [0..1]"
+            description: "Operation-weighted overall progress [0..1]"
         );
         _ = s_meter.CreateObservableGauge(
             "plexcleaner.eta.seconds",
@@ -142,50 +128,34 @@ internal static void BeginRun(long totalFiles, long totalBytes)
     {
         _ = Interlocked.Exchange(ref s_runFilesTotal, totalFiles);
         _ = Interlocked.Exchange(ref s_runBytesTotal, totalBytes);
-        _ = Interlocked.Exchange(ref s_runBytesCompleted, 0);
+        _ = Interlocked.Exchange(ref s_runWorkTotal, 0);
+        _ = Interlocked.Exchange(ref s_runWorkCompleted, 0);
         _ = Interlocked.Exchange(ref s_runInflight, 0);
         _ = Interlocked.Exchange(ref s_runStartTimestamp, Stopwatch.GetTimestamp());
-        s_runInflightSinks.Clear();
     }
 
     internal static void FileStarted(long sizeBytes)
     {
-        FileSink sink = new() { Bytes = sizeBytes };
-        _ = s_runInflightSinks.TryAdd(sink, 0);
-        s_currentFileSink.Value = sink;
+        s_currentFileSize.Value = sizeBytes;
         _ = Interlocked.Increment(ref s_runInflight);
     }
 
-    // Remove the current file's partial credit before FileCompleted folds its whole size.
     internal static void FileInflightDone()
     {
-        FileSink? sink = s_currentFileSink.Value;
-        s_currentFileSink.Value = null;
-        if (sink != null)
-        {
-            _ = s_runInflightSinks.TryRemove(sink, out _);
-        }
+        s_currentFileSize.Value = 0;
         _ = Interlocked.Decrement(ref s_runInflight);
     }
 
-    internal static FileSink? CurrentFileSink => s_currentFileSink.Value;
+    // A heavy full-file operation started, count the current file's size as work to do.
+    internal static void OpStarted() =>
+        Interlocked.Add(ref s_runWorkTotal, s_currentFileSize.Value);
 
-    internal static void SetCurrentFileDurationUs(long durationUs)
-    {
-        if (s_currentFileSink.Value is { } sink)
-        {
-            sink.SetDurationUs(durationUs);
-        }
-    }
-
-    // Safe to call from a tool's pipe thread.
-    internal static void ReportFileFraction(FileSink? sink, double fraction) =>
-        sink?.SetFraction(fraction);
+    // The heavy operation finished, count the same size as work done.
+    internal static void OpCompleted() =>
+        Interlocked.Add(ref s_runWorkCompleted, s_currentFileSize.Value);
 
-    // A finished file credits its whole size and its wall-clock time.
-    internal static void FileCompleted(long sizeBytes, TimeSpan wall)
+    internal static void FileCompleted(TimeSpan wall)
     {
-        _ = Interlocked.Add(ref s_runBytesCompleted, sizeBytes);
         s_filesCompleted.Add(1);
         s_fileDuration.Record(wall.TotalMilliseconds);
     }
@@ -212,23 +182,19 @@ internal static void RecordToolDuration(MediaTool.ToolType tool, double millisec
 
     internal static void Dispose()
     {
-        s_currentFileSink.Dispose();
+        s_currentFileSize.Dispose();
         s_meter.Dispose();
     }
 
-    // Completed bytes plus each in-flight file's partial credit, byte-weighted, guarding a zero total.
+    // Completed operation work over discovered operation work, guarding a zero total.
     internal static double ComputeProgress()
     {
-        long total = Interlocked.Read(ref s_runBytesTotal);
+        long total = Interlocked.Read(ref s_runWorkTotal);
         if (total <= 0)
         {
             return 0.0;
         }
-        double completed = Interlocked.Read(ref s_runBytesCompleted);
-        foreach (KeyValuePair<FileSink, byte> entry in s_runInflightSinks)
-        {
-            completed += entry.Key.Bytes * entry.Key.Fraction;
-        }
+        double completed = Interlocked.Read(ref s_runWorkCompleted);
         return Math.Clamp(completed / total, 0.0, 1.0);
     }
 

PlexCleaner/PlexCleaner.csproj

@@ -35,6 +35,10 @@
   <PropertyGroup Condition="'$(PublishAot)' != 'true'">
     <DefineConstants>$(DefineConstants);PLUGINS</DefineConstants>
   </PropertyGroup>
+  <!-- NativeAOT links the disabled EventPipe and omits the diagnostics IPC server by default, opt in so dotnet-counters and dotnet-monitor can read the runtime metrics from an AOT build -->
+  <PropertyGroup Condition="'$(PublishAot)' == 'true'">
+    <EventSourceSupport>true</EventSourceSupport>
+  </PropertyGroup>
   <ItemGroup>
     <None Include="..\README.md">
       <Pack>True</Pack>

PlexCleaner/Process.cs

@@ -152,11 +152,6 @@ out string? failedOperation
                 break;
             }
 
-            // Seed the ffmpeg progress denominator now that the duration is known
-            Metrics.SetCurrentFileDurationUs(
-                (long)processFile.FfProbeProps.Duration.TotalMicroseconds
-            );
-
             // ReMux non-MKV containers using MKV file extensions
             // Conditional on ReMux option, fails if not Matroska and ReMux is not enabled
             operation = nameof(processFile.RemuxNonMkvContainer);