PodWatcher: retry net.OpError errors

k8s_pod_watcher.go

@@ -317,6 +317,14 @@ func (p *PodWatcher) Run(ctx context.Context) error {
 						t.Status().Code, watchStartErr)
 				}
 			default:
+				noe := (*net.OpError)(nil)
+				if errors.As(watchStartErr, &noe) {
+					// If it's a connection error of some sort, we want to backoff and retry
+					if sleepBackoff() {
+						return nil
+					}
+					continue
+				}
 				return fmt.Errorf("failed to startup watcher: %w", watchStartErr)
 			}
 			switch resync() {

k8s_pod_watcher_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"net"
 	"net/http"
+	"net/url"
 	"reflect"
 	"sync"
 	"testing"
@@ -769,6 +770,22 @@ func TestPodWatcherErrorRecovery(t *testing.T) {
 				&DeletePod{name: "foobar"},
 			},
 		},
+		{
+			name: "one_ready_then_dies_one_reconnect_network_error",
+			listRets: []listRetPI{{pi: []podInfo{{name: "foobar", ip: "10.42.42.42", labels: map[string]string{"app": "fimbat"},
+				ready: true, phase: k8score.PodRunning}}}},
+			watchRets: []watchRet{
+				{watch: []watchEvent{}},
+				{err: &url.Error{Op: "foo", Err: &net.OpError{}}},
+				{err: goneErr{}},
+				{watch: []watchEvent{}},
+			},
+			expectedEvents: []PodEvent{
+				&CreatePod{name: "foobar", IP: &net.IPAddr{IP: net.IPv4(10, 42, 42, 42)},
+					Def: genPod("foobar", "10.42.42.42", map[string]string{"app": "fimbat"}, true, k8score.PodRunning)},
+				&DeletePod{name: "foobar"},
+			},
+		},
 		{
 			name: "one_ready_then_dies_one_reconnect_change_error_no_watch_gone_error",
 			listRets: []listRetPI{{pi: []podInfo{{name: "foobar", ip: "10.42.42.42", labels: map[string]string{"app": "fimbat"},