Add scale down to HistoryPubnetParallelCatchupV2

src/App/Program.fs

@@ -30,7 +30,7 @@ type PollOptions(kubeConfig: string, namespaceProperty: string option) =
     inherit KubeOption(kubeConfig, namespaceProperty)
 
 [<Verb("force-clean-namespace",
-       HelpText = "Delete every Service/ConfigMap/StatefulSet/Ingress/Job/DaemonSet/Deployment in the namespace and helm-uninstall any parallel-catchup-* releases. Destructive: meant for explicit recovery of a shared namespace, not for routine use.")>]
+       HelpText = "Delete every Service/ConfigMap/StatefulSet/Ingress/Job/DaemonSet/Deployment and ownerless Pod in the namespace and helm-uninstall any parallel-catchup-* releases. Destructive: meant for explicit recovery of a shared namespace, not for routine use.")>]
 type ForceCleanNamespaceOptions(kubeConfig: string, namespaceProperty: string option) =
     inherit KubeOption(kubeConfig, namespaceProperty)
 

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,10 @@ 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
 
 let jobMonitorHostName (context: MissionContext) =
     match context.jobMonitorExternalHost with
@@ -227,6 +231,8 @@ let installProject (context: MissionContext) =
                        "install"
                        helmReleaseName
                        helmChartPath
+                       "--namespace"
+                       context.namespaceProperty
                        "--values"
                        valuesFilePath
                        "--set"
@@ -236,22 +242,14 @@ 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 worker pods; a pod whose collection fails is skipped.
+let collectLogsFromPods (context: MissionContext) (podNames: string list) : unit =
     LogInfo "Collecting logs from %d worker pods to directory: %s" (List.length podNames) context.destination.Path
 
     for podName in podNames do
@@ -275,6 +273,7 @@ let collectLogsFromPods (context: MissionContext) =
                 command = command,
                 outputFilePath = outputFile
             )
+            |> ignore
 
             let fileInfo = FileInfo(outputFile)
 
@@ -307,7 +306,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 +318,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)
                 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 +402,11 @@ 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" helmReleaseName i ]
+
     let mutable timeoutLeft = jobMonitorStatusCheckTimeOutSecs
     let mutable timeBeforeNextMetricsCheck = jobMonitorMetricsCheckIntervalSecs
     let mutable stalledForSecs = 0
@@ -431,6 +439,35 @@ 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
+
+                // The job monitor owns the marking decision; we only collect logs and delete.
+                try
+                    let retirable = status.["retirable"] :?> JArray
+
+                    // Already-deleted names stay in the monitor's set, so filter by what is still live.
+                    let toRetire =
+                        retirable
+                        |> Seq.map (fun name -> name.ToString())
+                        |> Seq.filter livePods.Contains
+                        |> Seq.truncate maxRetiredPerPass
+                        |> List.ofSeq
+
+                    if not toRetire.IsEmpty then
+                        // /data is emptyDir, so collect before deleting or the logs are lost.
+                        collectLogsFromPods context toRetire
+
+                        for pod in toRetire do
+                            try
+                                context.kube.DeleteNamespacedPod(pod, context.namespaceProperty) |> ignore
+                            with :? k8s.Autorest.HttpOperationException as ex when
+                                ex.Response.StatusCode = Net.HttpStatusCode.NotFound ->
+                                LogInfo "Pod %s already gone" pod
+
+                        livePods <- Set.difference livePods (Set.ofList toRetire)
+                        LogInfo "Retired %d workers (%d outstanding)" toRetire.Length 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, which we fail the
                 // mission 2. the job monitor itself gets stuck unable to

src/FSLibrary/StellarOrphanSweep.fs

@@ -54,38 +54,39 @@ let private sweepKind
 
 // Delete resources older than `cutoff` in the given namespace. Targets the same
 // resource type set the retired `clean` verb did (Service, ConfigMap,
-// StatefulSet, Ingress, Job, DaemonSet, Deployment) and also helm-uninstalls
-// any `parallel-catchup-*` releases (PCv2) so helm's release secrets get
-// tidied along with the workloads.
+// StatefulSet, Ingress, Job, DaemonSet, Deployment) plus ownerless Pods, and
+// also helm-uninstalls any `parallel-catchup-*` releases (PCv2) so helm's
+// release secrets get tidied along with the workloads.
 //
 // Used by both the automatic on-startup orphan sweep (cutoff = now - 2 days)
 // and the explicit `force-clean-namespace` verb (cutoff = DateTime.MaxValue,
 // i.e. everything).
 let private sweepWithCutoff (cutoff: DateTime) (kube: Kubernetes) (ns: string) (apiRateLimit: int) : unit =
     LogInfo "Orphan sweep starting: namespace=%s cutoff=%s" ns (cutoff.ToString("o"))
 
-    // 1. helm-uninstall PCv2 releases whose StatefulSet is older than the cutoff.
+    // 1. helm-uninstall PCv2 releases whose worker pod is older than the cutoff.
     //    Done before the kubectl-style deletes so helm's release secrets get
     //    cleaned up properly, rather than left dangling pointing at deleted
     //    resources.
     ApiRateLimit.sleepUntilNextRateLimitedApiCallTime apiRateLimit
 
     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
+    let podItems = kube.ListNamespacedPod(namespaceParameter = ns).Items
+
+    for release in podItems
+                   |> Seq.filter (fun pod -> isOlderThan cutoff pod.Metadata)
+                   |> Seq.map (fun pod -> pod.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
@@ -111,6 +112,19 @@ let private sweepWithCutoff (cutoff: DateTime) (kube: Kubernetes) (ns: string) (
             |> ignore)
         cutoff
 
+    // PCv2 workers are bare pods, so nothing else in this sweep would reap them.
+    // Restricted to ownerless pods: everything else goes away with its controller.
+    // Listed here, after the helm uninstalls above, so pods they already removed are not deleted again.
+    sweepKind
+        apiRateLimit
+        "Pod"
+        (fun () ->
+            kube.ListNamespacedPod(namespaceParameter = ns).Items
+            |> Seq.filter (fun p -> isNull p.Metadata.OwnerReferences || p.Metadata.OwnerReferences.Count = 0)
+            |> Seq.map (fun p -> p.Metadata))
+        (fun n -> kube.DeleteNamespacedPod(namespaceParameter = ns, name = n) |> ignore)
+        cutoff
+
     sweepKind
         apiRateLimit
         "Deployment"

src/MissionParallelCatchup/job_monitor.py

@@ -1,5 +1,6 @@
 import os
 import redis
+import socket
 import requests
 import json
 import sys
@@ -26,6 +27,8 @@
 PROGRESS_QUEUE = os.getenv('PROGRESS_QUEUE', 'in_progress') #LIST
 METRICS = os.getenv('METRICS', 'metrics') # SET
 JOB_OWNERS = os.getenv('JOB_OWNERS', 'job_owners') # HASH
+RETIRING = os.getenv('RETIRING', 'retiring') # SET
+MIN_UNMARKED_WORKERS = int(os.getenv('MIN_UNMARKED_WORKERS', 8))
 WORKER_PREFIX = os.getenv('WORKER_PREFIX', 'stellar-core')
 NAMESPACE = os.getenv('NAMESPACE', 'default')
 WORKER_COUNT = int(os.getenv('WORKER_COUNT', 3))
@@ -77,6 +80,7 @@ def get_logging_level():
     'jobs_failed': [],
     'jobs_in_progress': [],
     'workers': [],
+    'retirable': [],
     'workers_up': 0,
     'workers_down': 0,
     'workers_refresh_duration': 0,
@@ -148,6 +152,14 @@ def ping_worker(pod_name, retries=1):
                 time.sleep(STUCK_JOB_PING_DELAY_SECS)
     return False
 
+def pod_exists(pod_name):
+    # Trailing dot skips the resolv.conf search list, so a miss costs one query.
+    try:
+        socket.gethostbyname(f"{pod_name}.{WORKER_PREFIX}.{NAMESPACE}.svc.cluster.local.")
+        return True
+    except socket.gaierror:
+        return False
+
 def update_status_and_metrics():
     global status
     mission_start_time = time.time()
@@ -174,8 +186,27 @@ def update_status_and_metrics():
                 queue_in_progress_count = len(jobs_in_progress)
                 queue_remain_count = redis_client.llen(JOB_QUEUE)
 
+            # --- Phase 1c: Mark surplus idle workers as retiring; worker.sh stops claiming once marked ---
+            # Names here are pod names ("{WORKER_PREFIX}-{i}") as worker.sh stores them in JOB_OWNERS.
+            busy = set(job_owners.values())
+            retiring = redis_client.smembers(RETIRING)
+            candidates = sorted(w for w in (f"{WORKER_PREFIX}-{i}" for i in range(WORKER_COUNT))
+                                if w not in busy and w not in retiring)
+            outstanding = queue_remain_count + queue_in_progress_count
+            keep = max(outstanding, MIN_UNMARKED_WORKERS)
+            to_mark = []
+            # Resolve only when the name count says something could be marked: at t=0 outstanding >= fleet, so zero lookups.
+            if len(busy - retiring) + len(candidates) > keep:
+                unmarked_idle = [w for w in candidates if pod_exists(w)]
+                to_mark = unmarked_idle[:max(0, len(busy - retiring) + len(unmarked_idle) - keep)]
+            if to_mark:
+                redis_client.sadd(RETIRING, *to_mark)
+                logger.info("Marked %d workers retiring (%d outstanding)", len(to_mark), outstanding)
+            # Marked on an earlier pass and still idle; names stay here after the driver deletes them.
+            retirable = sorted(retiring - busy)
+
             # --- Phase 2: Quick single-ping check of workers that own in-progress jobs ---
-            active_workers = set(job_owners.values())
+            active_workers = busy
             worker_statuses = []
             workers_up = 0
             workers_down = 0
@@ -231,6 +262,7 @@ def update_status_and_metrics():
                     'jobs_failed': jobs_failed,
                     'jobs_in_progress': jobs_in_progress,
                     'workers': worker_statuses,
+                    'retirable': retirable,
                     'workers_up': workers_up,
                     'workers_down': workers_down,
                     'workers_refresh_duration': workers_refresh_duration,

src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh

@@ -9,6 +9,7 @@ if [ -z "$FAILED_QUEUE" ]; then echo "FAILED_QUEUE not set"; exit 1; fi
 if [ -z "$SUCCESS_QUEUE" ]; then echo "SUCCESS_QUEUE not set"; exit 1; fi
 if [ -z "$METRICS" ]; then echo "METRICS not set"; exit 1; fi
 if [ -z "$JOB_OWNERS" ]; then echo "JOB_OWNERS not set"; exit 1; fi
+if [ -z "$RETIRING" ]; then echo "RETIRING not set"; exit 1; fi
 if [ -z "$RELEASE_NAME" ]; then echo "RELEASE_NAME not set"; exit 1; fi
 if [ -z "$POD_NAME" ]; then echo "POD_NAME not set"; exit 1; fi
 
@@ -27,6 +28,15 @@ if job then redis.call("HSET", KEYS[3], job, ARGV[1]) end
 return job'
 
 while true; do
+# Stop claiming once the job monitor marks us, so the driver can remove us without interrupting a range.
+# Fail closed: anything but an explicit 0 (marked, or redis-cli error) means don't claim.
+if [ "$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" SISMEMBER "$RETIRING" "$POD_NAME")" != "0" ]; then
+    echo "$(date) $POD_NAME is retiring or redis unreachable; 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 +94,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"