diff --git a/pkg/dind/containers.go b/pkg/dind/containers.go index 74cfd56..94ad577 100644 --- a/pkg/dind/containers.go +++ b/pkg/dind/containers.go @@ -801,32 +801,42 @@ func (s *Server) handleContainerStart(w http.ResponseWriter, r *http.Request, id netns := fmt.Sprintf("/proc/%d/ns/net", pid) result, err := s.network.Setup(ctx, id, netns) if err != nil { - s.log.Warn("failed to setup network for dind container", "id", id, "error", err) - } else { - entry.NetNS = result.NetNS - entry.IP = result.IP - s.log.Info("network attached to dind container", "id", id, "ip", entry.IP) - - // Update the "bridge" network entry with the real CNI-assigned IP. - s.mu.Lock() - if info, ok := entry.Networks["bridge"]; ok { - info.IPAddress = result.IP - info.MacAddress = generateMAC(result.IP) - entry.Networks["bridge"] = info - } - if br := s.defaultNetwork(); br != nil { - br.Containers[id] = result.IP - } - s.mu.Unlock() - - // Register this sibling as a member of the job so the runner and the - // job's other `services:`/sibling containers can reach it, while the - // bridge keeps other jobs out. Keyed by the same container id used - // for Setup/Teardown. Best-effort: a failure only limits within-job - // reach and never opens a cross-job path. - if jErr := s.network.JoinJobNetwork(id, s.jobID, result.IP); jErr != nil { - s.log.Warn("failed to register dind container in job network (intra-job reach may be limited)", "id", id, "error", jErr) + // Fail the start. A container left running without an IP has no + // port forwards either, so the job just waits on a port that never + // opens (ephpm run 39690226239 sat 4 minutes on one). + s.log.Error("failed to setup network for dind container", "id", id, "error", err) + if _, delErr := task.Delete(ctx, client.WithProcessKill); delErr != nil { + s.log.Debug("task cleanup after failed network setup", "error", delErr) } + entry.Task = nil + writeJSON(w, http.StatusInternalServerError, map[string]string{ + "message": fmt.Sprintf("setting up network for container %s: %v", id, err), + }) + return + } + entry.NetNS = result.NetNS + entry.IP = result.IP + s.log.Info("network attached to dind container", "id", id, "ip", entry.IP) + + // Update the "bridge" network entry with the real CNI-assigned IP. + s.mu.Lock() + if info, ok := entry.Networks["bridge"]; ok { + info.IPAddress = result.IP + info.MacAddress = generateMAC(result.IP) + entry.Networks["bridge"] = info + } + if br := s.defaultNetwork(); br != nil { + br.Containers[id] = result.IP + } + s.mu.Unlock() + + // Register this sibling as a member of the job so the runner and the + // job's other `services:`/sibling containers can reach it, while the + // bridge keeps other jobs out. Keyed by the same container id used + // for Setup/Teardown. Best-effort: a failure only limits within-job + // reach and never opens a cross-job path. + if jErr := s.network.JoinJobNetwork(id, s.jobID, result.IP); jErr != nil { + s.log.Warn("failed to register dind container in job network (intra-job reach may be limited)", "id", id, "error", jErr) } } @@ -840,8 +850,10 @@ func (s *Server) handleContainerStart(w http.ResponseWriter, r *http.Request, id } } - if err := task.Start(ctx); err != nil { - // Clean up on failure. + // abortStart undoes everything above and fails the request, leaving the + // container in "created" exactly as Docker does when a start fails. + abortStart := func(msg string) { + s.removePortForwards(entry) if _, delErr := task.Delete(ctx, client.WithProcessKill); delErr != nil { s.log.Debug("task cleanup after failed start", "error", delErr) } @@ -854,9 +866,25 @@ func (s *Server) handleContainerStart(w http.ResponseWriter, r *http.Request, id entry.IP = "" } entry.Task = nil - writeJSON(w, http.StatusInternalServerError, map[string]string{ - "message": fmt.Sprintf("starting task: %v", err), - }) + writeJSON(w, http.StatusInternalServerError, map[string]string{"message": msg}) + } + + // Publish port bindings before the task starts, so a binding that cannot + // be honoured fails the start instead of leaving a running container whose + // port never opens. KIND's `kind create cluster` creates the kindest/node + // with -p 127.0.0.1::6443 and writes a kubeconfig pointing at that + // 127.0.0.1:; without the forward in the runner's netns, tilt + + // kubectl get connection refused. Listening early is harmless: a client + // that connects before the service is up is refused, just as it would be + // at the service's own port. + if err := s.installPortForwards(entry); err != nil { + s.log.Error("port forward install failed", "id", id, "error", err) + abortStart(fmt.Sprintf("starting container %s: %v", id, err)) + return + } + + if err := task.Start(ctx); err != nil { + abortStart(fmt.Sprintf("starting task: %v", err)) return } @@ -871,12 +899,6 @@ func (s *Server) handleContainerStart(w http.ResponseWriter, r *http.Request, id close(entry.started) } - // Install DNAT rules for any port bindings. KIND's `kind create cluster` - // creates the kindest/node with -p 127.0.0.1::6443 and writes a - // kubeconfig pointing at that 127.0.0.1:. Without DNAT in the - // runner's netns, tilt + kubectl get connection refused. - s.installPortForwards(entry) - // Monitor for unexpected exit so we can diagnose crashes. bgCtx := namespaces.WithNamespace(context.Background(), s.jobNamespace) exitCh, waitErr := task.Wait(bgCtx) @@ -1393,12 +1415,20 @@ func (s *Server) assignContainerNetwork(entry *containerEntry, req createRequest // installPortForwards starts userspace TCP proxies for every PortBindings // entry on the container. Each proxy listens on hostIP:hostPort inside the -// runner's net namespace and forwards to entry.IP:containerPort. Best-effort -// — failures log a warning and don't abort container start. -func (s *Server) installPortForwards(entry *containerEntry) { - if s.runnerNetNS == "" || entry.IP == "" || len(entry.PortBindings) == 0 { - return +// runner's net namespace and forwards to entry.IP:containerPort. +// +// A binding that cannot be honoured is an error, as it is for `docker run -p` +// ("port is already allocated"): the caller fails the start rather than run a +// container whose published port silently never opens. On error the proxies +// already started for this container are left in entry.PortForwards for the +// caller's cleanup to stop. +func (s *Server) installPortForwards(entry *containerEntry) error { + if s.runnerNetNS == "" || len(entry.PortBindings) == 0 { + return nil } + // A container started a second time must not collide with its own + // listeners from the first start. + s.removePortForwards(entry) for containerSpec, bindings := range entry.PortBindings { // containerSpec is e.g. "6443/tcp"; we only handle tcp. port, proto, ok := strings.Cut(containerSpec, "/") @@ -1409,14 +1439,12 @@ func (s *Server) installPortForwards(entry *containerEntry) { if b.HostPort == "" { continue } - stop, err := startPortForwardProxy(s.runnerNetNS, b.HostIP, b.HostPort, entry.IP, port) + if entry.IP == "" { + return fmt.Errorf("cannot publish %s:%s: container has no IP to forward to", b.HostIP, b.HostPort) + } + stop, err := startPortForwardProxy(s.runnerNetNS, b.HostIP, b.HostPort, entry.IP, port, s.log) if err != nil { - s.log.Warn("port forward install failed", - "container", entry.ID[:12], - "host", b.HostIP+":"+b.HostPort, - "target", entry.IP+":"+port, - "error", err) - continue + return fmt.Errorf("publishing %s:%s -> %s:%s: %w", b.HostIP, b.HostPort, entry.IP, port, err) } entry.PortForwards = append(entry.PortForwards, stop) s.log.Info("port forward installed", @@ -1425,6 +1453,7 @@ func (s *Server) installPortForwards(entry *containerEntry) { "target", entry.IP+":"+port) } } + return nil } // removePortForwards stops every running port-forward proxy goroutine for diff --git a/pkg/dind/portforward.go b/pkg/dind/portforward.go index ae2a030..166ebec 100644 --- a/pkg/dind/portforward.go +++ b/pkg/dind/portforward.go @@ -6,14 +6,23 @@ import ( "errors" "fmt" "io" + "log/slog" "net" "os" "runtime" "sync" + "time" "golang.org/x/sys/unix" ) +// forwardDialTimeout bounds the dial to the container behind a published +// port. Without it a dial whose SYN is never answered blocks for the kernel's +// ~2 minute connect timeout while the client, already accepted on 127.0.0.1, +// sees an open connection that never sends a byte. A var so tests can +// shorten it. +var forwardDialTimeout = 5 * time.Second + // startPortForwardProxy starts a userspace TCP proxy that listens on // hostIP:hostPort *inside the runner container's net namespace* and forwards // every accepted connection to targetIP:targetPort (reachable from main ns @@ -31,13 +40,16 @@ import ( // // Returns a stop function that closes the listener (idempotent), or an error // if setup fails. -func startPortForwardProxy(netnsPath, hostIP, hostPort, targetIP, targetPort string) (stop func(), err error) { +func startPortForwardProxy(netnsPath, hostIP, hostPort, targetIP, targetPort string, log *slog.Logger) (stop func(), err error) { if netnsPath == "" { return nil, errors.New("runner netns not set") } if hostIP == "" { hostIP = "127.0.0.1" } + if log == nil { + log = slog.Default() + } // Channel to receive (listener, error) from the namespace-locked goroutine. type listenResult struct { @@ -74,16 +86,24 @@ func startPortForwardProxy(netnsPath, hostIP, hostPort, targetIP, targetPort str } resultCh <- listenResult{listener: listener} - // Accept loop runs in the runner's netns. Each forwarded connection - // dials targetIP:targetPort from THIS namespace too — the kindest/node - // is on the same CNI bridge, so the runner can reach it directly. + // Accept loop runs in the runner's netns. Each accepted connection is + // handed to a new goroutine, which is NOT locked to this thread, so + // the dial to targetIP:targetPort goes out from the daemon's own + // namespace — the container sits on the CNI bridge the daemon owns, + // so it is reachable from there. target := net.JoinHostPort(targetIP, targetPort) + listen := net.JoinHostPort(hostIP, hostPort) for { conn, err := listener.Accept() if err != nil { return } - go forwardConn(conn, target) + go func() { + if err := forwardConn(conn, target); err != nil { + log.Warn("port forward: cannot reach container; closed the client connection", + "listen", listen, "target", target, "error", err) + } + }() } }() @@ -102,12 +122,16 @@ func startPortForwardProxy(netnsPath, hostIP, hostPort, targetIP, targetPort str return stop, nil } -func forwardConn(client net.Conn, target string) { +// forwardConn relays client to target until either side closes. If target +// cannot be dialed within forwardDialTimeout it closes client and returns the +// dial error, so the client sees the failure at once instead of an open +// connection that never answers. +func forwardConn(client net.Conn, target string) error { defer func() { _ = client.Close() }() - server, err := net.Dial("tcp", target) + server, err := net.DialTimeout("tcp", target, forwardDialTimeout) if err != nil { - return + return err } defer func() { _ = server.Close() }() @@ -121,4 +145,5 @@ func forwardConn(client net.Conn, target string) { done <- struct{}{} }() <-done + return nil } diff --git a/pkg/dind/portforward_install_test.go b/pkg/dind/portforward_install_test.go new file mode 100644 index 0000000..e57b139 --- /dev/null +++ b/pkg/dind/portforward_install_test.go @@ -0,0 +1,63 @@ +package dind + +import ( + "io" + "log/slog" + "strings" + "testing" +) + +// TestInstallPortForwards_NoIPIsAnError pins the fix for ephpm run +// 39690226239: a postgres sibling whose CNI setup failed started with no IP, +// its -p binding was skipped without a word, and the job waited minutes on a +// port that never opened. A binding that cannot be honoured must now be an +// error the start handler turns into a failed `docker start`. +func TestInstallPortForwards_NoIPIsAnError(t *testing.T) { + s := &Server{ + runnerNetNS: "/proc/self/ns/net", + log: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + entry := &containerEntry{ + ID: "27bdc8173467b088e1a75928c0c067a9", + PortBindings: map[string][]portBinding{ + "5432/tcp": {{HostIP: "127.0.0.1", HostPort: "26241"}}, + }, + } + + err := s.installPortForwards(entry) + if err == nil { + t.Fatal("installPortForwards with no container IP returned nil; want an error") + } + if !strings.Contains(err.Error(), "127.0.0.1:26241") { + t.Errorf("error %q does not name the binding it could not publish", err) + } + if len(entry.PortForwards) != 0 { + t.Errorf("got %d forwards installed, want 0", len(entry.PortForwards)) + } +} + +// TestInstallPortForwards_NothingToPublish confirms the no-op cases stay +// no-ops: no bindings, a binding with no host port (docker picks one), and a +// server that was never told the runner's netns (Windows siblings). +func TestInstallPortForwards_NothingToPublish(t *testing.T) { + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + cases := []struct { + name string + netns string + ports map[string][]portBinding + }{ + {"no bindings", "/proc/self/ns/net", nil}, + {"no host port", "/proc/self/ns/net", map[string][]portBinding{"80/tcp": {{HostIP: "127.0.0.1"}}}}, + {"udp only", "/proc/self/ns/net", map[string][]portBinding{"53/udp": {{HostIP: "127.0.0.1", HostPort: "5353"}}}}, + {"no runner netns", "", map[string][]portBinding{"80/tcp": {{HostIP: "127.0.0.1", HostPort: "8080"}}}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s := &Server{runnerNetNS: tc.netns, log: log} + entry := &containerEntry{ID: "0123456789abcdef", PortBindings: tc.ports} + if err := s.installPortForwards(entry); err != nil { + t.Fatalf("installPortForwards: %v", err) + } + }) + } +} diff --git a/pkg/dind/portforward_other.go b/pkg/dind/portforward_other.go index 4dfee22..1cde1c3 100644 --- a/pkg/dind/portforward_other.go +++ b/pkg/dind/portforward_other.go @@ -2,12 +2,15 @@ package dind -import "errors" +import ( + "errors" + "log/slog" +) // startPortForwardProxy is a no-op on non-Linux platforms — the dind subsystem // only runs inside the Linux VM, so this build is reached only by Windows / // macOS host-side compiles where the Linux ephemerd-linux is the one that // actually serves dind. -func startPortForwardProxy(_, _, _, _, _ string) (func(), error) { +func startPortForwardProxy(_, _, _, _, _ string, _ *slog.Logger) (func(), error) { return nil, errors.New("port forwarding only implemented on Linux") } diff --git a/pkg/dind/portforward_test.go b/pkg/dind/portforward_test.go index c18ab8a..24ebbe9 100644 --- a/pkg/dind/portforward_test.go +++ b/pkg/dind/portforward_test.go @@ -44,7 +44,7 @@ func TestForwardConn_Bidirectional(t *testing.T) { done := make(chan struct{}) go func() { - forwardConn(server, target.Addr().String()) + _ = forwardConn(server, target.Addr().String()) close(done) }() @@ -87,7 +87,7 @@ func TestForwardConn_TargetUnreachable(t *testing.T) { done := make(chan struct{}) go func() { - forwardConn(server, "127.0.0.1:1") // port 1 is reserved → ECONNREFUSED + _ = forwardConn(server, "127.0.0.1:1") // port 1 is reserved → ECONNREFUSED close(done) }() @@ -119,7 +119,7 @@ func TestForwardConn_TargetUnreachable(t *testing.T) { // when called without a netns path — guarding against misconfiguration where // the dind server hasn't been told about the runner's namespace yet. func TestStartPortForwardProxy_NoNetns(t *testing.T) { - stop, err := startPortForwardProxy("", "127.0.0.1", "1234", "10.0.0.1", "80") + stop, err := startPortForwardProxy("", "127.0.0.1", "1234", "10.0.0.1", "80", nil) if err == nil { if stop != nil { stop() @@ -131,7 +131,7 @@ func TestStartPortForwardProxy_NoNetns(t *testing.T) { // TestStartPortForwardProxy_InvalidNetns confirms a bad netns path produces // an error rather than a partially-initialized proxy. func TestStartPortForwardProxy_InvalidNetns(t *testing.T) { - stop, err := startPortForwardProxy("/proc/self/ns/nonexistent", "127.0.0.1", "1234", "10.0.0.1", "80") + stop, err := startPortForwardProxy("/proc/self/ns/nonexistent", "127.0.0.1", "1234", "10.0.0.1", "80", nil) if err == nil { if stop != nil { stop() @@ -177,7 +177,7 @@ func TestStartPortForwardProxy_OwnNetns(t *testing.T) { stop, err := startPortForwardProxy( "/proc/self/ns/net", "127.0.0.1", itoa(addr.Port), - "127.0.0.1", itoa(targetAddr.Port), + "127.0.0.1", itoa(targetAddr.Port), nil, ) if err != nil { t.Skipf("setns into own ns failed (likely sandbox restriction): %v", err) @@ -217,3 +217,39 @@ func itoa(n int) string { } return string(b[i:]) } + +// TestForwardConn_DialIsBounded pins the dial timeout. Before it, a client +// accepted on 127.0.0.1 whose container never answered the SYN sat on an +// open, silent connection for the kernel's ~2 minute connect timeout (ephpm +// run 39710247641 saw PDO hang until max_execution_time). 192.0.2.1 is +// TEST-NET-1: it either black-holes the SYN (timeout) or is unroutable +// (immediate error); either way forwardConn must give up fast and close the +// client. +func TestForwardConn_DialIsBounded(t *testing.T) { + old := forwardDialTimeout + forwardDialTimeout = 200 * time.Millisecond + t.Cleanup(func() { forwardDialTimeout = old }) + + client, server := net.Pipe() + defer func() { _ = client.Close() }() + + errCh := make(chan error, 1) + go func() { errCh <- forwardConn(server, "192.0.2.1:3306") }() + + select { + case err := <-errCh: + if err == nil { + t.Fatal("forwardConn to an unreachable target returned nil; want the dial error") + } + case <-time.After(3 * time.Second): + t.Fatal("forwardConn still dialing after 3s; the dial timeout is not applied") + } + + // The pipe refusing a deadline already means forwardConn closed it. + if err := client.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + return + } + if _, err := client.Read(make([]byte, 1)); err == nil { + t.Error("client still open after the dial failed; want it closed") + } +} diff --git a/pkg/networking/network_linux.go b/pkg/networking/network_linux.go index e80a7f2..8d74461 100644 --- a/pkg/networking/network_linux.go +++ b/pkg/networking/network_linux.go @@ -87,8 +87,18 @@ func (l *linuxNetworking) init(cfg Config) error { } func (l *linuxNetworking) setup(ctx context.Context, id string, netns string) (*SetupResult, error) { - result, err := l.cni.Setup(ctx, id, netns) + // A failed ADD can leave an allocated IP or a veth behind; CNI DEL is + // idempotent, so release unconditionally after any failure. + undo := func() { + if rmErr := l.cni.Remove(ctx, id, netns); rmErr != nil { + l.cfg.Log.Debug("CNI cleanup after failed setup", "id", id, "error", rmErr) + } + } + result, err := retrySetup(ctx, + func() (*cni.Result, error) { return l.cni.Setup(ctx, id, netns) }, + undo) if err != nil { + undo() return nil, fmt.Errorf("CNI setup for %s: %w", id, err) } diff --git a/pkg/networking/retry.go b/pkg/networking/retry.go new file mode 100644 index 0000000..ac40bc7 --- /dev/null +++ b/pkg/networking/retry.go @@ -0,0 +1,52 @@ +package networking + +import ( + "context" + "strings" + "time" +) + +// setupAttempts and setupRetryDelay bound how hard a CNI ADD is retried +// after a transient failure. Vars, not consts, so tests can shrink the delay. +var ( + setupAttempts = 3 + setupRetryDelay = 200 * time.Millisecond +) + +// isTransientSetupError reports whether a CNI ADD failed for a reason that a +// second attempt can clear. +// +// The one seen on the fleet is EINTR from a plugin's netlink calls: the +// plugin runs as a separate process and reports the errno as text, so the +// error reaches us as "plugin type="loopback" failed (add): interrupted +// system call" rather than as a wrapped syscall.Errno. It shows up when +// other containers on the host are being wired up at the same moment +// (ducks, 2026-10-22: a postgres sibling started with no IP and no port +// forward because of it). +func isTransientSetupError(err error) bool { + return err != nil && strings.Contains(err.Error(), "interrupted system call") +} + +// retrySetup runs attempt until it succeeds, fails with a non-transient +// error, or setupAttempts is used up. Between attempts it runs undo so a +// half-finished CNI ADD (an allocated IP, a created veth) is released +// before the next one starts. +func retrySetup[T any](ctx context.Context, attempt func() (T, error), undo func()) (T, error) { + var ( + res T + err error + ) + for i := 0; i < setupAttempts; i++ { + res, err = attempt() + if err == nil || !isTransientSetupError(err) || i == setupAttempts-1 { + return res, err + } + undo() + select { + case <-ctx.Done(): + return res, err + case <-time.After(setupRetryDelay): + } + } + return res, err +} diff --git a/pkg/networking/retry_test.go b/pkg/networking/retry_test.go new file mode 100644 index 0000000..bc0c949 --- /dev/null +++ b/pkg/networking/retry_test.go @@ -0,0 +1,62 @@ +package networking + +import ( + "context" + "errors" + "testing" + "time" +) + +func shortRetryDelay(t *testing.T) { + t.Helper() + old := setupRetryDelay + setupRetryDelay = time.Millisecond + t.Cleanup(func() { setupRetryDelay = old }) +} + +// The EINTR seen on ducks (2026-10-22) arrives as text from the plugin +// process, not as a wrapped errno. +var errEINTR = errors.New(`plugin type="loopback" failed (add): interrupted system call`) + +func TestRetrySetup_RecoversFromEINTR(t *testing.T) { + shortRetryDelay(t) + calls, undos := 0, 0 + got, err := retrySetup(context.Background(), func() (string, error) { + calls++ + if calls == 1 { + return "", errEINTR + } + return "10.88.0.5", nil + }, func() { undos++ }) + if err != nil { + t.Fatalf("retrySetup: %v", err) + } + if got != "10.88.0.5" || calls != 2 || undos != 1 { + t.Errorf("got %q after %d calls and %d undos; want 10.88.0.5 after 2 calls and 1 undo", got, calls, undos) + } +} + +func TestRetrySetup_DoesNotRetryOtherErrors(t *testing.T) { + shortRetryDelay(t) + calls := 0 + want := errors.New(`plugin type="bridge" failed (add): no IP addresses available in range set`) + _, err := retrySetup(context.Background(), func() (string, error) { + calls++ + return "", want + }, func() { t.Error("undo ran for a non-transient error") }) + if !errors.Is(err, want) || calls != 1 { + t.Errorf("got err=%v after %d calls; want the original error after 1 call", err, calls) + } +} + +func TestRetrySetup_GivesUp(t *testing.T) { + shortRetryDelay(t) + calls := 0 + _, err := retrySetup(context.Background(), func() (string, error) { + calls++ + return "", errEINTR + }, func() {}) + if !errors.Is(err, errEINTR) || calls != setupAttempts { + t.Errorf("got err=%v after %d calls; want EINTR after %d calls", err, calls, setupAttempts) + } +}