Add scale down to HistoryPubnetParallelCatchupV2

src/CSLibrary/RemoteCommandRunner.cs

@@ -66,14 +66,15 @@ await kube.MuxedStreamNamespacedPodExecAsync(name: podName, @namespace: ns,
         }
 
         // Execute a command and capture stdout to a file (for copying files from pod)
-        public static void RunRemoteCommandAndCaptureOutput(Kubernetes kube, string ns, string podName,
+        // Returns the command's exit code
+        public static int RunRemoteCommandAndCaptureOutput(Kubernetes kube, string ns, string podName,
             string containerName, string[] command, string outputFilePath)
         {
-            Task task = RunRemoteCommandAndCaptureOutputAsync(kube, ns, podName, containerName, command, outputFilePath);
-            task.Wait();
+            Task<int> task = RunRemoteCommandAndCaptureOutputAsync(kube, ns, podName, containerName, command, outputFilePath);
+            return task.Result;
         }
 
-        public static async Task RunRemoteCommandAndCaptureOutputAsync(Kubernetes kube, string ns, string podName,
+        public static async Task<int> RunRemoteCommandAndCaptureOutputAsync(Kubernetes kube, string ns, string podName,
             string containerName, string[] command, string outputFilePath)
         {
             // The `using` lifetime guard ensure these objects lifetimes last the entire task,
@@ -89,30 +90,26 @@ await kube.MuxedStreamNamespacedPodExecAsync(
                    stderr: true,
                    tty: false).ConfigureAwait(false))
             using (System.IO.Stream stdout = mstr.GetStream(ChannelIndex.StdOut, null))
-            using (System.IO.Stream stderr = mstr.GetStream(ChannelIndex.Error, null))
-            using (System.IO.StreamReader errorReader = new System.IO.StreamReader(stderr))
+            using (System.IO.Stream statusChannel = mstr.GetStream(ChannelIndex.Error, null))
+            using (System.IO.StreamReader statusReader = new System.IO.StreamReader(statusChannel))
             using (System.IO.FileStream fileStream = new System.IO.FileStream(outputFilePath, FileMode.Create, FileAccess.Write, FileShare.None,
         bufferSize: 128 * 1024, useAsync: true))
             {
                 // Start the MuxStream, this establishes the connection and routes bytes back into separate channels.
-                // We only care about stdout(1) and stderr(2). 
+                // We only care about stdout(1) and the status channel(3).
                 mstr.Start();
 
-                // Copy stdout → file asynchronously, and drain stderr concurrently.
+                // Copy stdout → file asynchronously, and drain the status channel concurrently.
                 var copyTask = stdout.CopyToAsync(fileStream);
-                var errorTask = errorReader.ReadToEndAsync();
+                var statusTask = statusReader.ReadToEndAsync();
 
-                await Task.WhenAll(copyTask, errorTask).ConfigureAwait(false);
+                await Task.WhenAll(copyTask, statusTask).ConfigureAwait(false);
 
                 // Flush the file stream to ensure all data is written
                 await fileStream.FlushAsync().ConfigureAwait(false);
 
-                // Log any errors to console
-                string errors = errorTask.Result;
-                if (!string.IsNullOrEmpty(errors))
-                {
-                    Console.WriteLine($"Command stderr from pod {podName}: {errors}");
-                }
+                string status = statusTask.Result;
+                return Kubernetes.GetExitCodeOrThrow(SafeJsonConvert.DeserializeObject<V1Status>(status));
             }
         }
     }

src/FSLibrary/MissionHistoryPubnetParallelCatchupV2.fs

@@ -51,6 +51,13 @@ let failedJobLogStreamLineCount = 1000
 
 let mutable nonce : String = ""
 let mutable helmReleaseName : String = ""
+// Pods not yet retired; module scope because cleanup runs from a signal handler.
+let mutable livePods : Set<string> = Set.empty
+// Log collection is serial and runs inside the poll loop, so bound what one pass can block on.
+let maxRetiredPerPass = 64
+
+// A job the monitor requeues needs a worker still willing to claim it.
+let minUnmarkedWorkers = 3
 
 let jobMonitorHostName (context: MissionContext) =
     match context.jobMonitorExternalHost with
@@ -227,6 +234,8 @@ let installProject (context: MissionContext) =
                        "install"
                        helmReleaseName
                        helmChartPath
+                       "--namespace"
+                       context.namespaceProperty
                        "--values"
                        valuesFilePath
                        "--set"
@@ -236,21 +245,16 @@ let installProject (context: MissionContext) =
     match RunShellCommand [| "helm"
                              "get"
                              "values"
-                             helmReleaseName |] with
+                             helmReleaseName
+                             "--namespace"
+                             context.namespaceProperty |] with
     | Some valuesOutput -> LogInfo "%s" valuesOutput
     | _ -> ()
 
-// Collect log files from all parallel catchup worker pods
-// This function:
-// 1. Automatically determines worker pod names from context.pubnetParallelCatchupNumWorkers
-// 2. For each pod, finds all files matching "stellar-core-*.log" in /data
-// 3. Creates a tar.gz archive and copies it to context.destination directory
-let collectLogsFromPods (context: MissionContext) =
-    // Generate pod names based on number of workers
-    // Pod names follow the pattern: <helmReleaseName>-stellar-core-0, <helmReleaseName>-stellar-core-1, etc.
-    let podNames =
-        [ 0 .. context.pubnetParallelCatchupNumWorkers - 1 ]
-        |> List.map (fun i -> sprintf "%s-stellar-core-%d" helmReleaseName i)
+// Collect log files from the given parallel catchup worker pods.
+// Returns the pods whose collection raised; an empty archive is success.
+let collectLogsFromPods (context: MissionContext) (podNames: string list) : string list =
+    let mutable failed = []
 
     LogInfo "Collecting logs from %d worker pods to directory: %s" (List.length podNames) context.destination.Path
 
@@ -275,6 +279,7 @@ let collectLogsFromPods (context: MissionContext) =
                 command = command,
                 outputFilePath = outputFile
             )
+            |> ignore
 
             let fileInfo = FileInfo(outputFile)
 
@@ -285,6 +290,48 @@ let collectLogsFromPods (context: MissionContext) =
 
         with ex ->
             LogWarn "Could not collect logs from pod %s (this is expected if pod doesn't exist): %s" podName ex.Message
+            failed <- podName :: failed
+
+    failed
+
+// Running pods only, since a Pending pod reads as idle capacity it cannot supply.
+let readyPods (context: MissionContext) : Set<string> =
+    let selector = "app=" + helmReleaseName + "-stellar-core"
+
+    let pods =
+        context.kube.ListNamespacedPod(context.namespaceProperty, labelSelector = selector)
+
+    pods.Items
+    |> Seq.filter (fun pod -> pod.Status.Phase = "Running" && isNull (box pod.Metadata.DeletionTimestamp))
+    |> Seq.map (fun pod -> pod.Metadata.Name)
+    |> Set.ofSeq
+
+// Runs redis-cli in a ready worker and returns its output lines, or none if there is no host.
+let redisIn (context: MissionContext) (ready: Set<string>) (args: string) : string list =
+    match Seq.tryHead ready with
+    | None -> []
+    | Some host ->
+        let outFile = Path.Combine(Path.GetTempPath(), helmReleaseName + "-redis.txt")
+        let sh = sprintf "redis-cli -h \"$REDIS_HOST\" -p \"$REDIS_PORT\" %s" args
+        let cmd = [| "sh"; "-c"; sh |]
+
+        // A failed exec would otherwise return no lines, which reads as "no worker
+        // is busy" and makes every worker look retirable.
+        let rc =
+            RemoteCommandRunner.RunRemoteCommandAndCaptureOutput(
+                context.kube,
+                context.namespaceProperty,
+                host,
+                "stellar-core",
+                cmd,
+                outFile
+            )
+
+        if rc <> 0 then failwithf "redis-cli in %s exited %d" host rc
+
+        File.ReadAllLines outFile
+        |> Array.toList
+        |> List.filter (fun l -> l.Trim() <> "")
 
 // Cleanup on exit. `signalTriggered` indicates we're running under a hard
 // deadline (Jenkins' SoftKillWaitSeconds, ~5s by default, before SIGKILL).
@@ -307,7 +354,9 @@ let cleanup (signalTriggered: bool) (context: MissionContext) =
 
             RunShellCommand [| "helm"
                                "uninstall"
-                               helmReleaseName |]
+                               helmReleaseName
+                               "--namespace"
+                               context.namespaceProperty |]
             |> ignore
         else
             // Normal / legitimate-failure path: pods are still alive through
@@ -317,14 +366,16 @@ let cleanup (signalTriggered: bool) (context: MissionContext) =
             try
                 LogInfo "Attempting to collect worker logs before cleanup..."
                 let stopwatch = Stopwatch.StartNew()
-                collectLogsFromPods context
+                collectLogsFromPods context (List.ofSeq livePods) |> ignore
                 stopwatch.Stop()
                 LogInfo "Log collection completed in %.2f seconds" stopwatch.Elapsed.TotalSeconds
             with ex -> LogWarn "Failed to collect some or all worker logs: %s" ex.Message
 
             RunShellCommand [| "helm"
                                "uninstall"
-                               helmReleaseName |]
+                               helmReleaseName
+                               "--namespace"
+                               context.namespaceProperty |]
             |> ignore
 
 let mutable cleanupContext : MissionContext option = None
@@ -399,6 +450,12 @@ let historyPubnetParallelCatchupV2 (context: MissionContext) =
     installProject context
 
     let mutable allJobsFinished = false
+
+    livePods <-
+        Set.ofList [ for i in 0 .. context.pubnetParallelCatchupNumWorkers - 1 ->
+                         sprintf "%s-stellar-core-%d-0" helmReleaseName i ]
+    // Marks from earlier passes only, so nothing is removed in the pass that marked it.
+    let mutable marked : Set<string> = Set.empty
     let mutable timeoutLeft = jobMonitorStatusCheckTimeOutSecs
     let mutable timeBeforeNextMetricsCheck = jobMonitorMetricsCheckIntervalSecs
     let mutable stalledForSecs = 0
@@ -430,6 +487,56 @@ let historyPubnetParallelCatchupV2 (context: MissionContext) =
 
                     failwith "Catch up failed, check logs for more info"
 
+                // `queue_remain_count`, not `num_remain`, which is 1 as a pre-first-poll sentinel.
+                let outstanding = status.Value<int>("queue_remain_count") + JobsInProgress.Count
+
+                try
+                    // Each read costs an apiserver call or a pod exec, so skip both when nothing is marked and the queue still outruns the fleet.
+                    let doReads = not marked.IsEmpty || outstanding < livePods.Count
+                    let mutable ready = if doReads then readyPods context else Set.empty
+                    // Read job_owners directly so it is current, not as old as the status snapshot.
+                    let busy = redisIn context ready "HVALS \"$JOB_OWNERS\"" |> Set.ofList
+
+                    let idle p = ready.Contains p && not (busy.Contains p)
+                    let removable = marked |> Seq.filter idle |> Seq.truncate maxRetiredPerPass |> Set.ofSeq
+
+                    if not removable.IsEmpty then
+                        // /data is emptyDir, so a pod removed before its logs are read loses them.
+                        match collectLogsFromPods context (List.ofSeq removable) with
+                        | [] ->
+                            for pod in removable do
+                                let sts = pod.Substring(0, pod.Length - 2)
+
+                                context.kube.DeleteNamespacedStatefulSet(sts, context.namespaceProperty)
+                                |> ignore
+
+                            marked <- Set.difference marked removable
+                            livePods <- Set.difference livePods removable
+                            ready <- Set.difference ready removable
+                            LogInfo "Retired %d workers (%d outstanding)" removable.Count outstanding
+                        | failed -> LogWarn "Not retiring: log collection failed for %d workers" failed.Length
+
+                    // Counted against unmarked workers, not `ready`: marked pods linger until
+                    // they are deleted, and counting them erodes the reserve to nothing.
+                    let unmarked = ready |> Seq.filter (fun p -> not (marked.Contains p)) |> List.ofSeq
+
+                    let toMark =
+                        unmarked
+                        |> List.filter (fun p -> not (busy.Contains p))
+                        |> List.truncate (max 0 (unmarked.Length - max outstanding minUnmarkedWorkers))
+
+                    // Chunked because RunRemoteCommand rejects a command of 4096 bytes or more.
+                    for chunk in List.chunkBySize 30 toMark do
+                        let names = chunk |> List.map (sprintf "'%s'") |> String.concat " "
+
+                        redisIn context ready (sprintf "SADD \"%s-retiring\" %s" helmReleaseName names)
+                        |> ignore
+
+                        marked <- Set.union marked (Set.ofList chunk)
+
+                    if not toMark.IsEmpty then
+                        LogInfo "Marked %d retiring (%d ready, %d outstanding)" toMark.Length ready.Count outstanding
+                with ex -> LogWarn "Worker scale-down skipped this pass: %s" ex.Message
                 // Detect if the mission is stuck from two signals: 1. job queue
                 // has in progress items but no live workers 2. the job monitor
                 // itself gets stuck unable to updating its internal metrics and

src/FSLibrary/StellarOrphanSweep.fs

@@ -72,20 +72,20 @@ let private sweepWithCutoff (cutoff: DateTime) (kube: Kubernetes) (ns: string) (
 
     let stsItems = kube.ListNamespacedStatefulSet(namespaceParameter = ns).Items
 
-    for sts in stsItems do
-        if isOlderThan cutoff sts.Metadata then
-            let name = sts.Metadata.Name
-
-            if name.StartsWith("parallel-catchup-") && name.EndsWith("-stellar-core") then
-                let release = name.Substring(0, name.Length - "-stellar-core".Length)
-                LogInfo "Orphan sweep: helm uninstall %s" release
-
-                RunShellCommand [| "helm"
-                                   "uninstall"
-                                   release
-                                   "-n"
-                                   ns |]
-                |> ignore
+    for release in stsItems
+                   |> Seq.filter (fun sts -> isOlderThan cutoff sts.Metadata)
+                   |> Seq.map (fun sts -> sts.Metadata.Name)
+                   |> Seq.filter (fun name -> name.StartsWith("parallel-catchup-") && name.Contains("-stellar-core"))
+                   |> Seq.map (fun name -> name.Substring(0, name.LastIndexOf("-stellar-core")))
+                   |> Set.ofSeq do
+        LogInfo "Orphan sweep: helm uninstall %s" release
+
+        RunShellCommand [| "helm"
+                           "uninstall"
+                           release
+                           "-n"
+                           ns |]
+        |> ignore
 
     // 2. Delete the same resource type set the retired `clean` verb targeted.
     //    The order matches the old NamespaceContent.Cleanup so dependent

src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh

@@ -27,6 +27,14 @@ if job then redis.call("HSET", KEYS[3], job, ARGV[1]) end
 return job'
 
 while true; do
+# Stop claiming once the driver marks us, so it can remove us without interrupting a range.
+if [ "$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" SISMEMBER "$RELEASE_NAME-retiring" "$POD_NAME")" = "1" ]; then
+    echo "$(date) $POD_NAME is retiring; not claiming."
+    sleep $SLEEP_INTERVAL
+    continue
+fi
+
+
 # Claim the next job: atomically move it from the job queue to the progress
 # queue and record this pod as its owner. Our ranges are generated in the order
 # we want to run them from left to right, so we always pull from the left
@@ -84,8 +92,7 @@ if [ $CLAIM_EXIT_CODE -eq 0 ] && [ "$CLAIM_VALID" = true ]; then
     fi
 
     # Push metrics to redis in a transaction to ensure data consistency. Retry for 5min on failures
-    # Extract the pod ordinal (last hyphen-separated segment) from pod name like "release-name-stellar-core-0"
-    core_id=$(echo "$POD_NAME" | awk -F'-' '{print $NF}')
+    core_id="$WORKER_INDEX"
     # Validate core_id was extracted successfully
     if [ -z "$core_id" ]; then
         echo "Error: Failed to extract core_id from POD_NAME: $POD_NAME"