fix: wait for terminal pod phase on scheduler-k3s run
After streaming logs from a run pod on scheduler-k3s, the apiserver may still report `PodRunning` for a short window due to kubelet status propagation lag, causing `dokku run` to fail with `Unable to attach as the pod is in an unknown state: Running`. Wait briefly for the pod to reach a terminal phase before classifying the outcome.
This commit is contained in:
@@ -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{}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user