diff --git a/internal/guest/runtime/hcsv2/container.go b/internal/guest/runtime/hcsv2/container.go index 3695fa6647..e759b3e2a9 100644 --- a/internal/guest/runtime/hcsv2/container.go +++ b/internal/guest/runtime/hcsv2/container.go @@ -89,6 +89,10 @@ type Container struct { // sandboxRoot is the root directory of the pod within the guest. // Used during cleanup to unmount sandbox-specific paths. sandboxRoot string + + // stdioReconnectStopped, when set, stops the init process stdout/stderr + // relay reconnect loop; set on teardown. + stdioReconnectStopped atomic.Bool } func (c *Container) Start(ctx context.Context, conSettings stdio.ConnectionSettings) (_ int, err error) { @@ -127,6 +131,15 @@ func (c *Container) Start(ctx context.Context, conSettings stdio.ConnectionSetti return -1, err } + // Keep the init process's stdout/stderr flowing across a host disconnect by + // re-dialing the same vsock port; cancelled on container teardown. + if stdioSet.Out != nil && conSettings.StdOut != nil { + stdioSet.Out = transport.NewReconnectConnection(t, *conSettings.StdOut, stdioSet.Out, &c.stdioReconnectStopped) + } + if stdioSet.Err != nil && conSettings.StdErr != nil { + stdioSet.Err = transport.NewReconnectConnection(t, *conSettings.StdErr, stdioSet.Err, &c.stdioReconnectStopped) + } + if c.initProcess.spec.Terminal { ttyr := c.container.Tty() ttyr.ReplaceConnectionSet(stdioSet) @@ -237,12 +250,14 @@ func (c *Container) Kill(ctx context.Context, signal syscall.Signal) error { return err } c.setExitType(signal) + c.stdioReconnectStopped.Store(true) return nil } func (c *Container) Delete(ctx context.Context) error { entity := log.G(ctx).WithField(logfields.ContainerID, c.id) entity.Info("opengcs::Container::Delete") + c.stdioReconnectStopped.Store(true) if c.isSandbox && c.sandboxRoot != "" { // remove user mounts in sandbox container if err := storage.UnmountAllInPath(ctx, specGuest.SandboxMountsDirFromRoot(c.sandboxRoot), true); err != nil { diff --git a/internal/guest/transport/reconnect.go b/internal/guest/transport/reconnect.go new file mode 100644 index 0000000000..d836feddf8 --- /dev/null +++ b/internal/guest/transport/reconnect.go @@ -0,0 +1,192 @@ +//go:build linux +// +build linux + +package transport + +import ( + gerrors "errors" + "os" + "sync" + "sync/atomic" + "syscall" + "time" + + "github.com/sirupsen/logrus" +) + +// stdioReconnectInterval is the fixed cadence for re-dialing a dropped connection. +const stdioReconnectInterval = 100 * time.Millisecond + +// stdioReplayBytes is how many of the most-recently-written bytes we keep around +// so we can re-send them after reconnecting. When the host connection drops, any +// bytes that write() already accepted but the host never read are gone. Re-sending +// this tail recovers them (at the cost of re-sending some the host did read). +// +// So it has to be at least as large as the most data that can be "in flight" - +// buffered somewhere between our write() and the host actually reading it - when a +// drop happens. The important case is the live-migration blackout: the guest VM is +// paused and the host tears down its side of the socket, discarding whatever it had +// buffered. That in-flight data sits in two fixed-size OS buffers: +// +// - Guest send buffer (guest -> host): 24 KiB. Fixed by the Linux hv_sock driver +// (RINGBUFFER_HVS_SND_SIZE); we never enlarge it with SO_SNDBUF. +// - Host receive buffer: 64 KiB. The Windows hvsocket default (SO_RCVBUF). +// +// So the most we can lose in one drop is about 24 + 64 = ~88 KiB. +// 128 KiB rounds that up with headroom. These two sizes are set by the OS and +// do not grow with how long the container runs or how fast it logs, +// so a bigger buffer wouldn't help - and a smaller one could actually drop +// logs if the host falls behind and both buffers are full when the connection drops. +const stdioReplayBytes = 128 * 1024 // 128 KiB + +// reconnectConnection is a Connection that transparently re-dials the same port +// when a write fails because the host end of an established connection went away. +type reconnectConnection struct { + t Transport + port uint32 + stopped *atomic.Bool + + mu sync.Mutex + conn Connection + replay *replayBuffer +} + +var _ Connection = &reconnectConnection{} + +// NewReconnectConnection wraps a live host connection so a container's outbound +// stream keeps flowing across a brief host disappearance by re-dialing and resuming. +func NewReconnectConnection(t Transport, port uint32, conn Connection, stopped *atomic.Bool) Connection { + return &reconnectConnection{t: t, port: port, conn: conn, stopped: stopped, replay: newReplayBuffer(stdioReplayBytes)} +} + +// current returns the connection in use, waiting out any in-progress re-dial so it +// never hands back a broken one. +func (c *reconnectConnection) current() Connection { + c.mu.Lock() + defer c.mu.Unlock() + + return c.conn +} + +// Write sends bytes to the host, transparently reconnecting and resuming on a +// mid-stream drop; a caller only sees an error once reconnection is abandoned at teardown. +func (c *reconnectConnection) Write(b []byte) (int, error) { + // Retain each chunk we successfully write so redial can re-send it if the + // drop lost it (see stdioReplayBytes). written tracks progress through b. + written := 0 + for { + n, err := c.current().Write(b[written:]) + if n > 0 { + c.replay.append(b[written : written+n]) + written += n + } + if err == nil { + return written, nil + } + if !isDisconnectErr(err) { + return written, err + } + if rerr := c.redial(); rerr != nil { + return written, err + } + } +} + +// redial closes the dead connection and re-dials the same host endpoint every 100ms +// until reconnected or teardown is signalled, holding the lock so current never sees the broken conn. +func (c *reconnectConnection) redial() error { + c.mu.Lock() + defer c.mu.Unlock() + + if c.conn != nil { + _ = c.conn.Close() + } + for { + if c.stopped.Load() { + return gerrors.New("stdio reconnect stopped") + } + conn, err := c.t.Dial(c.port) + if err != nil { + time.Sleep(stdioReconnectInterval) + continue + } + // Re-send the retained tail to recover bytes the dropped connection lost. + // This may re-deliver some the host already read, which is fine for logs. + if tail := c.replay.bytes(); len(tail) > 0 { + if _, werr := conn.Write(tail); werr != nil { + _ = conn.Close() + time.Sleep(stdioReconnectInterval) + continue + } + } + c.conn = conn + logrus.WithField("port", c.port).Info("opengcs::reconnectConnection - reconnected stdio") + return nil + } +} + +// Read reads from the current host connection; used only by the relay's clean-close handshake. +func (c *reconnectConnection) Read(b []byte) (int, error) { return c.current().Read(b) } + +// Close tears down the current host connection. +func (c *reconnectConnection) Close() error { return c.current().Close() } + +// CloseRead closes the read half of the current host connection. +func (c *reconnectConnection) CloseRead() error { return c.current().CloseRead() } + +// CloseWrite closes the write half, signalling end-of-stream to the host. +func (c *reconnectConnection) CloseWrite() error { return c.current().CloseWrite() } + +// File exposes the current connection's file; a later reconnect is not reflected in +// an fd already handed out, so this is only meaningful before any drop. +func (c *reconnectConnection) File() (*os.File, error) { return c.current().File() } + +// isDisconnectErr reports whether a write failure means the host end went away +// (worth reconnecting) rather than an unrecoverable error to surface to the caller. +func isDisconnectErr(err error) bool { + var errno syscall.Errno + if gerrors.As(err, &errno) { + switch errno { + case syscall.EPIPE, syscall.ECONNRESET, syscall.ENOTCONN: + return true + } + } + return false +} + +// replayBuffer keeps the most recent max bytes written to the connection so they +// can be re-sent after a reconnect; older bytes are dropped. +// +// To avoid recopying on every append, it lets the backing slice grow to 2*max and +// only then trims it back to the last max bytes. A trim (an O(max) copy) therefore +// happens at most once per max bytes appended, so on average an append costs time +// proportional to the bytes added, and the buffer never holds more than ~2*max. +type replayBuffer struct { + data []byte + max int +} + +func newReplayBuffer(max int) *replayBuffer { + return &replayBuffer{max: max} +} + +// append records p, keeping only the most recent max bytes. +func (r *replayBuffer) append(p []byte) { + r.data = append(r.data, p...) + // Trim only once the buffer exceeds 2*max, so the copy happens ~once per max bytes. + if len(r.data) > 2*r.max { + // Copy the last max bytes to the front, reusing the array (no realloc, no overlap). + r.data = append(r.data[:0], r.data[len(r.data)-r.max:]...) + } +} + +// bytes returns a copy of the retained tail (at most max bytes), in write order. +func (r *replayBuffer) bytes() []byte { + tail := r.data + if len(tail) > r.max { + tail = tail[len(tail)-r.max:] + } + out := make([]byte, len(tail)) + copy(out, tail) + return out +} diff --git a/internal/guest/transport/reconnect_test.go b/internal/guest/transport/reconnect_test.go new file mode 100644 index 0000000000..734fe84a93 --- /dev/null +++ b/internal/guest/transport/reconnect_test.go @@ -0,0 +1,251 @@ +//go:build linux +// +build linux + +package transport + +import ( + "bytes" + "errors" + "fmt" + "os" + "sync" + "sync/atomic" + "syscall" + "testing" +) + +// fakeConn is a scriptable Connection: writeFn drives per-call behavior, and +// anything written with the default behavior is captured for assertions. +type fakeConn struct { + writeFn func(p []byte) (int, error) + written bytes.Buffer + closed bool +} + +func (c *fakeConn) Write(p []byte) (int, error) { + if c.writeFn != nil { + return c.writeFn(p) + } + return c.written.Write(p) +} +func (c *fakeConn) Read([]byte) (int, error) { return 0, nil } +func (c *fakeConn) Close() error { c.closed = true; return nil } +func (c *fakeConn) CloseRead() error { return nil } +func (c *fakeConn) CloseWrite() error { return nil } +func (c *fakeConn) File() (*os.File, error) { return nil, nil } + +// fakeTransport hands back scripted connections/errors on each dial, in order. +type fakeTransport struct { + mu sync.Mutex + conns []Connection + errs []error + dials int +} + +func (t *fakeTransport) Dial(uint32) (Connection, error) { + t.mu.Lock() + defer t.mu.Unlock() + i := t.dials + t.dials++ + var conn Connection + var err error + if i < len(t.conns) { + conn = t.conns[i] + } + if i < len(t.errs) { + err = t.errs[i] + } + return conn, err +} + +// A healthy stream forwards straight through and never re-dials. +func TestReconnectWritePassthrough(t *testing.T) { + base := &fakeConn{} + tp := &fakeTransport{} + var stopped atomic.Bool + c := NewReconnectConnection(tp, 1, base, &stopped) + + n, err := c.Write([]byte("hello")) + if err != nil || n != 5 { + t.Fatalf("write = (%d, %v), want (5, nil)", n, err) + } + if tp.dials != 0 { + t.Fatalf("dials = %d, want 0", tp.dials) + } + if got := base.written.String(); got != "hello" { + t.Fatalf("delivered = %q, want %q", got, "hello") + } +} + +// A mid-write drop reconnects and re-sends the retained tail (the accepted-but- +// maybe-lost bytes) followed by the remainder, so nothing is lost. +func TestReconnectResumesAfterDrop(t *testing.T) { + dropped := &fakeConn{writeFn: func(p []byte) (int, error) { + return 2, syscall.EPIPE // "accepted" "he", then the host went away + }} + healthy := &fakeConn{} + tp := &fakeTransport{conns: []Connection{healthy}, errs: []error{nil}} + var stopped atomic.Bool + c := NewReconnectConnection(tp, 7, dropped, &stopped) + + n, err := c.Write([]byte("hello")) + if err != nil || n != 5 { + t.Fatalf("write = (%d, %v), want (5, nil)", n, err) + } + if !dropped.closed { + t.Fatalf("dropped connection was not closed before reconnect") + } + if got := healthy.written.String(); got != "hello" { + t.Fatalf("delivered after reconnect = %q, want %q (replayed \"he\" + \"llo\")", got, "hello") + } + if tp.dials != 1 { + t.Fatalf("dials = %d, want 1", tp.dials) + } +} + +// A failed dial is retried until one succeeds (exercises the retry loop once). +func TestReconnectRetriesUntilDial(t *testing.T) { + dropped := &fakeConn{writeFn: func(p []byte) (int, error) { + return 0, syscall.ECONNRESET + }} + healthy := &fakeConn{} + tp := &fakeTransport{ + conns: []Connection{nil, healthy}, + errs: []error{errors.New("connection refused"), nil}, + } + var stopped atomic.Bool + c := NewReconnectConnection(tp, 1, dropped, &stopped) + + n, err := c.Write([]byte("hi")) + if err != nil || n != 2 { + t.Fatalf("write = (%d, %v), want (2, nil)", n, err) + } + if got := healthy.written.String(); got != "hi" { + t.Fatalf("delivered = %q, want %q", got, "hi") + } + if tp.dials != 2 { + t.Fatalf("dials = %d, want 2", tp.dials) + } +} + +// A non-disconnect error surfaces to the caller and never triggers a re-dial. +func TestReconnectNonDisconnectErrorPassesThrough(t *testing.T) { + boom := errors.New("boom") + base := &fakeConn{writeFn: func(p []byte) (int, error) { return 0, boom }} + tp := &fakeTransport{} + var stopped atomic.Bool + c := NewReconnectConnection(tp, 1, base, &stopped) + + if _, err := c.Write([]byte("x")); !errors.Is(err, boom) { + t.Fatalf("err = %v, want %v", err, boom) + } + if tp.dials != 0 { + t.Fatalf("dials = %d, want 0", tp.dials) + } +} + +// Once teardown is signalled, a drop surfaces the error instead of looping. +func TestReconnectStopAbortsRedial(t *testing.T) { + base := &fakeConn{writeFn: func(p []byte) (int, error) { return 0, syscall.EPIPE }} + tp := &fakeTransport{} + var stopped atomic.Bool + stopped.Store(true) + c := NewReconnectConnection(tp, 1, base, &stopped) + + n, err := c.Write([]byte("x")) + if err == nil || n != 0 { + t.Fatalf("write = (%d, %v), want (0, non-nil)", n, err) + } + if tp.dials != 0 { + t.Fatalf("dials = %d, want 0 (must not dial once stopped)", tp.dials) + } +} + +// Half-close/close operations act on the underlying connection. +func TestReconnectCloseDelegates(t *testing.T) { + base := &fakeConn{} + tp := &fakeTransport{} + var stopped atomic.Bool + c := NewReconnectConnection(tp, 1, base, &stopped) + + if err := c.Close(); err != nil { + t.Fatalf("close = %v, want nil", err) + } + if !base.closed { + t.Fatalf("close was not delegated to the underlying connection") + } +} + +// Only genuine peer-gone errors are treated as reconnectable. +func TestIsDisconnectErr(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {syscall.EPIPE, true}, + {syscall.ECONNRESET, true}, + {syscall.ENOTCONN, true}, + {fmt.Errorf("wrapped: %w", syscall.EPIPE), true}, + {syscall.ECONNABORTED, false}, + {syscall.ESHUTDOWN, false}, + {errors.New("plain"), false}, + {nil, false}, + } + for _, tc := range cases { + if got := isDisconnectErr(tc.err); got != tc.want { + t.Errorf("isDisconnectErr(%v) = %v, want %v", tc.err, got, tc.want) + } + } +} + +// Bytes a connection accepted before it dropped are replayed on the new +// connection, so nothing is lost even though the guest advanced past them. +func TestReconnectReplaysLostBytes(t *testing.T) { + var down atomic.Bool + conn1 := &fakeConn{} + conn1.writeFn = func(p []byte) (int, error) { + if down.Load() { + return 0, syscall.EPIPE + } + return conn1.written.Write(p) // accepts the bytes (host may never read them) + } + conn2 := &fakeConn{} + tp := &fakeTransport{conns: []Connection{conn2}, errs: []error{nil}} + var stopped atomic.Bool + c := NewReconnectConnection(tp, 1, conn1, &stopped) + + if n, err := c.Write([]byte("AAA")); err != nil || n != 3 { + t.Fatalf("first write = (%d, %v), want (3, nil)", n, err) + } + down.Store(true) // conn1's buffered "AAA" is lost and the next write drops + if n, err := c.Write([]byte("BBB")); err != nil || n != 3 { + t.Fatalf("second write = (%d, %v), want (3, nil)", n, err) + } + if got := conn2.written.String(); got != "AAABBB" { + t.Fatalf("delivered = %q, want %q (lost \"AAA\" must be replayed)", got, "AAABBB") + } +} + +// replayBuffer keeps only the most recent size bytes, in write order. +func TestReplayBuffer(t *testing.T) { + r := newReplayBuffer(4) + if got := string(r.bytes()); got != "" { + t.Fatalf("empty = %q, want \"\"", got) + } + r.append([]byte("ab")) + if got := string(r.bytes()); got != "ab" { + t.Fatalf("after ab = %q, want ab", got) + } + r.append([]byte("cd")) // exactly full + if got := string(r.bytes()); got != "abcd" { + t.Fatalf("after abcd = %q, want abcd", got) + } + r.append([]byte("ef")) // wraps, drops "ab" + if got := string(r.bytes()); got != "cdef" { + t.Fatalf("after wrap = %q, want cdef", got) + } + r.append([]byte("XYZ123")) // larger than size: keep last 4 + if got := string(r.bytes()); got != "Z123" { + t.Fatalf("after oversized = %q, want Z123", got) + } +}