diff --git a/src/CSLibrary/RemoteCommandRunner.cs b/src/CSLibrary/RemoteCommandRunner.cs index ca9b2f58..bc18a101 100644 --- a/src/CSLibrary/RemoteCommandRunner.cs +++ b/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 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 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(status)); } } } diff --git a/src/FSLibrary/MissionHistoryPubnetParallelCatchupV2.fs b/src/FSLibrary/MissionHistoryPubnetParallelCatchupV2.fs index 314c928b..e66f87fa 100644 --- a/src/FSLibrary/MissionHistoryPubnetParallelCatchupV2.fs +++ b/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 = 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: -stellar-core-0, -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 = + 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) (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 = 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("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 diff --git a/src/FSLibrary/StellarOrphanSweep.fs b/src/FSLibrary/StellarOrphanSweep.fs index dd56aaf5..12c5074f 100644 --- a/src/FSLibrary/StellarOrphanSweep.fs +++ b/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 diff --git a/src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh b/src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh index 89fd94d7..b391b76f 100644 --- a/src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh +++ b/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" diff --git a/src/MissionParallelCatchup/parallel_catchup_helm/templates/catchup_workers.yaml b/src/MissionParallelCatchup/parallel_catchup_helm/templates/catchup_workers.yaml index e861a3e3..5975f275 100644 --- a/src/MissionParallelCatchup/parallel_catchup_helm/templates/catchup_workers.yaml +++ b/src/MissionParallelCatchup/parallel_catchup_helm/templates/catchup_workers.yaml @@ -21,39 +21,43 @@ metadata: {{- end }} {{- end }} --- +{{- range $i := until (int $.Values.worker.replicas) }} apiVersion: apps/v1 kind: StatefulSet metadata: - name: {{ .Release.Name }}-stellar-core + name: {{ $.Release.Name }}-stellar-core-{{ $i }} labels: - app: {{ .Release.Name }}-stellar-core + app: {{ $.Release.Name }}-stellar-core + worker-index: {{ $i | quote }} spec: - serviceName: "{{ .Release.Name }}-stellar-core" + serviceName: "{{ $.Release.Name }}-stellar-core" podManagementPolicy: Parallel - replicas: {{ .Values.worker.replicas }} + replicas: 1 selector: matchLabels: - app: {{ .Release.Name }}-stellar-core + app: {{ $.Release.Name }}-stellar-core + worker-index: {{ $i | quote }} template: metadata: labels: - app: {{ .Release.Name }}-stellar-core + app: {{ $.Release.Name }}-stellar-core + worker-index: {{ $i | quote }} spec: - serviceAccountName: stellar-supercluster-{{ .Release.Name }} - {{- if or .Values.worker.requireNodeLabels .Values.worker.avoidNodeLabels }} + serviceAccountName: stellar-supercluster-{{ $.Release.Name }} + {{- if or $.Values.worker.requireNodeLabels $.Values.worker.avoidNodeLabels }} affinity: nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: nodeSelectorTerms: - matchExpressions: - {{- range .Values.worker.requireNodeLabels }} + {{- range $.Values.worker.requireNodeLabels }} - key: {{ .key }} operator: {{ .operator }} {{- with .values }} values: {{ toJson . }} {{- end }} {{- end }} - {{- range .Values.worker.avoidNodeLabels }} + {{- range $.Values.worker.avoidNodeLabels }} - key: {{ .key }} operator: {{ .operator }} {{- with .values }} @@ -61,26 +65,26 @@ spec: {{- end }} {{- end }} {{- end }} - {{- if .Values.worker.tolerateNodeTaints }} + {{- if $.Values.worker.tolerateNodeTaints }} tolerations: - {{- range .Values.worker.tolerateNodeTaints }} + {{- range $.Values.worker.tolerateNodeTaints }} - key: {{ .key }} effect: {{ .effect }} {{- end }} {{- end }} containers: - name: stellar-core - image: {{ .Values.worker.stellar_core_image }} + image: {{ $.Values.worker.stellar_core_image }} imagePullPolicy: Always resources: requests: - cpu: "{{ .Values.worker.resources.requests.cpu}}" - memory: "{{ .Values.worker.resources.requests.memory}}" - ephemeral-storage: "{{ .Values.worker.resources.requests.ephemeral_storage}}" + cpu: "{{ $.Values.worker.resources.requests.cpu}}" + memory: "{{ $.Values.worker.resources.requests.memory}}" + ephemeral-storage: "{{ $.Values.worker.resources.requests.ephemeral_storage}}" limits: - cpu: "{{ .Values.worker.resources.limits.cpu}}" - memory: "{{ .Values.worker.resources.limits.memory}}" - ephemeral-storage: "{{ .Values.worker.resources.limits.ephemeral_storage}}" + cpu: "{{ $.Values.worker.resources.limits.cpu}}" + memory: "{{ $.Values.worker.resources.limits.memory}}" + ephemeral-storage: "{{ $.Values.worker.resources.limits.ephemeral_storage}}" command: ["/bin/sh", "/scripts/worker.sh"] ports: - containerPort: 11626 @@ -90,10 +94,12 @@ spec: fieldRef: fieldPath: metadata.name - name: ASAN_OPTIONS - value: {{ .Values.worker.asanOptions | quote }} + value: {{ $.Values.worker.asanOptions | quote }} + - name: WORKER_INDEX + value: {{ $i | quote }} envFrom: - configMapRef: - name: {{ .Release.Name }}-worker-config + name: {{ $.Release.Name }}-worker-config volumeMounts: - name: config mountPath: /config @@ -104,17 +110,17 @@ spec: volumes: - name: config configMap: - name: {{ .Release.Name }}-stellar-core-config + name: {{ $.Release.Name }}-stellar-core-config - name: script configMap: - name: {{ .Release.Name }}-worker-script + name: {{ $.Release.Name }}-worker-script - emptyDir: {} name: data-volume - {{- if not .Values.worker.unevenSched }} + {{- if not $.Values.worker.unevenSched }} topologySpreadConstraints: - labelSelector: matchLabels: - app: {{ .Release.Name }}-stellar-core + app: {{ $.Release.Name }}-stellar-core # Note: maxSkew affects dynamic node scheduling with karpenter # See https://github.com/stellar/supercluster/issues/330 maxSkew: 2 @@ -122,6 +128,8 @@ spec: whenUnsatisfiable: DoNotSchedule {{- end }} --- +{{- end }} +--- apiVersion: v1 kind: ConfigMap metadata: