fix(k8s): resume Job watch after HTTP/2 stream drops

GKE's apiserver LB emits watch.Error "http2: client connection lost"
instead of closing the channel. That is a dead stream, not a failed Job;
treat it like 410 Gone and re-establish the watch so long validators
are not marked failed while still running.

Signed-off-by: Rohit Rajani <rorajani@nvidia.com>

Author: rorajani

pkg/k8s/pod/job.go

@@ -17,6 +17,7 @@ package pod
 import (
 	"context"
 	"log/slog"
+	"strings"
 	"time"
 
 	"github.com/NVIDIA/aicr/pkg/defaults"
@@ -49,24 +50,48 @@ func resumeContext(namespace, name string) map[string]any {
 }
 
 // isRetryableWatchError reports whether a watch.Error event is a routine,
-// resumable signal (HTTP 410 Gone / ResourceExpired — the apiserver compacted
-// past the watch's ResourceVersion) rather than a fatal stream error. Callers
-// treat a retryable error like a channel close and re-establish the watch;
-// anything else aborts the wait.
+// resumable signal rather than a fatal stream error. Callers treat a retryable
+// error like a channel close and re-establish the watch; anything else aborts
+// the wait.
 //
-// Against a real apiserver this is the *common* shape of a stale-ResourceVersion
-// rejection: the Watch call itself succeeds and the 410 arrives as the stream's
-// first ERROR event, not as a synchronous call error.
+// Retryable shapes:
+//   - HTTP 410 Gone / ResourceExpired — the apiserver compacted past the
+//     watch's ResourceVersion. Against a real apiserver this is the *common*
+//     shape of a stale-RV rejection: the Watch call itself succeeds and the
+//     410 arrives as the stream's first ERROR event, not as a synchronous
+//     call error.
+//   - Timeout / server-timeout / 429 / 503 — the apiserver or an LB asked
+//     the client to retry; the Job did not fail.
+//   - client-go StreamWatcher decode failures, including
+//     "http2: client connection lost". GKE (and any HTTP/2 kube-apiserver
+//     fronted by a load balancer) drops idle watch streams this way. The
+//     ERROR event is the same class as a channel close: the stream died;
+//     the Job did not. UAT run 32765635777 failed inference-perf on this
+//     exact Status while the AIPerf Job was still running.
+//
+// Generic InternalError (RBAC bugs, etcd faults, "boom") stays fatal so a
+// real defect is not retried until the deadline.
 func isRetryableWatchError(event watch.Event) bool {
 	if event.Type != watch.Error {
 		return false
 	}
 	err := apierrors.FromObject(event.Object)
-	return apierrors.IsResourceExpired(err) || apierrors.IsGone(err)
+	if err == nil {
+		return false
+	}
+	if apierrors.IsResourceExpired(err) || apierrors.IsGone(err) ||
+		apierrors.IsTimeout(err) || apierrors.IsServerTimeout(err) ||
+		apierrors.IsTooManyRequests(err) || apierrors.IsServiceUnavailable(err) {
+		return true
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "http2: client connection lost") ||
+		strings.Contains(msg, "unable to decode an event from the watch stream")
 }
 
 // resumeJobWatch reconnects a Job watch that ended — the channel closed or the
-// apiserver emitted a retryable 410 — before the Job reached a terminal state.
+// apiserver emitted a retryable watch.Error (410, HTTP/2 drop) — before the
+// Job reached a terminal state.
 //
 // It resyncs via a field-selected List rather than a Get, which is load-bearing:
 //   - The List's collection ResourceVersion (metadata.resourceVersion) is the

pkg/k8s/pod/job_test.go

@@ -456,6 +456,49 @@ func TestWaitForJobCompletion_RetryableWatchErrorResumes(t *testing.T) {
 	}
 }
 
+// TestWaitForJobCompletion_HTTP2DropWatchErrorResumes is the UAT flake from
+// run 32765635777: client-go StreamWatcher emits InternalError
+// "unable to decode an event from the watch stream: http2: client connection
+// lost" when GKE's apiserver LB drops the watch. That is a dead stream, not
+// a failed Job — resume and keep waiting.
+func TestWaitForJobCompletion_HTTP2DropWatchErrorResumes(t *testing.T) {
+	t.Parallel()
+
+	client := fake.NewSimpleClientset(&batchv1.Job{ //nolint:staticcheck
+		ObjectMeta: metav1.ObjectMeta{Name: "j", Namespace: "default", ResourceVersion: "1"},
+	})
+
+	var watches atomic.Int32
+	first := watch.NewFake()
+	second := watch.NewFake()
+	client.PrependWatchReactor("jobs", func(_ k8stesting.Action) (bool, watch.Interface, error) {
+		switch watches.Add(1) {
+		case 1:
+			return true, first, nil
+		case 2:
+			return true, second, nil
+		default:
+			return true, watch.NewFake(), nil
+		}
+	})
+
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		lost := apierrors.NewInternalError(stderrors.New(
+			"unable to decode an event from the watch stream: http2: client connection lost"))
+		first.Error(&lost.ErrStatus)
+		time.Sleep(30 * time.Millisecond)
+		second.Modify(terminalJobRV("2", batchv1.JobComplete))
+	}()
+
+	if err := pod.WaitForJobCompletion(context.Background(), client, "default", "j", 2*time.Second); err != nil {
+		t.Fatalf("expected nil after http2-drop watch error resume, got: %v", err)
+	}
+	if got := watches.Load(); got < 2 {
+		t.Errorf("expected at least 2 watch attempts (resume after http2 drop), got %d", got)
+	}
+}
+
 // TestWaitForJobCompletion_FatalWatchErrorReturns confirms a non-retryable
 // watch.Error still aborts the wait with ErrCodeInternal rather than resuming.
 func TestWaitForJobCompletion_FatalWatchErrorReturns(t *testing.T) {

pkg/k8s/pod/wait_internal_test.go

@@ -362,3 +362,70 @@ func TestWatchClosedContext(t *testing.T) {
 		}
 	})
 }
+
+func TestIsRetryableWatchError(t *testing.T) {
+	t.Parallel()
+
+	errorEvent := func(obj runtime.Object) watch.Event {
+		return watch.Event{Type: watch.Error, Object: obj}
+	}
+	statusOf := func(err error) runtime.Object {
+		var statusErr *apierrors.StatusError
+		if !stderrors.As(err, &statusErr) {
+			t.Fatalf("expected *apierrors.StatusError, got %T", err)
+		}
+		return &statusErr.ErrStatus
+	}
+
+	tests := []struct {
+		name  string
+		event watch.Event
+		want  bool
+	}{
+		{name: "added event is not retryable", event: watch.Event{Type: watch.Added}, want: false},
+		{
+			name:  "410 ResourceExpired is retryable",
+			event: errorEvent(statusOf(apierrors.NewResourceExpired("compacted"))),
+			want:  true,
+		},
+		{
+			name:  "410 Gone is retryable",
+			event: errorEvent(statusOf(apierrors.NewGone("gone"))),
+			want:  true,
+		},
+		{
+			name:  "503 ServiceUnavailable is retryable",
+			event: errorEvent(statusOf(apierrors.NewServiceUnavailable("apiserver down"))),
+			want:  true,
+		},
+		{
+			name:  "timeout is retryable",
+			event: errorEvent(statusOf(apierrors.NewTimeoutError("timed out", 1))),
+			want:  true,
+		},
+		{
+			name: "http2 client connection lost is retryable",
+			event: errorEvent(statusOf(apierrors.NewInternalError(
+				stderrors.New("unable to decode an event from the watch stream: http2: client connection lost")))),
+			want: true,
+		},
+		{
+			name:  "generic InternalError stays fatal",
+			event: errorEvent(statusOf(apierrors.NewInternalError(stderrors.New("boom")))),
+			want:  false,
+		},
+		{
+			name:  "Forbidden stays fatal",
+			event: errorEvent(statusOf(apierrors.NewForbidden(batchv1.Resource("jobs"), "j", stderrors.New("denied")))),
+			want:  false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			if got := isRetryableWatchError(tt.event); got != tt.want {
+				t.Errorf("isRetryableWatchError() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}