diff --git a/plugins/scheduler-k3s/functions.go b/plugins/scheduler-k3s/functions.go index faaddffca..4970ef4e7 100644 --- a/plugins/scheduler-k3s/functions.go +++ b/plugins/scheduler-k3s/functions.go @@ -134,6 +134,23 @@ type WaitForPodBySelectorRunningInput struct { Waiter func(ctx context.Context, clientset KubernetesClient, podName, namespace string) wait.ConditionWithContextFunc } +type WaitForPodBySelectorCompletedInput struct { + // Clientset is the kubernetes clientset + Clientset KubernetesClient + + // Namespace is the namespace to search in + Namespace string + + // LabelSelector is the label selector to search for + LabelSelector string + + // PodName is the pod name to search for + PodName string + + // Timeout is the timeout in seconds to wait for the pod to reach a terminal phase + Timeout float64 +} + type WaitForPodToExistInput struct { Clientset KubernetesClient Namespace string @@ -1981,6 +1998,52 @@ func isPodReady(ctx context.Context, clientset KubernetesClient, podName, namesp } } +// waitForPodBySelectorCompleted polls the first pod matching the input +// selector until it reaches a terminal phase (PodSucceeded / PodFailed) or +// the timeout elapses. Used after a Follow=true StreamLogs returns, to +// handle the kubelet -> apiserver phase propagation delay for short-lived +// run pods. +func waitForPodBySelectorCompleted(ctx context.Context, input WaitForPodBySelectorCompletedInput) (v1.Pod, error) { + pods, err := waitForPodToExist(ctx, WaitForPodToExistInput{ + Clientset: input.Clientset, + LabelSelector: input.LabelSelector, + Namespace: input.Namespace, + PodName: input.PodName, + RetryCount: 3, + }) + if err != nil { + return v1.Pod{}, fmt.Errorf("Error waiting for pod to exist: %w", err) + } + + if len(pods) == 0 { + return v1.Pod{}, fmt.Errorf("no pods in %s with selector %s", input.Namespace, input.LabelSelector) + } + + podName := pods[0].Name + if input.PodName != "" { + podName = input.PodName + } + + timeout := time.Duration(input.Timeout * float64(time.Second)) + var pod v1.Pod + err = wait.PollUntilContextTimeout(ctx, time.Second, timeout, true, func(ctx context.Context) (bool, error) { + p, err := input.Clientset.GetPod(ctx, GetPodInput{ + Name: podName, + Namespace: input.Namespace, + }) + if err != nil { + return false, err + } + pod = p + switch p.Status.Phase { + case v1.PodSucceeded, v1.PodFailed: + return true, nil + } + return false, nil + }) + return pod, err +} + // kubernetesNodeToNode converts a kubernetes node to a Node func kubernetesNodeToNode(node v1.Node) Node { roles := []string{} diff --git a/plugins/scheduler-k3s/triggers.go b/plugins/scheduler-k3s/triggers.go index 8714f0754..0a0182966 100644 --- a/plugins/scheduler-k3s/triggers.go +++ b/plugins/scheduler-k3s/triggers.go @@ -1585,17 +1585,15 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] return fmt.Errorf("Error streaming logs: %w", streamErr) } - pods, listErr := clientset.ListPods(ctx, ListPodsInput{ + selectedPod, waitErr := waitForPodBySelectorCompleted(ctx, WaitForPodBySelectorCompletedInput{ + Clientset: clientset, Namespace: namespace, LabelSelector: batchJobSelector, + Timeout: 10, }) - if listErr != nil { - return fmt.Errorf("Error getting pod: %w", listErr) + if waitErr != nil { + return fmt.Errorf("Pod did not reach terminal state after logs streamed: %w", waitErr) } - if len(pods) == 0 { - return fmt.Errorf("No pods found matching specified labels") - } - selectedPod := pods[0] switch selectedPod.Status.Phase { case v1.PodSucceeded: return nil @@ -1610,9 +1608,8 @@ func TriggerSchedulerRun(scheduler string, appName string, envCount int, args [] return fmt.Errorf("Unable to attach as the pod has already exited with a failed exit code: %s", status.State.Terminated.Message) } return fmt.Errorf("Unable to attach as the pod has already exited with a failed exit code") - default: - return fmt.Errorf("Unable to attach as the pod is in an unknown state: %s", selectedPod.Status.Phase) } + return fmt.Errorf("Unable to attach as the pod is in an unknown state: %s", selectedPod.Status.Phase) } if err != nil { if errors.Is(err, conditions.ErrPodCompleted) {