From 6d375c8428cfd19d0141525410eca3b3339b6821 Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Tue, 1 Sep 2026 11:39:11 +0300 Subject: [PATCH 1/8] feat(attachmux): add stdio multiplexing frame protocol Signed-off-by: Eugene Kalinin --- pkg/attachmux/proto.go | 113 ++++++++++++++++++++++++++ pkg/attachmux/proto_test.go | 158 ++++++++++++++++++++++++++++++++++++ 2 files changed, 271 insertions(+) create mode 100644 pkg/attachmux/proto.go create mode 100644 pkg/attachmux/proto_test.go diff --git a/pkg/attachmux/proto.go b/pkg/attachmux/proto.go new file mode 100644 index 00000000000..35ef1db9a33 --- /dev/null +++ b/pkg/attachmux/proto.go @@ -0,0 +1,113 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +// Package attachmux multiplexes a container's stdio between the process that +// owns it and any number of attached CLI sessions. +// +// A container's stdio FIFOs cannot be shared: a FIFO has a single queue, so two +// readers on the stdout FIFO each receive a random subset of the container's +// output. Instead, one process owns the FIFOs and every session talks to it +// over a socket, framed with the protocol in this file. +package attachmux + +import ( + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" +) + +// Stream identifiers carried in a frame header. +const ( + StreamStdin byte = 0 + StreamStdout byte = 1 + StreamStderr byte = 2 + StreamControl byte = 3 +) + +// Types of a StreamControl message. +const ( + // ControlHello is sent by the broker to a session as soon as it connects. + ControlHello = "hello" + // ControlExit is sent by the broker when the container has exited. It + // carries no exit code: a session reads that from containerd, so that there + // is a single source of truth for it. + ControlExit = "exit" +) + +// ProtocolVersion is announced in the hello message so that a future change to +// the framing can be detected by an older client. +const ProtocolVersion = 1 + +const ( + headerSize = 8 + // maxPayload bounds a single frame so that a corrupt or hostile peer cannot + // make the reader allocate without limit. Container output is read in 32 KiB + // chunks, so this is never hit in practice. + maxPayload = 1 << 20 +) + +// ErrPayloadTooLarge is returned for a frame whose payload exceeds maxPayload. +var ErrPayloadTooLarge = errors.New("attachmux: frame payload too large") + +// Control is the JSON body of a StreamControl frame. +type Control struct { + Type string `json:"type"` + Version int `json:"version,omitempty"` + TTY bool `json:"tty,omitempty"` +} + +// EncodeFrame returns the wire representation of a single frame: one stream +// byte, three reserved zero bytes, a big-endian uint32 payload length, then the +// payload. +func EncodeFrame(stream byte, payload []byte) ([]byte, error) { + if len(payload) > maxPayload { + return nil, fmt.Errorf("%w: %d bytes", ErrPayloadTooLarge, len(payload)) + } + buf := make([]byte, headerSize+len(payload)) + buf[0] = stream + binary.BigEndian.PutUint32(buf[4:headerSize], uint32(len(payload))) + copy(buf[headerSize:], payload) + return buf, nil +} + +// EncodeControl returns a StreamControl frame carrying c. +func EncodeControl(c Control) ([]byte, error) { + b, err := json.Marshal(c) + if err != nil { + return nil, err + } + return EncodeFrame(StreamControl, b) +} + +// ReadFrame reads exactly one frame. The returned payload is a fresh slice +// owned by the caller. +func ReadFrame(r io.Reader) (stream byte, payload []byte, err error) { + var hdr [headerSize]byte + if _, err = io.ReadFull(r, hdr[:]); err != nil { + return 0, nil, err + } + n := binary.BigEndian.Uint32(hdr[4:headerSize]) + if n > maxPayload { + return 0, nil, fmt.Errorf("%w: %d bytes", ErrPayloadTooLarge, n) + } + payload = make([]byte, n) + if _, err = io.ReadFull(r, payload); err != nil { + return 0, nil, err + } + return hdr[0], payload, nil +} diff --git a/pkg/attachmux/proto_test.go b/pkg/attachmux/proto_test.go new file mode 100644 index 00000000000..e8f35135d55 --- /dev/null +++ b/pkg/attachmux/proto_test.go @@ -0,0 +1,158 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "bytes" + "encoding/json" + "io" + "strings" + "testing" + + "gotest.tools/v3/assert" +) + +func TestEncodeFrameRoundTrip(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + stream byte + payload []byte + }{ + {"stdout", StreamStdout, []byte("hello")}, + {"stderr", StreamStderr, []byte("oops")}, + {"stdin", StreamStdin, []byte{0x16, 0x11}}, + {"empty", StreamStdout, []byte{}}, + {"binary", StreamStdout, []byte{0x00, 0xff, 0x1b, 0x5b, 0x41}}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + frame, err := EncodeFrame(tc.stream, tc.payload) + assert.NilError(t, err) + + stream, payload, err := ReadFrame(bytes.NewReader(frame)) + assert.NilError(t, err) + assert.Equal(t, stream, tc.stream) + assert.DeepEqual(t, payload, tc.payload) + }) + } +} + +func TestReadFrameConsecutive(t *testing.T) { + t.Parallel() + + var buf bytes.Buffer + for _, s := range []string{"one", "two", "three"} { + frame, err := EncodeFrame(StreamStdout, []byte(s)) + assert.NilError(t, err) + buf.Write(frame) + } + + for _, want := range []string{"one", "two", "three"} { + stream, payload, err := ReadFrame(&buf) + assert.NilError(t, err) + assert.Equal(t, stream, StreamStdout) + assert.Equal(t, string(payload), want) + } + + _, _, err := ReadFrame(&buf) + assert.ErrorIs(t, err, io.EOF) +} + +func TestEncodeFrameRejectsOversizedPayload(t *testing.T) { + t.Parallel() + + _, err := EncodeFrame(StreamStdout, make([]byte, maxPayload+1)) + assert.ErrorIs(t, err, ErrPayloadTooLarge) +} + +func TestReadFrameRejectsOversizedHeader(t *testing.T) { + t.Parallel() + + // A header announcing more than maxPayload must be refused before the + // reader allocates for it. + hdr := []byte{StreamStdout, 0, 0, 0, 0xff, 0xff, 0xff, 0xff} + _, _, err := ReadFrame(bytes.NewReader(hdr)) + assert.ErrorIs(t, err, ErrPayloadTooLarge) +} + +func TestReadFrameTruncated(t *testing.T) { + t.Parallel() + + frame, err := EncodeFrame(StreamStdout, []byte("hello")) + assert.NilError(t, err) + + _, _, err = ReadFrame(bytes.NewReader(frame[:len(frame)-2])) + assert.ErrorIs(t, err, io.ErrUnexpectedEOF) +} + +func TestEncodeControl(t *testing.T) { + t.Parallel() + + frame, err := EncodeControl(Control{Type: ControlHello, Version: ProtocolVersion, TTY: true}) + assert.NilError(t, err) + + stream, payload, err := ReadFrame(bytes.NewReader(frame)) + assert.NilError(t, err) + assert.Equal(t, stream, StreamControl) + + var c Control + assert.NilError(t, json.Unmarshal(payload, &c)) + assert.Equal(t, c.Type, ControlHello) + assert.Equal(t, c.Version, 1) + assert.Equal(t, c.TTY, true) +} + +func TestEncodeControlExitCarriesNoCode(t *testing.T) { + t.Parallel() + + // The exit message only says the container is gone. The exit code always + // comes from containerd, so that there is one source of truth for it. + frame, err := EncodeControl(Control{Type: ControlExit}) + assert.NilError(t, err) + + _, payload, err := ReadFrame(bytes.NewReader(frame)) + assert.NilError(t, err) + assert.Equal(t, string(payload), `{"type":"exit"}`) +} + +func TestReadFrameFromStream(t *testing.T) { + t.Parallel() + + // A reader that hands out one byte at a time still yields a whole frame. + frame, err := EncodeFrame(StreamStdout, []byte("drip")) + assert.NilError(t, err) + + stream, payload, err := ReadFrame(iotest(strings.NewReader(string(frame)))) + assert.NilError(t, err) + assert.Equal(t, stream, StreamStdout) + assert.Equal(t, string(payload), "drip") +} + +// iotest wraps r so that each Read returns at most one byte. +func iotest(r io.Reader) io.Reader { return &oneByteReader{r: r} } + +type oneByteReader struct{ r io.Reader } + +func (o *oneByteReader) Read(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + return o.r.Read(p[:1]) +} From 5c06be901fdf389125e0197715ab25e9176026c4 Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Tue, 1 Sep 2026 11:47:37 +0300 Subject: [PATCH 2/8] feat(attachmux): add the container stdio broker Signed-off-by: Eugene Kalinin --- pkg/attachmux/broker.go | 418 ++++++++++++++++++++++++++ pkg/attachmux/broker_test.go | 553 +++++++++++++++++++++++++++++++++++ 2 files changed, 971 insertions(+) create mode 100644 pkg/attachmux/broker.go create mode 100644 pkg/attachmux/broker_test.go diff --git a/pkg/attachmux/broker.go b/pkg/attachmux/broker.go new file mode 100644 index 00000000000..fc40e1216c2 --- /dev/null +++ b/pkg/attachmux/broker.go @@ -0,0 +1,418 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "context" + "errors" + "io" + "net" + "sync" + "time" + + "github.com/containerd/log" +) + +const ( + // defaultQueueDepth is how many frames are buffered for a single session. A + // session that falls this far behind is disconnected: the container's output + // must never wait on a consumer. + // https://github.com/containerd/nerdctl/issues/5137 + defaultQueueDepth = 256 + // writeTimeout bounds a single write to a session, so that a session that is + // connected but no longer draining its socket is eventually dropped instead + // of leaking a goroutine and its queue for the container's lifetime. + writeTimeout = 30 * time.Second + // closeDrainTimeout bounds how long Close waits for the writers to flush + // what is already queued. containerd calls os.Exit as soon as the logging + // function returns (core/runtime/v2/logging/logging_unix.go), so a writer + // still draining at that point is killed mid-frame. + closeDrainTimeout = 2 * time.Second + // DrainTimeout bounds how long a client waits, after the container has + // exited, for the broker to deliver what it still has queued. + // + // It has to be longer than closeDrainTimeout. A client that gave up first + // would cancel its stream and close the socket while the broker was still + // flushing, dropping the tail of the container's output and reporting + // success for it. + DrainTimeout = 3 * closeDrainTimeout +) + +// Broker owns a container's stdio. It fans container output out to every +// connected session and merges the sessions' input into the container's stdin. +// +// Every method is safe for concurrent use, and none of them block on a session. +type Broker struct { + tty bool + + mu sync.Mutex + sessions map[*session]struct{} + closed bool + // stdin is the write end of the container's stdin, installed by NewBroker + // or later by SetStdin, and taken back by Close. It is guarded by mu so + // that closing and installing cannot interleave. + stdin io.WriteCloser + + // writers tracks the per-session writer goroutines so that Close can flush + // what is queued before the process is torn down. + writers sync.WaitGroup + + // stdinWriteMu keeps a session's write to the container's stdin from being + // interleaved with another session's. It guards the write, not the field: + // a write to a FIFO the container is not reading blocks until the pipe + // drains, and Close must be able to release the descriptor meanwhile + // rather than queue behind it. stdin itself is guarded by mu. + stdinWriteMu sync.Mutex +} + +type session struct { + conn net.Conn + frames chan []byte + + // mu guards closed together with the send on frames. Without it, Close + // could send on a channel that a concurrent dropSession has just closed, + // which panics even from inside a select. + mu sync.Mutex + closed bool + // exit is the final frame writeLoop delivers after frames has drained. See + // closeWith. + exit []byte +} + +// send queues frame for the session. It reports false only when the session's +// queue is full, meaning the session has fallen behind and has to be dropped. A +// session that is already closed reports true: it is gone, not slow. +func (s *session) send(frame []byte) bool { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return true + } + select { + case s.frames <- frame: + return true + default: + return false + } +} + +// closeWith closes the session's queue, leaving exit for writeLoop to deliver +// once everything already queued has gone out. exit is nil when the session is +// being dropped rather than told the container is gone. +// +// The exit frame does not go through the queue. A session whose queue happens +// to be exactly full at that moment has received all of the container's output +// and is not behind; pushing the exit through the same bounded channel would +// drop it and leave the client reporting a lost connection for a container that +// finished normally. +func (s *session) closeWith(exit []byte) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + s.closed = true + s.exit = exit + close(s.frames) +} + +// close stops the session's writer. The writer closes the connection once it +// has drained whatever is already queued. +func (s *session) close() { s.closeWith(nil) } + +// NewBroker returns a Broker fanning container output out to its sessions. +// +// stdin is the write end of the container's stdin and may be nil, either +// because the container has none or because it is not known yet; SetStdin +// installs it later. The Broker holds it open for the container's lifetime, +// which is what makes detaching leave the container running, and closes it only +// in Close. +// +// Note that closing this end would not deliver EOF to the container anyway. +// The shim opens the same FIFO O_WRONLY and holds that descriptor until +// task.CloseIO (cmd/containerd-shim-runc-v2/process/init.go, openStdin), so +// closing the container's stdin is a task operation, not a broker one. +func NewBroker(tty bool, stdin io.WriteCloser) *Broker { + return &Broker{ + tty: tty, + stdin: stdin, + sessions: map[*session]struct{}{}, + } +} + +// Write fans p out to every connected session. It never blocks: a session whose +// queue is full is disconnected instead. p is copied, so the caller is free to +// reuse it immediately. +func (b *Broker) Write(stream byte, p []byte) { + if len(p) == 0 { + return + } + + b.mu.Lock() + if b.closed || len(b.sessions) == 0 { + b.mu.Unlock() + return + } + frame, err := EncodeFrame(stream, p) + if err != nil { + b.mu.Unlock() + log.L.WithError(err).Warn("attachmux: dropping an oversized output frame") + return + } + var slow []*session + for s := range b.sessions { + if !s.send(frame) { + slow = append(slow, s) + } + } + for _, s := range slow { + delete(b.sessions, s) + } + b.mu.Unlock() + + for _, s := range slow { + log.L.Warn("attachmux: an attach session stopped keeping up, disconnecting it") + s.close() + } +} + +// SessionCount returns how many sessions are currently connected. +func (b *Broker) SessionCount() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.sessions) +} + +// Serve accepts sessions on l until ctx is done or l is closed. +func (b *Broker) Serve(ctx context.Context, l net.Listener) error { + go func() { + <-ctx.Done() + l.Close() + }() + + for { + conn, err := l.Accept() + if err != nil { + if ctx.Err() != nil || errors.Is(err, net.ErrClosed) { + return nil + } + return err + } + b.addSession(conn) + } +} + +// Close disconnects every session and releases the container's stdin. +// +// exited says whether the container is known to have exited. Only then are the +// sessions told so, and they treat it as proof: `nerdctl attach` stops +// streaming and reads the exit code from containerd. The owner cannot infer it +// from its own stdio reaching EOF, because that also happens when the logging +// process is asked to shut down while the container keeps running, and a client +// told the container had gone would then wait for an exit that never comes. +// +// A false value tears the sessions down without that claim, and a client +// reports the session as broken, which is what it is. +// +// The exit carries no code: a session reads that from containerd. +func (b *Broker) Close(exited bool) { + b.mu.Lock() + if b.closed { + b.mu.Unlock() + return + } + b.closed = true + sessions := make([]*session, 0, len(b.sessions)) + for s := range b.sessions { + sessions = append(sessions, s) + } + b.sessions = map[*session]struct{}{} + // Take the descriptor out under the same lock that SetStdin uses, so that + // a concurrent SetStdin either loses the race and closes its own file, or + // wins it and is closed here. + stdin := b.stdin + b.stdin = nil + b.mu.Unlock() + + // The exit is handed over before stdin is released, and that order matters. + // A readLoop blocked writing to a FIFO the container stopped reading wakes + // up with an error the moment the descriptor closes, and its deferred + // dropSession closes the session's queue; doing this afterwards would leave + // the client reporting an unexpected disconnect for a container that exited + // normally. + var exit []byte + if exited { + frame, err := EncodeControl(Control{Type: ControlExit}) + if err != nil { + log.L.WithError(err).Warn("attachmux: failed to encode the exit frame") + } else { + exit = frame + } + } + for _, s := range sessions { + s.closeWith(exit) + } + + // Wait for the writers to flush. Closing the queues is not enough: the + // caller is the logging process, and containerd calls os.Exit as soon as + // the logging function returns, which would kill a writer mid-frame and + // lose the container's last output along with the exit message. + drained := make(chan struct{}) + go func() { + b.writers.Wait() + close(drained) + }() + select { + case <-drained: + case <-time.After(closeDrainTimeout): + log.L.Warn("attachmux: timed out flushing attach sessions on close") + } + + // Released last, outside the lock and outside stdinWriteMu: a readLoop can + // be blocked in Write on a FIFO the container stopped reading, and waiting + // for it would hang the logging process on SIGTERM. The FIFO is registered + // with the runtime poller, so closing it unblocks that write with ErrClosed. + if stdin != nil { + stdin.Close() + } +} + +// SetStdin gives the broker the write end of the container's stdin once it is +// known to be usable, and reports whether the broker took it. It returns false +// for a broker that is already closed, and the caller then closes w itself. +// +// Stdin arrives late because whether the container has one at all can only be +// established by opening the FIFO, which the owner has to do off the path that +// reads the container's output. See pkg/logging/broker_unix.go. +func (b *Broker) SetStdin(w io.WriteCloser) bool { + b.mu.Lock() + defer b.mu.Unlock() + if b.closed || b.stdin != nil { + return false + } + b.stdin = w + return true +} + +// addSession registers conn, greets it and starts its two pumps. +func (b *Broker) addSession(conn net.Conn) { + hello, err := EncodeControl(Control{Type: ControlHello, Version: ProtocolVersion, TTY: b.tty}) + if err != nil { + conn.Close() + return + } + + s := &session{conn: conn, frames: make(chan []byte, defaultQueueDepth)} + + b.mu.Lock() + if b.closed { + b.mu.Unlock() + conn.Close() + return + } + b.sessions[s] = struct{}{} + // The hello has to be queued before anything else so that it is the first + // frame the session sees. + s.send(hello) + // Add under the lock, before the session becomes reachable to Close. + // Adding after unlocking would let Close observe the session, close it and + // return from Wait with the counter still at zero, which both breaks the + // sync.WaitGroup contract and lets the logging process exit before this + // writer has flushed anything. + b.writers.Add(1) + b.mu.Unlock() + + go b.writeLoop(s) + go b.readLoop(s) +} + +// writeLoop drains the session's queue onto its connection. +func (b *Broker) writeLoop(s *session) { + defer b.writers.Done() + defer s.conn.Close() + + for frame := range s.frames { + if err := s.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + b.dropSession(s) + return + } + if _, err := s.conn.Write(frame); err != nil { + b.dropSession(s) + return + } + } + + // Everything queued has gone out. If the container exited, say so now: this + // is the last thing the session ever sees, and it is what tells the client + // apart from a broker that vanished. + s.mu.Lock() + exit := s.exit + s.mu.Unlock() + if exit == nil { + return + } + if err := s.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + return + } + if _, err := s.conn.Write(exit); err != nil { + log.L.WithError(err).Debug("attachmux: failed to deliver the exit frame") + } +} + +// readLoop forwards the session's input to the container. +func (b *Broker) readLoop(s *session) { + defer b.dropSession(s) + + for { + stream, payload, err := ReadFrame(s.conn) + if err != nil { + return + } + switch stream { + case StreamStdin: + // stdin can be installed by SetStdin after this loop has started, + // and taken away by Close while it is running, so it is read under + // mu. The write itself is held only by stdinWriteMu: taking mu + // across a write that can block on a full FIFO would hang Close. + b.mu.Lock() + w := b.stdin + b.mu.Unlock() + if w == nil { + continue + } + + b.stdinWriteMu.Lock() + _, err = w.Write(payload) + b.stdinWriteMu.Unlock() + if err != nil { + // Close closed the descriptor under a blocked write, or the + // container is gone. Either way this session's input has + // nowhere to go. + log.L.WithError(err).Debug("attachmux: failed to write to the container stdin") + return + } + } + } +} + +// dropSession unregisters s and stops its writer. +func (b *Broker) dropSession(s *session) { + b.mu.Lock() + delete(b.sessions, s) + b.mu.Unlock() + s.close() +} diff --git a/pkg/attachmux/broker_test.go b/pkg/attachmux/broker_test.go new file mode 100644 index 00000000000..e8ad661a715 --- /dev/null +++ b/pkg/attachmux/broker_test.go @@ -0,0 +1,553 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "bytes" + "context" + "encoding/json" + "net" + "os" + "sync" + "testing" + "time" + + "gotest.tools/v3/assert" +) + +// syncBuffer is an io.WriteCloser that records what the container's stdin +// received. It is safe for concurrent use because several sessions may write +// at once. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer + closed bool +} + +func (s *syncBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.Write(p) +} + +func (s *syncBuffer) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + s.closed = true + return nil +} + +func (s *syncBuffer) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.String() +} + +func (s *syncBuffer) isClosed() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.closed +} + +// connect wires a new session into b over an in-memory connection and consumes +// its hello message, returning the session's end of the pipe. +func connect(t *testing.T, b *Broker) net.Conn { + t.Helper() + + mine, theirs := net.Pipe() + t.Cleanup(func() { mine.Close() }) + + go b.addSession(theirs) + + assert.NilError(t, mine.SetReadDeadline(time.Now().Add(5*time.Second))) + stream, payload, err := ReadFrame(mine) + assert.NilError(t, err) + assert.Equal(t, stream, StreamControl) + + var c Control + assert.NilError(t, json.Unmarshal(payload, &c)) + assert.Equal(t, c.Type, ControlHello) + assert.Equal(t, c.Version, ProtocolVersion) + + // Leave a deadline armed for the rest of the test rather than clearing it. + // net.Pipe refuses SetReadDeadline once the peer has closed, and the broker + // closes its end as soon as a session is dropped, so a test cannot reliably + // arm one later. This is only a guard against hanging. + assert.NilError(t, mine.SetReadDeadline(time.Now().Add(30*time.Second))) + return mine +} + +// waitSessions blocks until the broker reports n sessions, or fails the test. +func waitSessions(t *testing.T, b *Broker, n int) { + t.Helper() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if b.SessionCount() == n { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("expected %d sessions, got %d", n, b.SessionCount()) +} + +func TestBrokerFansOutToEverySession(t *testing.T) { + t.Parallel() + + b := NewBroker(true, nil) + first := connect(t, b) + second := connect(t, b) + waitSessions(t, b, 2) + + b.Write(StreamStdout, []byte("shared output")) + + for _, conn := range []net.Conn{first, second} { + stream, payload, err := ReadFrame(conn) + assert.NilError(t, err) + assert.Equal(t, stream, StreamStdout) + assert.Equal(t, string(payload), "shared output") + } +} + +func TestBrokerMergesStdinFromEverySession(t *testing.T) { + t.Parallel() + + stdin := &syncBuffer{} + b := NewBroker(false, stdin) + first := connect(t, b) + second := connect(t, b) + waitSessions(t, b, 2) + + frame, err := EncodeFrame(StreamStdin, []byte("from-first\n")) + assert.NilError(t, err) + _, err = first.Write(frame) + assert.NilError(t, err) + + frame, err = EncodeFrame(StreamStdin, []byte("from-second\n")) + assert.NilError(t, err) + _, err = second.Write(frame) + assert.NilError(t, err) + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + got := stdin.String() + if len(got) == len("from-first\nfrom-second\n") { + break + } + time.Sleep(5 * time.Millisecond) + } + + got := stdin.String() + assert.Assert(t, bytes.Contains([]byte(got), []byte("from-first\n")), "got %q", got) + assert.Assert(t, bytes.Contains([]byte(got), []byte("from-second\n")), "got %q", got) +} + +func TestBrokerDisconnectsASessionThatStopsReading(t *testing.T) { + t.Parallel() + + // The container must never be held up by a session that stopped consuming. + b := NewBroker(true, nil) + stalled := connect(t, b) + waitSessions(t, b, 1) + + // Never read from `stalled` again. Write far more than the session queue + // can hold; every call must return promptly. + done := make(chan struct{}) + go func() { + defer close(done) + for range defaultQueueDepth * 4 { + b.Write(StreamStdout, []byte("x")) + } + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("Broker.Write blocked on a session that stopped reading") + } + + waitSessions(t, b, 0) + assert.Assert(t, stalled != nil) +} + +func TestBrokerKeepsRunningWithNoSessions(t *testing.T) { + t.Parallel() + + b := NewBroker(true, nil) + + done := make(chan struct{}) + go func() { + defer close(done) + for range 1000 { + b.Write(StreamStdout, []byte("output with nobody attached")) + } + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("Broker.Write blocked with no sessions attached") + } +} + +func TestBrokerDoesNotCloseStdinWhenASessionLeaves(t *testing.T) { + t.Parallel() + + // Detaching must not send EOF to the container: this is what makes + // ctrl-p ctrl-q leave a container running. + stdin := &syncBuffer{} + b := NewBroker(false, stdin) + conn := connect(t, b) + waitSessions(t, b, 1) + + assert.NilError(t, conn.Close()) + waitSessions(t, b, 0) + + assert.Equal(t, stdin.isClosed(), false) +} + +func TestBrokerAcceptsStdinAfterASessionConnects(t *testing.T) { + t.Parallel() + + // The owner opens the container's stdin FIFO off the path that reads its + // output, so stdin can arrive after sessions are already streaming. A + // session that connected first still has to reach the container once it + // does. + // + // Input sent before stdin is installed is dropped rather than buffered, + // which readLoop documents, but a test cannot pin that down: net.Pipe + // reports a write as complete once the peer has read the bytes, not once + // readLoop has acted on them, so SetStdin can always land in between. + b := NewBroker(false, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + stdin := &syncBuffer{} + assert.Equal(t, b.SetStdin(stdin), true) + + frame, err := EncodeFrame(StreamStdin, []byte("late")) + assert.NilError(t, err) + _, err = conn.Write(frame) + assert.NilError(t, err) + + for range 100 { + if stdin.String() == "late" { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("stdin did not reach the broker, got %q", stdin.String()) +} + +func TestBrokerCloseReleasesStdin(t *testing.T) { + t.Parallel() + + // The container has exited, so nothing can read its stdin any more. Leaving + // the descriptor open would keep the FIFO alive for as long as the logging + // process lives. + stdin := &syncBuffer{} + b := NewBroker(false, stdin) + b.Close(true) + assert.Equal(t, stdin.isClosed(), true) + + assert.Equal(t, b.SetStdin(&syncBuffer{}), false) +} + +// blockingWriter models the container's stdin FIFO when the container has +// stopped reading it: the write blocks until the descriptor is closed. +type blockingWriter struct { + released chan struct{} + once sync.Once +} + +func newBlockingWriter() *blockingWriter { + return &blockingWriter{released: make(chan struct{})} +} + +func (w *blockingWriter) Write(p []byte) (int, error) { + <-w.released + return 0, os.ErrClosed +} + +func (w *blockingWriter) Close() error { + w.once.Do(func() { close(w.released) }) + return nil +} + +func TestBrokerCloseDoesNotWaitForABlockedStdinWrite(t *testing.T) { + t.Parallel() + + // A container that stopped reading its stdin leaves a session's write + // blocked on a full FIFO. Close has to release the descriptor rather than + // queue behind that write: it runs on SIGTERM, and the logging process is + // killed shortly after it returns. + stdin := newBlockingWriter() + b := NewBroker(false, stdin) + conn := connect(t, b) + waitSessions(t, b, 1) + + frame, err := EncodeFrame(StreamStdin, []byte("blocked")) + assert.NilError(t, err) + _, err = conn.Write(frame) + assert.NilError(t, err) + + // Give readLoop time to reach the write before closing. + time.Sleep(100 * time.Millisecond) + + done := make(chan struct{}) + go func() { + b.Close(true) + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Close blocked behind a stdin write") + } + + // And the session still learns that the container exited. Releasing stdin + // wakes the blocked readLoop, whose dropSession closes this session's queue, + // so the exit has to be queued before that happens. + for { + stream, payload, err := ReadFrame(conn) + assert.NilError(t, err) + if stream != StreamControl { + continue + } + var c Control + assert.NilError(t, json.Unmarshal(payload, &c)) + if c.Type == ControlHello { + continue + } + assert.Equal(t, c.Type, ControlExit) + return + } +} + +func TestBrokerSetStdinRacesWithClose(t *testing.T) { + t.Parallel() + + // startBroker opens the container's stdin in the background, so SetStdin + // can land at any moment, including after the container has exited. The + // loser of that race has to be told, so that it closes its own descriptor + // instead of leaking it. + for range 200 { + b := NewBroker(false, nil) + stdin := &syncBuffer{} + + var wg sync.WaitGroup + var took bool + wg.Go(func() { took = b.SetStdin(stdin) }) + wg.Go(func() { b.Close(true) }) + wg.Wait() + + // Whoever ended up owning it closed it: SetStdin lost and the caller + // closes, or SetStdin won and Close took it. + if !took { + stdin.Close() + } + assert.Equal(t, stdin.isClosed(), true) + } +} + +func TestBrokerCloseWithoutAnExitDoesNotAnnounceOne(t *testing.T) { + t.Parallel() + + // The owner's stdio ending is not proof that the container exited: the same + // thing happens when the logging process is shut down while the container + // keeps running. A session told otherwise would wait for an exit code that + // never arrives. + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + b.Close(false) + + for { + stream, payload, err := ReadFrame(conn) + if err != nil { + // EOF with no exit frame, which is what a broken session looks like. + return + } + if stream != StreamControl { + continue + } + var c Control + assert.NilError(t, json.Unmarshal(payload, &c)) + assert.Assert(t, c.Type != ControlExit, "Close(false) announced an exit") + } +} + +func TestBrokerCloseDeliversTheExitPastAFullQueue(t *testing.T) { + t.Parallel() + + // A session whose queue is exactly full has received all of the container's + // output and is not behind. Pushing the exit through that same queue would + // drop it, and the client would report a lost connection for a container + // that finished normally. + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + // Fill the queue without draining the connection. + for range defaultQueueDepth { + b.Write(StreamStdout, []byte("x")) + } + b.Close(true) + + for { + stream, payload, err := ReadFrame(conn) + assert.NilError(t, err) + if stream != StreamControl { + continue + } + var c Control + assert.NilError(t, json.Unmarshal(payload, &c)) + if c.Type == ControlHello { + continue + } + assert.Equal(t, c.Type, ControlExit) + return + } +} + +func TestBrokerCloseAnnouncesExit(t *testing.T) { + t.Parallel() + + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + b.Close(true) + + stream, payload, err := ReadFrame(conn) + assert.NilError(t, err) + assert.Equal(t, stream, StreamControl) + + var c Control + assert.NilError(t, json.Unmarshal(payload, &c)) + assert.Equal(t, c.Type, ControlExit) + + // The broker closes the connection right after announcing the exit. + _, _, err = ReadFrame(conn) + assert.Assert(t, err != nil) +} + +func TestBrokerCloseFlushesQueuedFrames(t *testing.T) { + t.Parallel() + + // containerd calls os.Exit as soon as the logging function returns, so + // whatever Close leaves queued is lost. The container's last chunk of + // output has to reach the session before Close returns. + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + type frame struct { + stream byte + payload []byte + } + frames := make(chan frame, 8) + go func() { + for { + stream, payload, err := ReadFrame(conn) + if err != nil { + close(frames) + return + } + frames <- frame{stream, payload} + } + }() + + b.Write(StreamStdout, []byte("last words")) + b.Close(true) + + var sawOutput, sawExit bool + for range 2 { + select { + case f, ok := <-frames: + if !ok { + t.Fatal("the connection closed before the queued frames arrived") + } + if f.stream == StreamStdout && string(f.payload) == "last words" { + sawOutput = true + } + if f.stream == StreamControl { + sawExit = true + } + case <-time.After(5 * time.Second): + t.Fatal("timed out reading the frames Close should have flushed") + } + } + assert.Assert(t, sawOutput, "the container's last output was dropped by Close") + assert.Assert(t, sawExit, "the exit message was dropped by Close") +} + +func TestBrokerCloseRacesWithDisconnectingSessions(t *testing.T) { + t.Parallel() + + // A container exiting at the same moment as a session detaches must not + // send on a channel that dropSession has just closed. That would panic, and + // since the broker lives inside the logging process it would take logging + // down with it. Run with -race. + for range 50 { + b := NewBroker(true, nil) + conns := make([]net.Conn, 0, 4) + for range 4 { + conns = append(conns, connect(t, b)) + } + waitSessions(t, b, 4) + + var wg sync.WaitGroup + for _, conn := range conns { + wg.Go(func() { conn.Close() }) + } + wg.Go(func() { b.Close(true) }) + wg.Wait() + } +} + +func TestBrokerServeAcceptsSessions(t *testing.T) { + t.Parallel() + + l, err := net.Listen("tcp", "127.0.0.1:0") + assert.NilError(t, err) + t.Cleanup(func() { l.Close() }) + + b := NewBroker(true, nil) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + served := make(chan error, 1) + go func() { served <- b.Serve(ctx, l) }() + + conn, err := net.Dial("tcp", l.Addr().String()) + assert.NilError(t, err) + t.Cleanup(func() { conn.Close() }) + + stream, _, err := ReadFrame(conn) + assert.NilError(t, err) + assert.Equal(t, stream, StreamControl) + + cancel() + select { + case err := <-served: + assert.NilError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Serve did not return after the context was cancelled") + } +} From 480e171aa5c067bae24b205cd797aef76a8ac057 Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Tue, 1 Sep 2026 11:49:20 +0300 Subject: [PATCH 3/8] feat(attachmux): add the client attach session Signed-off-by: Eugene Kalinin --- pkg/attachmux/client.go | 180 ++++++++++++++++++++++++++ pkg/attachmux/client_test.go | 240 +++++++++++++++++++++++++++++++++++ 2 files changed, 420 insertions(+) create mode 100644 pkg/attachmux/client.go create mode 100644 pkg/attachmux/client_test.go diff --git a/pkg/attachmux/client.go b/pkg/attachmux/client.go new file mode 100644 index 00000000000..056b1c8ca60 --- /dev/null +++ b/pkg/attachmux/client.go @@ -0,0 +1,180 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "sync" +) + +// Session is a client's end of an attach connection. +type Session struct { + conn net.Conn + hello Control + + // writeMu serialises writes to the connection. + writeMu sync.Mutex + + exitMu sync.Mutex + exited bool +} + +// NewSession completes the handshake on conn and returns the session. The +// caller keeps ownership of conn: closing the Session closes it. +func NewSession(conn net.Conn) (*Session, error) { + stream, payload, err := ReadFrame(conn) + if err != nil { + return nil, fmt.Errorf("attachmux: failed to read the greeting: %w", err) + } + if stream != StreamControl { + return nil, fmt.Errorf("attachmux: expected a control greeting, got stream %d", stream) + } + var hello Control + if err := json.Unmarshal(payload, &hello); err != nil { + return nil, fmt.Errorf("attachmux: failed to decode the greeting: %w", err) + } + if hello.Type != ControlHello { + return nil, fmt.Errorf("attachmux: expected a hello, got %q", hello.Type) + } + if hello.Version != ProtocolVersion { + return nil, fmt.Errorf("attachmux: unsupported protocol version %d, expected %d", hello.Version, ProtocolVersion) + } + return &Session{conn: conn, hello: hello}, nil +} + +// TTY reports whether the container was created with a terminal, in which case +// it has a single output stream. +func (s *Session) TTY() bool { return s.hello.TTY } + +// Exited reports whether Stream returned because the broker announced that the +// container is gone, as opposed to the user detaching or the context ending. +// The exit code itself comes from containerd. +func (s *Session) Exited() bool { + s.exitMu.Lock() + defer s.exitMu.Unlock() + return s.exited +} + +// Close closes the session's connection. +func (s *Session) Close() error { return s.conn.Close() } + +func (s *Session) writeFrame(frame []byte) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() + _, err := s.conn.Write(frame) + return err +} + +// Stream pumps the session until the broker announces the container's exit, the +// connection is closed, or ctx is done. Container output goes to stdout and +// stderr; when stderr is nil, the container's stderr is merged into stdout, +// which is what a TTY container needs. +// +// When stdin is non-nil, it is copied to the container in the background. +// Stream does not wait for that copy: a reader that is detachable returns an +// error when the user types the detach sequence, and the caller tears the +// session down. +func (s *Session) Stream(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer) error { + if stdin != nil { + go s.pumpStdin(stdin) + } + + done := make(chan struct{}) + defer close(done) + go func() { + select { + case <-ctx.Done(): + s.conn.Close() + case <-done: + } + }() + + for { + stream, payload, err := ReadFrame(s.conn) + if err != nil { + if ctx.Err() != nil { + // The caller ended this session: a detach, or a cancelled + // command. That is a normal outcome. + return nil + } + // Anything else is the broker going away without saying the + // container exited: the logging process died, or this session was + // dropped for falling behind. Reporting nil here would look exactly + // like a clean detach and the command would exit 0. + if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { + return fmt.Errorf("attachmux: lost the connection to the container attach socket: %w", err) + } + return err + } + + switch stream { + case StreamStdout: + if stdout == nil { + continue + } + if _, err := stdout.Write(payload); err != nil { + return err + } + case StreamStderr: + w := stderr + if w == nil { + w = stdout + } + if w == nil { + continue + } + if _, err := w.Write(payload); err != nil { + return err + } + case StreamControl: + var c Control + if err := json.Unmarshal(payload, &c); err != nil { + continue + } + if c.Type == ControlExit { + s.exitMu.Lock() + s.exited = true + s.exitMu.Unlock() + return nil + } + } + } +} + +func (s *Session) pumpStdin(stdin io.Reader) { + buf := make([]byte, 32<<10) + for { + n, err := stdin.Read(buf) + if n > 0 { + frame, ferr := EncodeFrame(StreamStdin, buf[:n]) + if ferr != nil { + return + } + if werr := s.writeFrame(frame); werr != nil { + return + } + } + if err != nil { + return + } + } +} diff --git a/pkg/attachmux/client_test.go b/pkg/attachmux/client_test.go new file mode 100644 index 00000000000..2f293d3b738 --- /dev/null +++ b/pkg/attachmux/client_test.go @@ -0,0 +1,240 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "bytes" + "context" + "io" + "net" + "strings" + "sync" + "testing" + "time" + + "gotest.tools/v3/assert" +) + +// lockedBuffer is an io.Writer safe for concurrent use. +type lockedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (l *lockedBuffer) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.buf.Write(p) +} + +func (l *lockedBuffer) String() string { + l.mu.Lock() + defer l.mu.Unlock() + return l.buf.String() +} + +// dropAllSessions disconnects every session without announcing an exit, which +// is what a broker whose process died looks like from the client side. +func dropAllSessions(b *Broker) { + b.mu.Lock() + sessions := make([]*session, 0, len(b.sessions)) + for s := range b.sessions { + sessions = append(sessions, s) + } + b.sessions = map[*session]struct{}{} + b.mu.Unlock() + + for _, s := range sessions { + s.close() + } +} + +// pair returns a broker and a session connected to it over an in-memory pipe. +func pair(t *testing.T, tty bool, stdin io.WriteCloser) (*Broker, *Session) { + t.Helper() + + b := NewBroker(tty, stdin) + mine, theirs := net.Pipe() + t.Cleanup(func() { mine.Close() }) + + go b.addSession(theirs) + + s, err := NewSession(mine) + assert.NilError(t, err) + return b, s +} + +func TestSessionHandshakeReportsTTY(t *testing.T) { + t.Parallel() + + _, s := pair(t, true, nil) + assert.Equal(t, s.TTY(), true) +} + +func TestSessionStreamsOutput(t *testing.T) { + t.Parallel() + + b, s := pair(t, false, nil) + waitSessions(t, b, 1) + + stdout := &lockedBuffer{} + stderr := &lockedBuffer{} + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + done := make(chan error, 1) + go func() { done <- s.Stream(ctx, nil, stdout, stderr) }() + + b.Write(StreamStdout, []byte("to stdout")) + b.Write(StreamStderr, []byte("to stderr")) + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if stdout.String() == "to stdout" && stderr.String() == "to stderr" { + break + } + time.Sleep(5 * time.Millisecond) + } + assert.Equal(t, stdout.String(), "to stdout") + assert.Equal(t, stderr.String(), "to stderr") + + b.Close(true) + select { + case err := <-done: + assert.NilError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Stream did not return after the broker announced the exit") + } + + assert.Equal(t, s.Exited(), true) +} + +func TestSessionExitedIsFalseAfterADetach(t *testing.T) { + t.Parallel() + + // Detaching closes the connection without an exit message. The caller uses + // this to tell "the user detached" from "the container is gone". + b, s := pair(t, true, nil) + waitSessions(t, b, 1) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.Stream(ctx, nil, io.Discard, nil) }() + + cancel() + select { + case err := <-done: + assert.NilError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Stream did not return after the context was cancelled") + } + + assert.Equal(t, s.Exited(), false) +} + +func TestSessionMergesStderrIntoStdoutWhenNoStderrWriter(t *testing.T) { + t.Parallel() + + // With a TTY there is a single stream, and callers pass a nil stderr. + b, s := pair(t, true, nil) + waitSessions(t, b, 1) + + stdout := &lockedBuffer{} + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + go func() { _ = s.Stream(ctx, nil, stdout, nil) }() + + b.Write(StreamStderr, []byte("merged")) + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if stdout.String() == "merged" { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("stderr was not merged into stdout, got %q", stdout.String()) +} + +func TestSessionSendsStdin(t *testing.T) { + t.Parallel() + + stdin := &syncBuffer{} + b, s := pair(t, false, stdin) + waitSessions(t, b, 1) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + go func() { _ = s.Stream(ctx, strings.NewReader("typed input\n"), io.Discard, nil) }() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if stdin.String() == "typed input\n" { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("stdin did not reach the broker, got %q", stdin.String()) +} + +func TestSessionStreamReportsAnUnexpectedDisconnect(t *testing.T) { + t.Parallel() + + // The broker dying is not the same as the user detaching. If it were + // reported as a clean end of session, the command would exit 0 and the user + // would never learn that the container's output stopped being delivered. + b, s := pair(t, true, nil) + waitSessions(t, b, 1) + + ctx := context.Background() + done := make(chan error, 1) + go func() { done <- s.Stream(ctx, nil, io.Discard, nil) }() + + // Drop the session the way a dead broker would: close the connection + // without sending an exit message. + dropAllSessions(b) + + select { + case err := <-done: + assert.Assert(t, err != nil, "an unexpected disconnect was reported as a clean detach") + assert.Equal(t, s.Exited(), false) + case <-time.After(5 * time.Second): + t.Fatal("Stream did not return after the broker went away") + } +} + +func TestSessionStreamReturnsWhenContextIsCancelled(t *testing.T) { + t.Parallel() + + b, s := pair(t, true, nil) + waitSessions(t, b, 1) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.Stream(ctx, nil, io.Discard, nil) }() + + cancel() + select { + case err := <-done: + assert.NilError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Stream did not return after the context was cancelled") + } +} From 0d1fefa01ae67a3afacc6a5c09051ca6d9f396bd Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Tue, 1 Sep 2026 11:53:01 +0300 Subject: [PATCH 4/8] feat(attachmux): add the attach socket transport Signed-off-by: Eugene Kalinin --- pkg/attachmux/attachmux.go | 75 ++++++++++++ pkg/attachmux/socket_supported.go | 148 +++++++++++++++++++++++ pkg/attachmux/socket_test.go | 180 ++++++++++++++++++++++++++++ pkg/attachmux/socket_unsupported.go | 33 +++++ 4 files changed, 436 insertions(+) create mode 100644 pkg/attachmux/attachmux.go create mode 100644 pkg/attachmux/socket_supported.go create mode 100644 pkg/attachmux/socket_test.go create mode 100644 pkg/attachmux/socket_unsupported.go diff --git a/pkg/attachmux/attachmux.go b/pkg/attachmux/attachmux.go new file mode 100644 index 00000000000..3ad6b9c21c5 --- /dev/null +++ b/pkg/attachmux/attachmux.go @@ -0,0 +1,75 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +// Package attachmux lets any number of sessions share a container's stdio. +// +// The server side runs inside the process that owns the container's stdio, its +// internal logging process, and the client side runs in every `nerdctl attach`, +// `nerdctl run -it` and `nerdctl start -a`. +package attachmux + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "path/filepath" +) + +// ErrUnsupported is returned on platforms where the attach transport is not +// implemented yet. Callers fall back to attaching directly to the container's +// stdio, which allows a single session at a time. It lives here rather than in +// socket_unsupported.go because callers test for it on every platform. +var ErrUnsupported = errors.New("attachmux: multi-session attach is not supported on this platform") + +// socketNameLen is how many hex characters of the digest go into a socket file +// name. A unix socket path has to fit in sockaddr_un.sun_path (108 bytes), and +// the data store path plus a 64 character container ID does not, hence the +// digest. +const socketNameLen = 16 + +// socketDir returns the directory holding a data store's attach sockets. +func socketDir(dataStore string) string { + return filepath.Join(dataStore, "attach") +} + +// SocketPath returns the attach socket path for a container. +// +// It lives under the container's data store rather than in a runtime directory, +// because the data store is the one location the broker and its clients can +// agree on. The broker runs inside the logging process, which containerd's shim +// spawns with an environment of exactly CONTAINER_ID and CONTAINER_NAMESPACE +// (cmd/containerd-shim-runc-v2/process/io_util.go, NewBinaryCmd): there is no +// XDG_RUNTIME_DIR there, and rootlessutil.XDGRuntimeDir would fail outright. +// The data store, by contrast, is handed to that process in its argv, and +// nerdctl already relies on it being the same path on both sides for the log +// files themselves. +// +// The name is a digest rather than a prefix of the ID, because all of a data +// store's attach sockets share one flat directory and Listen removes whatever +// is already at the path, so a collision would silently hand one container's +// terminal to another. The namespace is in the digest because nerdctl can +// attach to a container created by another client, whose ID may be short and +// identical to one elsewhere; nerdctl's own IDs are 64 random hex characters +// (see pkg/idgen). The data store no longer needs to be, since it is the +// directory. +// +// This is a pure function of its arguments, and it is how both the broker and +// every client find the socket: nothing about it is recorded on disk. See the +// head of Task 4 for why there is no pointer file. +func SocketPath(dataStore, ns, id string) string { + sum := sha256.Sum256([]byte(ns + "\x00" + id)) + return filepath.Join(socketDir(dataStore), hex.EncodeToString(sum[:])[:socketNameLen]+".sock") +} diff --git a/pkg/attachmux/socket_supported.go b/pkg/attachmux/socket_supported.go new file mode 100644 index 00000000000..e724f6b1c67 --- /dev/null +++ b/pkg/attachmux/socket_supported.go @@ -0,0 +1,148 @@ +//go:build linux || freebsd + +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "context" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "sync/atomic" + "time" +) + +// probeSeq keeps two concurrent probes inside one process off each other's +// socket path. Different processes are separated by the pid in the name. +var probeSeq atomic.Uint64 + +// Probe reports whether a container's attach socket directory can be created +// and bound in. +// +// nerdctl has to decide whether to hand a container's stdio to the logging +// process before the task exists, but whether the broker actually came up is +// only known afterwards. Probing first turns the common failures into a clean +// fallback to the legacy path, instead of a container whose output nobody can +// show. +func Probe(dataStore string) error { + dir := socketDir(dataStore) + if err := os.MkdirAll(dir, 0700); err != nil { + return err + } + + // Bind a real socket rather than creating a regular file: a file proves + // neither that the filesystem supports unix sockets nor that the path fits + // in sockaddr_un.sun_path, which a long data store path can break. + // + // The name is the same length as a real socket name so that the sun_path + // check is representative. It carries the pid so that concurrent nerdctl + // invocations do not probe over each other, and a counter so that two + // concurrent calls inside one process do not either. Real names are hex + // digests, so a name starting with "probe" cannot collide with one. + name := fmt.Sprintf("probe%d-%d", os.Getpid(), probeSeq.Add(1)) + if len(name) < socketNameLen { + name += strings.Repeat("p", socketNameLen-len(name)) + } + path := filepath.Join(dir, name+".sock") + + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + l, err := net.Listen("unix", path) + if err != nil { + return err + } + // Closing a unix listener unlinks its socket. + return l.Close() +} + +// Listen creates the attach socket at path with mode 0600. +func Listen(path string) (net.Listener, error) { + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return nil, err + } + // A container that was killed leaves its socket behind, and bind would then + // fail with EADDRINUSE. + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, err + } + l, err := net.Listen("unix", path) + if err != nil { + return nil, err + } + if err := os.Chmod(path, 0600); err != nil { + l.Close() + return nil, err + } + return l, nil +} + +const ( + // dialTimeout bounds how long Dial waits for the broker to come up. + dialTimeout = 5 * time.Second + // dialInterval is how often Dial retries while waiting. + dialInterval = 20 * time.Millisecond +) + +// Dial connects to the attach socket at path and completes the handshake. +// +// It retries until the socket answers, the context is done, or dialTimeout +// elapses. containerd spawns the logging process while creating the task and +// does not wait for it, so a client that dials straight after +// container.NewTask can arrive before the broker has bound its socket. For a +// container that is already running the first attempt succeeds. +func Dial(ctx context.Context, path string) (*Session, error) { + ctx, cancel := context.WithTimeout(ctx, dialTimeout) + defer cancel() + + var d net.Dialer + for { + conn, err := d.DialContext(ctx, "unix", path) + if err == nil { + // A connection can sit in the listener's backlog before the broker + // accepts it, so the handshake read needs a deadline of its own: + // a broker that died between Listen and Serve would otherwise hang + // the client for good. + if deadline, ok := ctx.Deadline(); ok { + if err := conn.SetReadDeadline(deadline); err != nil { + conn.Close() + return nil, err + } + } + session, err := NewSession(conn) + if err != nil { + conn.Close() + return nil, err + } + // Streaming has no deadline of its own. + if err := conn.SetReadDeadline(time.Time{}); err != nil { + session.Close() + return nil, err + } + return session, nil + } + select { + case <-ctx.Done(): + return nil, fmt.Errorf("failed to connect to %s: %w", path, err) + case <-time.After(dialInterval): + } + } +} diff --git a/pkg/attachmux/socket_test.go b/pkg/attachmux/socket_test.go new file mode 100644 index 00000000000..798b212303c --- /dev/null +++ b/pkg/attachmux/socket_test.go @@ -0,0 +1,180 @@ +//go:build linux || freebsd + +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "context" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "gotest.tools/v3/assert" +) + +func TestSocketPathIsShortEnoughForSunPath(t *testing.T) { + t.Parallel() + + // A unix socket path has to fit in sockaddr_un.sun_path, which is 108 + // bytes. A container ID is 64 hex characters, so the full ID cannot be part + // of the path. + id := strings.Repeat("a", 64) + path := SocketPath("/var/lib/nerdctl/1935db59", "default", id) + assert.Assert(t, len(path) < 100, "socket path %q is %d bytes", path, len(path)) + assert.Assert(t, strings.HasSuffix(path, ".sock"), "got %q", path) +} + +func TestSocketPathIsStableAndDistinct(t *testing.T) { + t.Parallel() + + first := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("a", 64)) + again := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("a", 64)) + assert.Equal(t, first, again) + + other := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("b", 64)) + assert.Assert(t, first != other) +} + +func TestSocketPathSeparatesNamespaces(t *testing.T) { + t.Parallel() + + // A data store's attach sockets share one flat directory and Listen removes + // whatever is already at the path, so two containers with the same short ID + // in different namespaces must not resolve to the same socket. nerdctl's own + // IDs are random and 64 characters long, but nerdctl can attach to a + // container created by another client, whose ID may be anything. + first := SocketPath("/var/lib/nerdctl/1935db59", "default", "shell") + other := SocketPath("/var/lib/nerdctl/1935db59", "staging", "shell") + assert.Assert(t, first != other, "both namespaces resolved to %q", first) +} + +func TestSocketPathSeparatesContainerdEndpoints(t *testing.T) { + t.Parallel() + + // The data store path encodes the containerd address, and the socket lives + // inside it, so two endpoints cannot collide. + first := SocketPath("/var/lib/nerdctl/1935db59", "default", "shell") + other := SocketPath("/var/lib/nerdctl/8fa1c02b", "default", "shell") + assert.Assert(t, first != other, "both endpoints resolved to %q", first) +} + +func TestSocketPathIsUnderTheDataStore(t *testing.T) { + t.Parallel() + + // The broker derives this from the data store it is handed in its argv, + // which is the only path it and its clients are guaranteed to agree on: the + // shim spawns it with an environment of just CONTAINER_ID and + // CONTAINER_NAMESPACE, so nothing XDG-based is available there. + path := SocketPath("/var/lib/nerdctl/1935db59", "default", "abc") + assert.Equal(t, filepath.Dir(path), "/var/lib/nerdctl/1935db59/attach") + assert.Equal(t, len(filepath.Base(path)), socketNameLen+len(".sock")) +} + +func TestListenAndDial(t *testing.T) { + t.Parallel() + + path := filepath.Join(t.TempDir(), "attach.sock") + + b := NewBroker(true, nil) + l, err := Listen(path) + assert.NilError(t, err) + t.Cleanup(func() { l.Close() }) + + info, err := os.Stat(path) + assert.NilError(t, err) + assert.Equal(t, info.Mode().Perm(), os.FileMode(0600)) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = b.Serve(ctx, l) }() + + session, err := Dial(ctx, path) + assert.NilError(t, err) + t.Cleanup(func() { session.Close() }) + assert.Equal(t, session.TTY(), true) +} + +func TestListenReplacesALeftoverSocket(t *testing.T) { + t.Parallel() + + // A container that was killed leaves its socket behind, and bind would then + // fail with EADDRINUSE. Go's unix listener unlinks the socket on Close, so + // the leftover has to be staged by hand. + path := filepath.Join(t.TempDir(), "attach.sock") + assert.NilError(t, os.WriteFile(path, nil, 0600)) + + l, err := Listen(path) + assert.NilError(t, err) + t.Cleanup(func() { l.Close() }) +} + +func TestProbe(t *testing.T) { + t.Parallel() + + // Probe has to leave nothing behind: it runs on every `nerdctl run -it`. + dataStore := t.TempDir() + assert.NilError(t, Probe(dataStore)) + + entries, err := os.ReadDir(socketDir(dataStore)) + assert.NilError(t, err) + for _, e := range entries { + assert.Assert(t, !strings.HasPrefix(e.Name(), "probe"), "Probe left %q behind", e.Name()) + } +} + +func TestProbeIsRepeatable(t *testing.T) { + t.Parallel() + + // Probe binds a real socket, so it has to clean up after itself well enough + // to run again, which it does on every foreground `nerdctl run -it`. + dataStore := t.TempDir() + assert.NilError(t, Probe(dataStore)) + assert.NilError(t, Probe(dataStore)) +} + +func TestProbeIsConcurrencySafe(t *testing.T) { + t.Parallel() + + // Nothing stops two goroutines in one process from probing at once, and a + // shared socket name would make one unlink the other's socket. + dataStore := t.TempDir() + var wg sync.WaitGroup + errs := make([]error, 8) + for i := range errs { + wg.Go(func() { errs[i] = Probe(dataStore) }) + } + wg.Wait() + for _, err := range errs { + assert.NilError(t, err) + } +} + +func TestDialToNothing(t *testing.T) { + t.Parallel() + + // Dial retries while waiting for the broker to bind, so it gives up only + // when the context is done. Keep that short here. + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + t.Cleanup(cancel) + + _, err := Dial(ctx, filepath.Join(t.TempDir(), "absent.sock")) + assert.Assert(t, err != nil) +} diff --git a/pkg/attachmux/socket_unsupported.go b/pkg/attachmux/socket_unsupported.go new file mode 100644 index 00000000000..5f824ef1131 --- /dev/null +++ b/pkg/attachmux/socket_unsupported.go @@ -0,0 +1,33 @@ +//go:build !(linux || freebsd) + +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "context" + "net" +) + +// Probe is not implemented on this platform, so callers keep the legacy path. +func Probe(string) error { return ErrUnsupported } + +// Listen is not implemented on this platform. +func Listen(string) (net.Listener, error) { return nil, ErrUnsupported } + +// Dial is not implemented on this platform. +func Dial(context.Context, string) (*Session, error) { return nil, ErrUnsupported } From e95c2f7575353941c9d2582f05d1b9f21c3dc21f Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Tue, 1 Sep 2026 20:56:39 +0300 Subject: [PATCH 5/8] fix(attachmux): keep a session's output alive when its stdin write fails A failed write to the container's stdin returned from readLoop, whose deferred dropSession took the whole session with it. A container that closed its stdin and kept printing lost its terminal on the next keystroke, and one that had just exited was reported as a broken session instead of a clean exit. Also from the same review pass: - Write splits a chunk larger than one frame instead of dropping it, and refuses the control stream, which is the broker's own. - EncodeFrame moved out of b.mu. - Dial stops retrying a refused connection, which unlike a missing socket means nothing is listening. - Session.Close is no longer reported by Stream as a lost connection. - The stdin pump stops forwarding once Stream returns, so it no longer swallows the next keystroke. - Serve's context watchdog no longer leaks when Serve returns first. - Probe trims an over-long name instead of probing a longer path than a real socket. - ErrUnsupported wraps errors.ErrUnsupported. - SocketPath tests no longer sit behind a build tag, and the unsupported stubs are covered. - RemoveSocket moves here from the logging step: it is package API. Signed-off-by: Eugene Kalinin --- pkg/attachmux/attachmux.go | 6 +- pkg/attachmux/attachmux_test.go | 83 ++++++++++++++++++++++++ pkg/attachmux/broker.go | 57 ++++++++++++---- pkg/attachmux/broker_test.go | 75 +++++++++++++++++++++ pkg/attachmux/client.go | 46 +++++++++++-- pkg/attachmux/client_test.go | 25 +++++++ pkg/attachmux/proto.go | 12 ++-- pkg/attachmux/socket_supported.go | 34 +++++++++- pkg/attachmux/socket_test.go | 58 ----------------- pkg/attachmux/socket_unsupported.go | 3 + pkg/attachmux/socket_unsupported_test.go | 45 +++++++++++++ 11 files changed, 361 insertions(+), 83 deletions(-) create mode 100644 pkg/attachmux/attachmux_test.go create mode 100644 pkg/attachmux/socket_unsupported_test.go diff --git a/pkg/attachmux/attachmux.go b/pkg/attachmux/attachmux.go index 3ad6b9c21c5..9da356fad2e 100644 --- a/pkg/attachmux/attachmux.go +++ b/pkg/attachmux/attachmux.go @@ -25,6 +25,7 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "fmt" "path/filepath" ) @@ -32,7 +33,10 @@ import ( // implemented yet. Callers fall back to attaching directly to the container's // stdio, which allows a single session at a time. It lives here rather than in // socket_unsupported.go because callers test for it on every platform. -var ErrUnsupported = errors.New("attachmux: multi-session attach is not supported on this platform") +// +// It wraps errors.ErrUnsupported, so a caller that does not care which +// subsystem is unavailable can test for that instead. +var ErrUnsupported = fmt.Errorf("attachmux: multi-session attach is not supported on this platform: %w", errors.ErrUnsupported) // socketNameLen is how many hex characters of the digest go into a socket file // name. A unix socket path has to fit in sockaddr_un.sun_path (108 bytes), and diff --git a/pkg/attachmux/attachmux_test.go b/pkg/attachmux/attachmux_test.go new file mode 100644 index 00000000000..672c9afbee3 --- /dev/null +++ b/pkg/attachmux/attachmux_test.go @@ -0,0 +1,83 @@ +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "path/filepath" + "strings" + "testing" + + "gotest.tools/v3/assert" +) + +func TestSocketPathIsShortEnoughForSunPath(t *testing.T) { + t.Parallel() + + // A unix socket path has to fit in sockaddr_un.sun_path, which is 108 + // bytes. A container ID is 64 hex characters, so the full ID cannot be part + // of the path. + id := strings.Repeat("a", 64) + path := SocketPath("/var/lib/nerdctl/1935db59", "default", id) + assert.Assert(t, len(path) < 100, "socket path %q is %d bytes", path, len(path)) + assert.Assert(t, strings.HasSuffix(path, ".sock"), "got %q", path) +} + +func TestSocketPathIsStableAndDistinct(t *testing.T) { + t.Parallel() + + first := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("a", 64)) + again := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("a", 64)) + assert.Equal(t, first, again) + + other := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("b", 64)) + assert.Assert(t, first != other) +} + +func TestSocketPathSeparatesNamespaces(t *testing.T) { + t.Parallel() + + // A data store's attach sockets share one flat directory and Listen removes + // whatever is already at the path, so two containers with the same short ID + // in different namespaces must not resolve to the same socket. nerdctl's own + // IDs are random and 64 characters long, but nerdctl can attach to a + // container created by another client, whose ID may be anything. + first := SocketPath("/var/lib/nerdctl/1935db59", "default", "shell") + other := SocketPath("/var/lib/nerdctl/1935db59", "staging", "shell") + assert.Assert(t, first != other, "both namespaces resolved to %q", first) +} + +func TestSocketPathSeparatesContainerdEndpoints(t *testing.T) { + t.Parallel() + + // The data store path encodes the containerd address, and the socket lives + // inside it, so two endpoints cannot collide. + first := SocketPath("/var/lib/nerdctl/1935db59", "default", "shell") + other := SocketPath("/var/lib/nerdctl/8fa1c02b", "default", "shell") + assert.Assert(t, first != other, "both endpoints resolved to %q", first) +} + +func TestSocketPathIsUnderTheDataStore(t *testing.T) { + t.Parallel() + + // The broker derives this from the data store it is handed in its argv, + // which is the only path it and its clients are guaranteed to agree on: the + // shim spawns it with an environment of just CONTAINER_ID and + // CONTAINER_NAMESPACE, so nothing XDG-based is available there. + path := SocketPath("/var/lib/nerdctl/1935db59", "default", "abc") + assert.Equal(t, filepath.Dir(path), "/var/lib/nerdctl/1935db59/attach") + assert.Equal(t, len(filepath.Base(path)), socketNameLen+len(".sock")) +} diff --git a/pkg/attachmux/broker.go b/pkg/attachmux/broker.go index fc40e1216c2..8feb516b86c 100644 --- a/pkg/attachmux/broker.go +++ b/pkg/attachmux/broker.go @@ -158,19 +158,38 @@ func NewBroker(tty bool, stdin io.WriteCloser) *Broker { // queue is full is disconnected instead. p is copied, so the caller is free to // reuse it immediately. func (b *Broker) Write(stream byte, p []byte) { - if len(p) == 0 { + if stream == StreamControl { + // Control is the broker's own channel. Letting the owner of the stdio + // put frames on it would let container output forge a hello or an exit. + log.L.Warn("attachmux: refusing to write container output on the control stream") return } + // A chunk larger than a single frame is split rather than dropped: Write is + // the exported entry point for whoever owns the container's stdio, and the + // size of its read buffer is not this package's to assume. + for len(p) > maxPayload { + b.writeFrame(stream, p[:maxPayload]) + p = p[maxPayload:] + } + b.writeFrame(stream, p) +} - b.mu.Lock() - if b.closed || len(b.sessions) == 0 { - b.mu.Unlock() +func (b *Broker) writeFrame(stream byte, p []byte) { + if len(p) == 0 { return } + + // Encoded outside the lock: it allocates and copies the whole chunk, and + // readLoop, addSession and SessionCount all contend for b.mu. frame, err := EncodeFrame(stream, p) if err != nil { + log.L.WithError(err).Warn("attachmux: dropping an output frame") + return + } + + b.mu.Lock() + if b.closed || len(b.sessions) == 0 { b.mu.Unlock() - log.L.WithError(err).Warn("attachmux: dropping an oversized output frame") return } var slow []*session @@ -199,9 +218,14 @@ func (b *Broker) SessionCount() int { // Serve accepts sessions on l until ctx is done or l is closed. func (b *Broker) Serve(ctx context.Context, l net.Listener) error { + done := make(chan struct{}) + defer close(done) go func() { - <-ctx.Done() - l.Close() + select { + case <-ctx.Done(): + l.Close() + case <-done: + } }() for { @@ -292,8 +316,11 @@ func (b *Broker) Close(exited bool) { } // SetStdin gives the broker the write end of the container's stdin once it is -// known to be usable, and reports whether the broker took it. It returns false -// for a broker that is already closed, and the caller then closes w itself. +// known to be usable, and reports whether the broker took it. +// +// It returns false when the broker is already closed, and also when it already +// has a stdin, whether from NewBroker or from an earlier call. The caller then +// closes w itself. A false is never evidence that the container exited. // // Stdin arrives late because whether the container has one at all can only be // established by opening the FIFO, which the owner has to do off the path that @@ -400,10 +427,16 @@ func (b *Broker) readLoop(s *session) { b.stdinWriteMu.Unlock() if err != nil { // Close closed the descriptor under a blocked write, or the - // container is gone. Either way this session's input has - // nowhere to go. + // container stopped reading its stdin. Either way this + // keystroke has nowhere to go, and nothing more. + // + // Returning here would run the deferred dropSession and take + // the session's *output* with it: a container that closed its + // stdin and kept printing would lose its terminal on the next + // keypress, and one that had just exited would be reported as a + // broken session instead of a clean exit. log.L.WithError(err).Debug("attachmux: failed to write to the container stdin") - return + continue } } } diff --git a/pkg/attachmux/broker_test.go b/pkg/attachmux/broker_test.go index e8ad661a715..2881bb40108 100644 --- a/pkg/attachmux/broker_test.go +++ b/pkg/attachmux/broker_test.go @@ -23,6 +23,7 @@ import ( "net" "os" "sync" + "syscall" "testing" "time" @@ -156,6 +157,48 @@ func TestBrokerMergesStdinFromEverySession(t *testing.T) { assert.Assert(t, bytes.Contains([]byte(got), []byte("from-second\n")), "got %q", got) } +func TestBrokerSplitsAnOversizedWrite(t *testing.T) { + t.Parallel() + + // Write is the entry point for whoever owns the container's stdio, and this + // package does not get to assume the size of their read buffer. A chunk + // larger than one frame is split, not dropped. + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + payload := bytes.Repeat([]byte("x"), maxPayload+1000) + go b.Write(StreamStdout, payload) + + var got []byte + for len(got) < len(payload) { + stream, chunk, err := ReadFrame(conn) + assert.NilError(t, err) + assert.Equal(t, stream, StreamStdout) + got = append(got, chunk...) + } + assert.Equal(t, len(got), len(payload)) + assert.DeepEqual(t, got, payload) +} + +func TestBrokerRefusesToWriteOnTheControlStream(t *testing.T) { + t.Parallel() + + // Control is the broker's own channel: container output must not be able to + // forge a hello or an exit. + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + b.Write(StreamControl, []byte(`{"type":"exit"}`)) + b.Write(StreamStdout, []byte("real output")) + + stream, payload, err := ReadFrame(conn) + assert.NilError(t, err) + assert.Equal(t, stream, StreamStdout) + assert.Equal(t, string(payload), "real output") +} + func TestBrokerDisconnectsASessionThatStopsReading(t *testing.T) { t.Parallel() @@ -253,6 +296,38 @@ func TestBrokerAcceptsStdinAfterASessionConnects(t *testing.T) { t.Fatalf("stdin did not reach the broker, got %q", stdin.String()) } +// failingWriter models the container's stdin FIFO after the container stopped +// reading it: every write fails, the way a FIFO with no reader gives EPIPE. +type failingWriter struct{} + +func (failingWriter) Write([]byte) (int, error) { return 0, syscall.EPIPE } +func (failingWriter) Close() error { return nil } + +func TestBrokerKeepsAnOutputSessionAfterAFailedStdinWrite(t *testing.T) { + t.Parallel() + + // A container that closed its stdin and keeps printing, or one that has + // just exited, must not cost the session its output on the next keystroke. + // Dropping the session there also loses the exit frame, which is exactly + // the "unexpected disconnect is an error, not a detach" case. + b := NewBroker(true, failingWriter{}) + conn := connect(t, b) + waitSessions(t, b, 1) + + frame, err := EncodeFrame(StreamStdin, []byte("keystroke")) + assert.NilError(t, err) + _, err = conn.Write(frame) + assert.NilError(t, err) + + // The session is still there, and still receives the container's output. + b.Write(StreamStdout, []byte("still printing")) + stream, payload, err := ReadFrame(conn) + assert.NilError(t, err) + assert.Equal(t, stream, StreamStdout) + assert.Equal(t, string(payload), "still printing") + assert.Equal(t, b.SessionCount(), 1) +} + func TestBrokerCloseReleasesStdin(t *testing.T) { t.Parallel() diff --git a/pkg/attachmux/client.go b/pkg/attachmux/client.go index 056b1c8ca60..e883b34873f 100644 --- a/pkg/attachmux/client.go +++ b/pkg/attachmux/client.go @@ -34,8 +34,17 @@ type Session struct { // writeMu serialises writes to the connection. writeMu sync.Mutex + // exitMu guards exited and closedByCaller. exitMu sync.Mutex exited bool + // closedByCaller records that Close was called, so that Stream can tell a + // deliberate teardown from the broker vanishing. + closedByCaller bool + + // stopStdin is closed when Stream returns, so that the stdin pump stops + // forwarding what it is about to read. It is only set up when Stream was + // given a stdin. + stopStdin chan struct{} } // NewSession completes the handshake on conn and returns the session. The @@ -75,7 +84,21 @@ func (s *Session) Exited() bool { } // Close closes the session's connection. -func (s *Session) Close() error { return s.conn.Close() } +// Close tears the session down. A Stream running concurrently returns nil +// rather than an error: this is the caller ending the session, not the broker +// going away. +func (s *Session) wasClosedByCaller() bool { + s.exitMu.Lock() + defer s.exitMu.Unlock() + return s.closedByCaller +} + +func (s *Session) Close() error { + s.exitMu.Lock() + s.closedByCaller = true + s.exitMu.Unlock() + return s.conn.Close() +} func (s *Session) writeFrame(frame []byte) error { s.writeMu.Lock() @@ -95,6 +118,7 @@ func (s *Session) writeFrame(frame []byte) error { // session down. func (s *Session) Stream(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer) error { if stdin != nil { + s.stopStdin = make(chan struct{}) go s.pumpStdin(stdin) } @@ -108,12 +132,20 @@ func (s *Session) Stream(ctx context.Context, stdin io.Reader, stdout, stderr io } }() + if stdin != nil { + // pumpStdin blocks in stdin.Read and cannot be interrupted, so it is + // left to end on its own. Tell it to stop forwarding once Stream is + // done, otherwise the keystroke it is already waiting for is swallowed + // from the terminal rather than reaching whatever runs next. + defer close(s.stopStdin) + } + for { stream, payload, err := ReadFrame(s.conn) if err != nil { - if ctx.Err() != nil { - // The caller ended this session: a detach, or a cancelled - // command. That is a normal outcome. + if ctx.Err() != nil || s.wasClosedByCaller() { + // The caller ended this session: a detach, a cancelled command, + // or an explicit Close. That is a normal outcome. return nil } // Anything else is the broker going away without saying the @@ -165,6 +197,12 @@ func (s *Session) pumpStdin(stdin io.Reader) { for { n, err := stdin.Read(buf) if n > 0 { + select { + case <-s.stopStdin: + // Stream has returned; this session is over. + return + default: + } frame, ferr := EncodeFrame(StreamStdin, buf[:n]) if ferr != nil { return diff --git a/pkg/attachmux/client_test.go b/pkg/attachmux/client_test.go index 2f293d3b738..c2ccfe2fa12 100644 --- a/pkg/attachmux/client_test.go +++ b/pkg/attachmux/client_test.go @@ -238,3 +238,28 @@ func TestSessionStreamReturnsWhenContextIsCancelled(t *testing.T) { t.Fatal("Stream did not return after the context was cancelled") } } + +func TestSessionStreamTreatsAnExplicitCloseAsClean(t *testing.T) { + t.Parallel() + + // Stream's own doc says the caller tears the session down, and Close is the + // obvious way. Reporting that as a lost connection would be the same false + // positive the unexpected-disconnect branch exists to avoid. + b, session := pair(t, true, nil) + _ = b + + streamed := make(chan error, 1) + go func() { + streamed <- session.Stream(context.Background(), nil, io.Discard, nil) + }() + + time.Sleep(100 * time.Millisecond) + assert.NilError(t, session.Close()) + + select { + case err := <-streamed: + assert.NilError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Stream did not return after Close") + } +} diff --git a/pkg/attachmux/proto.go b/pkg/attachmux/proto.go index 35ef1db9a33..41c1a69fca2 100644 --- a/pkg/attachmux/proto.go +++ b/pkg/attachmux/proto.go @@ -14,13 +14,6 @@ limitations under the License. */ -// Package attachmux multiplexes a container's stdio between the process that -// owns it and any number of attached CLI sessions. -// -// A container's stdio FIFOs cannot be shared: a FIFO has a single queue, so two -// readers on the stdout FIFO each receive a random subset of the container's -// output. Instead, one process owns the FIFOs and every session talks to it -// over a socket, framed with the protocol in this file. package attachmux import ( @@ -31,6 +24,11 @@ import ( "io" ) +// The wire format between the broker and its sessions: one stream byte, three +// reserved zero bytes, a big-endian uint32 payload length, then the payload. +// This is the same shape docker uses for its multiplexed stdio, with stream 3 +// added for control messages. + // Stream identifiers carried in a frame header. const ( StreamStdin byte = 0 diff --git a/pkg/attachmux/socket_supported.go b/pkg/attachmux/socket_supported.go index e724f6b1c67..e74452ccfcd 100644 --- a/pkg/attachmux/socket_supported.go +++ b/pkg/attachmux/socket_supported.go @@ -27,6 +27,7 @@ import ( "path/filepath" "strings" "sync/atomic" + "syscall" "time" ) @@ -58,8 +59,15 @@ func Probe(dataStore string) error { // concurrent calls inside one process do not either. Real names are hex // digests, so a name starting with "probe" cannot collide with one. name := fmt.Sprintf("probe%d-%d", os.Getpid(), probeSeq.Add(1)) - if len(name) < socketNameLen { + switch { + case len(name) < socketNameLen: name += strings.Repeat("p", socketNameLen-len(name)) + case len(name) > socketNameLen: + // A large pid plus the counter can overrun. Trim rather than probe with + // a longer name than a real socket: the point is to test a path of + // representative length, and a longer one would reject a data store + // that in fact fits. + name = name[:socketNameLen] } path := filepath.Join(dir, name+".sock") @@ -100,6 +108,9 @@ const ( dialTimeout = 5 * time.Second // dialInterval is how often Dial retries while waiting. dialInterval = 20 * time.Millisecond + // refusedGrace bounds how long Dial keeps retrying after a connection was + // refused, which unlike a missing socket means nothing is listening. + refusedGrace = 200 * time.Millisecond ) // Dial connects to the attach socket at path and completes the handshake. @@ -113,6 +124,7 @@ func Dial(ctx context.Context, path string) (*Session, error) { ctx, cancel := context.WithTimeout(ctx, dialTimeout) defer cancel() + refusedDeadline := time.Now().Add(refusedGrace) var d net.Dialer for { conn, err := d.DialContext(ctx, "unix", path) @@ -139,6 +151,16 @@ func Dial(ctx context.Context, path string) (*Session, error) { } return session, nil } + if errors.Is(err, syscall.ECONNREFUSED) { + // The socket file is there but nothing is listening. A broker that + // was killed leaves the file behind, because Go only unlinks it on a + // graceful Close, so this is proof rather than a race worth waiting + // out. Retry only long enough to cover Listen's own window between + // os.Remove and bind. + if time.Now().After(refusedDeadline) { + return nil, fmt.Errorf("failed to connect to %s: %w", path, err) + } + } select { case <-ctx.Done(): return nil, fmt.Errorf("failed to connect to %s: %w", path, err) @@ -146,3 +168,13 @@ func Dial(ctx context.Context, path string) (*Session, error) { } } } + +// RemoveSocket deletes an attach socket. Removing one that is not there is not +// an error. +func RemoveSocket(path string) error { + err := os.Remove(path) + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err +} diff --git a/pkg/attachmux/socket_test.go b/pkg/attachmux/socket_test.go index 798b212303c..74f2f0b6f01 100644 --- a/pkg/attachmux/socket_test.go +++ b/pkg/attachmux/socket_test.go @@ -30,64 +30,6 @@ import ( "gotest.tools/v3/assert" ) -func TestSocketPathIsShortEnoughForSunPath(t *testing.T) { - t.Parallel() - - // A unix socket path has to fit in sockaddr_un.sun_path, which is 108 - // bytes. A container ID is 64 hex characters, so the full ID cannot be part - // of the path. - id := strings.Repeat("a", 64) - path := SocketPath("/var/lib/nerdctl/1935db59", "default", id) - assert.Assert(t, len(path) < 100, "socket path %q is %d bytes", path, len(path)) - assert.Assert(t, strings.HasSuffix(path, ".sock"), "got %q", path) -} - -func TestSocketPathIsStableAndDistinct(t *testing.T) { - t.Parallel() - - first := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("a", 64)) - again := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("a", 64)) - assert.Equal(t, first, again) - - other := SocketPath("/var/lib/nerdctl/1935db59", "default", strings.Repeat("b", 64)) - assert.Assert(t, first != other) -} - -func TestSocketPathSeparatesNamespaces(t *testing.T) { - t.Parallel() - - // A data store's attach sockets share one flat directory and Listen removes - // whatever is already at the path, so two containers with the same short ID - // in different namespaces must not resolve to the same socket. nerdctl's own - // IDs are random and 64 characters long, but nerdctl can attach to a - // container created by another client, whose ID may be anything. - first := SocketPath("/var/lib/nerdctl/1935db59", "default", "shell") - other := SocketPath("/var/lib/nerdctl/1935db59", "staging", "shell") - assert.Assert(t, first != other, "both namespaces resolved to %q", first) -} - -func TestSocketPathSeparatesContainerdEndpoints(t *testing.T) { - t.Parallel() - - // The data store path encodes the containerd address, and the socket lives - // inside it, so two endpoints cannot collide. - first := SocketPath("/var/lib/nerdctl/1935db59", "default", "shell") - other := SocketPath("/var/lib/nerdctl/8fa1c02b", "default", "shell") - assert.Assert(t, first != other, "both endpoints resolved to %q", first) -} - -func TestSocketPathIsUnderTheDataStore(t *testing.T) { - t.Parallel() - - // The broker derives this from the data store it is handed in its argv, - // which is the only path it and its clients are guaranteed to agree on: the - // shim spawns it with an environment of just CONTAINER_ID and - // CONTAINER_NAMESPACE, so nothing XDG-based is available there. - path := SocketPath("/var/lib/nerdctl/1935db59", "default", "abc") - assert.Equal(t, filepath.Dir(path), "/var/lib/nerdctl/1935db59/attach") - assert.Equal(t, len(filepath.Base(path)), socketNameLen+len(".sock")) -} - func TestListenAndDial(t *testing.T) { t.Parallel() diff --git a/pkg/attachmux/socket_unsupported.go b/pkg/attachmux/socket_unsupported.go index 5f824ef1131..ab335d23c79 100644 --- a/pkg/attachmux/socket_unsupported.go +++ b/pkg/attachmux/socket_unsupported.go @@ -31,3 +31,6 @@ func Listen(string) (net.Listener, error) { return nil, ErrUnsupported } // Dial is not implemented on this platform. func Dial(context.Context, string) (*Session, error) { return nil, ErrUnsupported } + +// RemoveSocket is not implemented on this platform. +func RemoveSocket(string) error { return nil } diff --git a/pkg/attachmux/socket_unsupported_test.go b/pkg/attachmux/socket_unsupported_test.go new file mode 100644 index 00000000000..f77b66c6212 --- /dev/null +++ b/pkg/attachmux/socket_unsupported_test.go @@ -0,0 +1,45 @@ +//go:build !(linux || freebsd) + +/* + Copyright The containerd Authors. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package attachmux + +import ( + "context" + "testing" + + "gotest.tools/v3/assert" +) + +// On these platforms the stubs are the only transport code that runs, and +// callers branch on ErrUnsupported: run -it uses Probe to decide whether to +// hand the container's stdio to the logging process at all. A stub that +// returned nil would send it down a path with no broker behind it. +func TestTransportIsUnsupported(t *testing.T) { + t.Parallel() + + assert.ErrorIs(t, Probe(t.TempDir()), ErrUnsupported) + + _, err := Listen("ignored") + assert.ErrorIs(t, err, ErrUnsupported) + + _, err = Dial(context.Background(), "ignored") + assert.ErrorIs(t, err, ErrUnsupported) + + // Removing a socket that was never created is not an error anywhere. + assert.NilError(t, RemoveSocket("ignored")) +} From 08ea3701519019cce30404ad969b6fab59bfc141 Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Tue, 1 Sep 2026 21:12:12 +0300 Subject: [PATCH 6/8] fix(attachmux): bound a session's queue by bytes, and say why it was dropped Four decisions from the review that are cheapest to settle while the package still has no consumers. A session's queue is bounded by bytes rather than frames. A TTY echoes single keystrokes, so a frame count evicted a session that was a couple of kilobytes behind, while a frame can be up to maxPayload, so a frame count also put no real ceiling on memory in the logging process. An evicted session is told so with a control frame before it is disconnected. Otherwise falling behind is indistinguishable from the broker dying, and the client reports both as a lost connection. Listen removes the socket against its inode instead of its path. Two brokers can briefly overlap for one container while the restart monitor swaps tasks; the new one taking the path over is correct, but the old one exiting afterwards would unlink the live socket and leave the new broker on an inode nobody can reach. A client now refuses only a broker newer than itself. A broker lives as long as its container, so a client that is newer is the normal case after an upgrade, and it has nowhere to fall back to. Signed-off-by: Eugene Kalinin --- pkg/attachmux/broker.go | 45 ++++++++++++++++++++++++---- pkg/attachmux/broker_test.go | 50 +++++++++++++++++++++++++++++++ pkg/attachmux/client.go | 17 +++++++++-- pkg/attachmux/client_test.go | 40 +++++++++++++++++++++++++ pkg/attachmux/proto.go | 13 ++++++-- pkg/attachmux/socket_supported.go | 50 ++++++++++++++++++++++++++++++- pkg/attachmux/socket_test.go | 27 +++++++++++++++++ 7 files changed, 230 insertions(+), 12 deletions(-) diff --git a/pkg/attachmux/broker.go b/pkg/attachmux/broker.go index 8feb516b86c..327d0d57bf0 100644 --- a/pkg/attachmux/broker.go +++ b/pkg/attachmux/broker.go @@ -28,11 +28,19 @@ import ( ) const ( - // defaultQueueDepth is how many frames are buffered for a single session. A - // session that falls this far behind is disconnected: the container's output - // must never wait on a consumer. + // maxQueuedBytes is how much of the container's output is buffered for a + // single session before it is disconnected: the container's output must + // never wait on a consumer. // https://github.com/containerd/nerdctl/issues/5137 - defaultQueueDepth = 256 + // + // The bound is bytes rather than frames. A TTY echoes single keystrokes, so + // counting frames would evict a session that is a few kilobytes behind, + // while a frame can be up to maxPayload, so counting frames also puts no + // real ceiling on memory in the logging process. + maxQueuedBytes = 4 << 20 + // defaultQueueDepth caps the channel itself, as a backstop against a flood + // of tiny frames. maxQueuedBytes is what normally decides. + defaultQueueDepth = 4096 // writeTimeout bounds a single write to a session, so that a session that is // connected but no longer draining its socket is eventually dropped instead // of leaking a goroutine and its queue for the container's lifetime. @@ -91,6 +99,8 @@ type session struct { // exit is the final frame writeLoop delivers after frames has drained. See // closeWith. exit []byte + // queued is how many bytes are sitting in frames, waiting to be written. + queued int } // send queues frame for the session. It reports false only when the session's @@ -102,14 +112,25 @@ func (s *session) send(frame []byte) bool { if s.closed { return true } + if s.queued+len(frame) > maxQueuedBytes { + return false + } select { case s.frames <- frame: + s.queued += len(frame) return true default: return false } } +// dequeued records that n bytes have left the queue. +func (s *session) dequeued(n int) { + s.mu.Lock() + s.queued -= n + s.mu.Unlock() +} + // closeWith closes the session's queue, leaving exit for writeLoop to deliver // once everything already queued has gone out. exit is nil when the session is // being dropped rather than told the container is gone. @@ -203,9 +224,19 @@ func (b *Broker) writeFrame(stream byte, p []byte) { } b.mu.Unlock() + if len(slow) == 0 { + return + } + // Tell them why. Otherwise being evicted is indistinguishable from the + // broker dying, and the client reports both as a lost connection. + dropped, err := EncodeControl(Control{Type: ControlDropped}) + if err != nil { + log.L.WithError(err).Warn("attachmux: failed to encode the dropped notice") + dropped = nil + } for _, s := range slow { log.L.Warn("attachmux: an attach session stopped keeping up, disconnecting it") - s.close() + s.closeWith(dropped) } } @@ -377,7 +408,9 @@ func (b *Broker) writeLoop(s *session) { b.dropSession(s) return } - if _, err := s.conn.Write(frame); err != nil { + _, err := s.conn.Write(frame) + s.dequeued(len(frame)) + if err != nil { b.dropSession(s) return } diff --git a/pkg/attachmux/broker_test.go b/pkg/attachmux/broker_test.go index 2881bb40108..65e9982324c 100644 --- a/pkg/attachmux/broker_test.go +++ b/pkg/attachmux/broker_test.go @@ -626,3 +626,53 @@ func TestBrokerServeAcceptsSessions(t *testing.T) { t.Fatal("Serve did not return after the context was cancelled") } } + +func TestBrokerTellsAnEvictedSessionWhy(t *testing.T) { + t.Parallel() + + // Being dropped for falling behind has to be distinguishable from the + // broker dying: both end the session, but only one is the user's fault and + // only one means the container may still be running. + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + + // Never read: overflow the byte bound. + chunk := bytes.Repeat([]byte("x"), 64<<10) + for b.SessionCount() > 0 { + b.Write(StreamStdout, chunk) + } + + // Everything already queued arrives first, then the notice. + for { + stream, payload, err := ReadFrame(conn) + assert.NilError(t, err) + if stream != StreamControl { + continue + } + var c Control + assert.NilError(t, json.Unmarshal(payload, &c)) + if c.Type == ControlHello { + continue + } + assert.Equal(t, c.Type, ControlDropped) + return + } +} + +func TestBrokerBoundsAQueueByBytesNotFrames(t *testing.T) { + t.Parallel() + + // A TTY echoes single keystrokes. Counting frames would evict a session + // that is only a couple of kilobytes behind. + b := NewBroker(true, nil) + conn := connect(t, b) + waitSessions(t, b, 1) + _ = conn + + // Far more frames than the old 256-frame bound, but tiny ones. + for range 2000 { + b.Write(StreamStdout, []byte("x")) + } + assert.Equal(t, b.SessionCount(), 1) +} diff --git a/pkg/attachmux/client.go b/pkg/attachmux/client.go index e883b34873f..01e5f01c555 100644 --- a/pkg/attachmux/client.go +++ b/pkg/attachmux/client.go @@ -64,8 +64,13 @@ func NewSession(conn net.Conn) (*Session, error) { if hello.Type != ControlHello { return nil, fmt.Errorf("attachmux: expected a hello, got %q", hello.Type) } - if hello.Version != ProtocolVersion { - return nil, fmt.Errorf("attachmux: unsupported protocol version %d, expected %d", hello.Version, ProtocolVersion) + // Only a broker newer than this client is refused. A broker lives as long + // as its container, so after a nerdctl upgrade a new client routinely meets + // an older one, and there is no falling back: the container's stdio is + // already bound to that process. A client can read a framing older than its + // own; it cannot read a newer one. + if hello.Version > ProtocolVersion { + return nil, fmt.Errorf("attachmux: the container's attach socket speaks protocol version %d, this nerdctl understands up to %d", hello.Version, ProtocolVersion) } return &Session{conn: conn, hello: hello}, nil } @@ -182,11 +187,17 @@ func (s *Session) Stream(ctx context.Context, stdin io.Reader, stdout, stderr io if err := json.Unmarshal(payload, &c); err != nil { continue } - if c.Type == ControlExit { + switch c.Type { + case ControlExit: s.exitMu.Lock() s.exited = true s.exitMu.Unlock() return nil + case ControlDropped: + // The broker evicted this session for falling behind. Saying so + // is the whole point of the frame: otherwise this is + // indistinguishable from the broker dying. + return errors.New("attachmux: this session was disconnected because it could not keep up with the container's output") } } } diff --git a/pkg/attachmux/client_test.go b/pkg/attachmux/client_test.go index c2ccfe2fa12..e73a56781d1 100644 --- a/pkg/attachmux/client_test.go +++ b/pkg/attachmux/client_test.go @@ -263,3 +263,43 @@ func TestSessionStreamTreatsAnExplicitCloseAsClean(t *testing.T) { t.Fatal("Stream did not return after Close") } } + +func TestSessionAcceptsAnOlderBroker(t *testing.T) { + t.Parallel() + + // A broker lives as long as its container, so after an upgrade a new client + // routinely meets an older one, and it has nowhere to fall back to: the + // container's stdio is already bound to that process. + mine, theirs := net.Pipe() + t.Cleanup(func() { mine.Close() }) + + go func() { + frame, err := EncodeControl(Control{Type: ControlHello, Version: ProtocolVersion - 1, TTY: true}) + if err != nil { + return + } + theirs.Write(frame) + }() + + session, err := NewSession(mine) + assert.NilError(t, err) + assert.Equal(t, session.TTY(), true) +} + +func TestSessionRefusesANewerBroker(t *testing.T) { + t.Parallel() + + mine, theirs := net.Pipe() + t.Cleanup(func() { mine.Close() }) + + go func() { + frame, err := EncodeControl(Control{Type: ControlHello, Version: ProtocolVersion + 1}) + if err != nil { + return + } + theirs.Write(frame) + }() + + _, err := NewSession(mine) + assert.ErrorContains(t, err, "understands up to") +} diff --git a/pkg/attachmux/proto.go b/pkg/attachmux/proto.go index 41c1a69fca2..e582fd18d78 100644 --- a/pkg/attachmux/proto.go +++ b/pkg/attachmux/proto.go @@ -45,10 +45,19 @@ const ( // carries no exit code: a session reads that from containerd, so that there // is a single source of truth for it. ControlExit = "exit" + // ControlDropped is sent to a session that fell too far behind, right + // before it is disconnected. Without it a client cannot tell being evicted + // from the broker dying, and reports both as a lost connection. + ControlDropped = "dropped" ) -// ProtocolVersion is announced in the hello message so that a future change to -// the framing can be detected by an older client. +// ProtocolVersion is announced in the hello message so that a change to the +// framing can be detected. +// +// The compatibility rule is one-sided: a client refuses a broker whose version +// is higher than its own, and accepts anything lower. A broker lives as long as +// its container, so a client that is newer than the broker is the normal case +// after an upgrade, and it has nowhere to fall back to. const ProtocolVersion = 1 const ( diff --git a/pkg/attachmux/socket_supported.go b/pkg/attachmux/socket_supported.go index e74452ccfcd..1828fe4c056 100644 --- a/pkg/attachmux/socket_supported.go +++ b/pkg/attachmux/socket_supported.go @@ -100,7 +100,55 @@ func Listen(path string) (net.Listener, error) { l.Close() return nil, err } - return l, nil + + // Go removes the socket on Close by path, without checking that the file is + // still the one it created. Two brokers can briefly overlap for a single + // container: the restart monitor creates the new task, and with it a new + // logging process, while the old one is still finishing its log driver. The + // new broker taking the path over is correct, since it is the one attached + // to the live task, but the old one exiting afterwards would then unlink + // the live socket and leave the new broker listening on an inode nobody can + // reach. So the removal is done here instead, against the inode. + ul, ok := l.(*net.UnixListener) + if !ok { + return l, nil + } + ul.SetUnlinkOnClose(false) + ino, err := inodeOf(path) + if err != nil { + l.Close() + return nil, err + } + return &ownedListener{UnixListener: ul, path: path, ino: ino}, nil +} + +// ownedListener removes its socket file on Close, but only while that file is +// still the one Listen created. +type ownedListener struct { + *net.UnixListener + path string + ino uint64 +} + +func (l *ownedListener) Close() error { + err := l.UnixListener.Close() + if ino, serr := inodeOf(l.path); serr == nil && ino == l.ino { + os.Remove(l.path) + } + return err +} + +func inodeOf(path string) (uint64, error) { + fi, err := os.Stat(path) + if err != nil { + return 0, err + } + st, ok := fi.Sys().(*syscall.Stat_t) + if !ok { + return 0, fmt.Errorf("attachmux: cannot read the inode of %s", path) + } + // Ino is uint64 on both platforms this file is built for. + return st.Ino, nil } const ( diff --git a/pkg/attachmux/socket_test.go b/pkg/attachmux/socket_test.go index 74f2f0b6f01..4b8d2b0225b 100644 --- a/pkg/attachmux/socket_test.go +++ b/pkg/attachmux/socket_test.go @@ -120,3 +120,30 @@ func TestDialToNothing(t *testing.T) { _, err := Dial(ctx, filepath.Join(t.TempDir(), "absent.sock")) assert.Assert(t, err != nil) } + +func TestListenRemovesOnlyItsOwnSocket(t *testing.T) { + t.Parallel() + + // Two brokers can briefly overlap for one container while the restart + // monitor swaps tasks. The one that exits second must not unlink the live + // socket of the one that took the path over. + path := filepath.Join(t.TempDir(), "attach.sock") + + first, err := Listen(path) + assert.NilError(t, err) + + second, err := Listen(path) + assert.NilError(t, err) + t.Cleanup(func() { second.Close() }) + + // The old broker exits last. + assert.NilError(t, first.Close()) + + _, err = os.Stat(path) + assert.NilError(t, err, "the surviving broker lost its socket") + + // And a broker that is alone does clean up after itself. + assert.NilError(t, second.Close()) + _, err = os.Stat(path) + assert.Assert(t, os.IsNotExist(err), "the socket outlived its broker") +} From f17c0d1296969b6a24bfdc720ce48f2d507abdb8 Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Tue, 1 Sep 2026 21:16:53 +0300 Subject: [PATCH 7/8] fix(attachmux): compare the socket directory with a joined path The assertion spelled the separator out, so it failed on windows once the test moved out from behind the build tag. That move was the point: these cases never needed a socket, and windows is where the package is only the unsupported stubs. Signed-off-by: Eugene Kalinin --- pkg/attachmux/attachmux_test.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pkg/attachmux/attachmux_test.go b/pkg/attachmux/attachmux_test.go index 672c9afbee3..8eb1da55339 100644 --- a/pkg/attachmux/attachmux_test.go +++ b/pkg/attachmux/attachmux_test.go @@ -77,7 +77,9 @@ func TestSocketPathIsUnderTheDataStore(t *testing.T) { // which is the only path it and its clients are guaranteed to agree on: the // shim spawns it with an environment of just CONTAINER_ID and // CONTAINER_NAMESPACE, so nothing XDG-based is available there. - path := SocketPath("/var/lib/nerdctl/1935db59", "default", "abc") - assert.Equal(t, filepath.Dir(path), "/var/lib/nerdctl/1935db59/attach") + const dataStore = "/var/lib/nerdctl/1935db59" + path := SocketPath(dataStore, "default", "abc") + // Joined, not spelled out: the separator is the platform's. + assert.Equal(t, filepath.Dir(path), filepath.Join(dataStore, "attach")) assert.Equal(t, len(filepath.Base(path)), socketNameLen+len(".sock")) } From 310a28ed50d9d6c995693cb6864b6c29b8454439 Mon Sep 17 00:00:00 2001 From: Eugene Kalinin Date: Wed, 2 Sep 2026 22:26:30 +0300 Subject: [PATCH 8/8] fix(attachmux): let the caller choose whether Dial waits for the broker Dial guessed at the situation from the error it got, and guessed wrong in both directions. A refused connection gave up after a fixed grace period counted from the first attempt, so a container being restarted over a socket left by a killed broker fell back to the legacy path before the new broker had bound. A missing socket, meanwhile, was retried for the full five seconds, so attaching to a container that never had a broker stalled the CLI for five seconds before it could fall back. Which of those is a race and which is proof depends on the caller, not on the errno. DialStarting is for a caller that has just created the task, where containerd has not waited for the logging process it spawned and nothing is conclusive yet; Dial is for reaching a container that has been running, where a socket that is missing or refuses says there is no broker. Two more from the same review pass: - writeFrame checked for sessions only after encoding the frame, so every container with nobody attached paid for a copy of each chunk of its output, which is most containers on a host. - Listen took unlinking away from Go before it had the inode, so a failure to stat left the socket on disk with nobody to remove it. Signed-off-by: Eugene Kalinin --- pkg/attachmux/broker.go | 11 +++ pkg/attachmux/socket_supported.go | 103 ++++++++++++++++------------ pkg/attachmux/socket_test.go | 41 +++++++++-- pkg/attachmux/socket_unsupported.go | 3 + 4 files changed, 109 insertions(+), 49 deletions(-) diff --git a/pkg/attachmux/broker.go b/pkg/attachmux/broker.go index 327d0d57bf0..0d4b3d6e7ec 100644 --- a/pkg/attachmux/broker.go +++ b/pkg/attachmux/broker.go @@ -200,6 +200,16 @@ func (b *Broker) writeFrame(stream byte, p []byte) { return } + // Most containers on a host have nobody attached, and their output still + // comes through here. Checking first means they do not pay for a copy of + // every chunk that is then thrown away. + b.mu.Lock() + empty := b.closed || len(b.sessions) == 0 + b.mu.Unlock() + if empty { + return + } + // Encoded outside the lock: it allocates and copies the whole chunk, and // readLoop, addSession and SessionCount all contend for b.mu. frame, err := EncodeFrame(stream, p) @@ -208,6 +218,7 @@ func (b *Broker) writeFrame(stream byte, p []byte) { return } + // Checked again: a session can have gone away while the frame was encoded. b.mu.Lock() if b.closed || len(b.sessions) == 0 { b.mu.Unlock() diff --git a/pkg/attachmux/socket_supported.go b/pkg/attachmux/socket_supported.go index 1828fe4c056..ad03dac166e 100644 --- a/pkg/attachmux/socket_supported.go +++ b/pkg/attachmux/socket_supported.go @@ -113,12 +113,15 @@ func Listen(path string) (net.Listener, error) { if !ok { return l, nil } - ul.SetUnlinkOnClose(false) ino, err := inodeOf(path) if err != nil { + // Still Go's to unlink at this point, so closing cleans up. l.Close() return nil, err } + // Taken over only once the inode is known, so that no failure above leaves + // the socket on disk with nobody to remove it. + ul.SetUnlinkOnClose(false) return &ownedListener{UnixListener: ul, path: path, ino: ino}, nil } @@ -152,62 +155,46 @@ func inodeOf(path string) (uint64, error) { } const ( - // dialTimeout bounds how long Dial waits for the broker to come up. - dialTimeout = 5 * time.Second - // dialInterval is how often Dial retries while waiting. + // dialTimeout bounds how long DialStarting waits for the broker to come up. + dialTimeout = 5 * time.Second dialInterval = 20 * time.Millisecond - // refusedGrace bounds how long Dial keeps retrying after a connection was - // refused, which unlike a missing socket means nothing is listening. - refusedGrace = 200 * time.Millisecond ) // Dial connects to the attach socket at path and completes the handshake. // -// It retries until the socket answers, the context is done, or dialTimeout -// elapses. containerd spawns the logging process while creating the task and -// does not wait for it, so a client that dials straight after -// container.NewTask can arrive before the broker has bound its socket. For a -// container that is already running the first attempt succeeds. +// It makes a single attempt. The caller is reaching a container that has been +// running for a while, so a socket that is missing or refuses the connection is +// proof that there is no broker to talk to, not a race worth waiting out. func Dial(ctx context.Context, path string) (*Session, error) { - ctx, cancel := context.WithTimeout(ctx, dialTimeout) - defer cancel() + return dial(ctx, path, false) +} + +// DialStarting is Dial for a caller that has just created the task. +// +// containerd spawns the logging process while creating the task and does not +// wait for it, so for a short while neither a missing socket nor a refused +// connection says anything: the broker may still be binding, possibly over a +// file left behind by a previous one. This retries until the socket answers, +// the context is done, or dialTimeout elapses. +func DialStarting(ctx context.Context, path string) (*Session, error) { + return dial(ctx, path, true) +} + +func dial(ctx context.Context, path string, retry bool) (*Session, error) { + if retry { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, dialTimeout) + defer cancel() + } - refusedDeadline := time.Now().Add(refusedGrace) var d net.Dialer for { conn, err := d.DialContext(ctx, "unix", path) if err == nil { - // A connection can sit in the listener's backlog before the broker - // accepts it, so the handshake read needs a deadline of its own: - // a broker that died between Listen and Serve would otherwise hang - // the client for good. - if deadline, ok := ctx.Deadline(); ok { - if err := conn.SetReadDeadline(deadline); err != nil { - conn.Close() - return nil, err - } - } - session, err := NewSession(conn) - if err != nil { - conn.Close() - return nil, err - } - // Streaming has no deadline of its own. - if err := conn.SetReadDeadline(time.Time{}); err != nil { - session.Close() - return nil, err - } - return session, nil + return greet(ctx, conn) } - if errors.Is(err, syscall.ECONNREFUSED) { - // The socket file is there but nothing is listening. A broker that - // was killed leaves the file behind, because Go only unlinks it on a - // graceful Close, so this is proof rather than a race worth waiting - // out. Retry only long enough to cover Listen's own window between - // os.Remove and bind. - if time.Now().After(refusedDeadline) { - return nil, fmt.Errorf("failed to connect to %s: %w", path, err) - } + if !retry { + return nil, fmt.Errorf("failed to connect to %s: %w", path, err) } select { case <-ctx.Done(): @@ -217,6 +204,32 @@ func Dial(ctx context.Context, path string) (*Session, error) { } } +// greet completes the handshake on an accepted connection. +func greet(ctx context.Context, conn net.Conn) (*Session, error) { + // A connection can sit in the listener's backlog before the broker accepts + // it, so the handshake read needs a deadline of its own: a broker that died + // between Listen and Serve would otherwise hang the client for good. + deadline, ok := ctx.Deadline() + if !ok { + deadline = time.Now().Add(dialTimeout) + } + if err := conn.SetReadDeadline(deadline); err != nil { + conn.Close() + return nil, err + } + session, err := NewSession(conn) + if err != nil { + conn.Close() + return nil, err + } + // Streaming has no deadline of its own. + if err := conn.SetReadDeadline(time.Time{}); err != nil { + session.Close() + return nil, err + } + return session, nil +} + // RemoveSocket deletes an attach socket. Removing one that is not there is not // an error. func RemoveSocket(path string) error { diff --git a/pkg/attachmux/socket_test.go b/pkg/attachmux/socket_test.go index 4b8d2b0225b..5a8929f34e0 100644 --- a/pkg/attachmux/socket_test.go +++ b/pkg/attachmux/socket_test.go @@ -109,15 +109,48 @@ func TestProbeIsConcurrencySafe(t *testing.T) { } } -func TestDialToNothing(t *testing.T) { +func TestDialToNothingFailsAtOnce(t *testing.T) { + t.Parallel() + + // Attaching to a container that has been running for a while must not stall + // the CLI: a missing socket is proof there is no broker, not a race. + start := time.Now() + _, err := Dial(context.Background(), filepath.Join(t.TempDir(), "absent.sock")) + assert.Assert(t, err != nil) + assert.Assert(t, time.Since(start) < time.Second, "Dial waited %s", time.Since(start)) +} + +func TestDialStartingWaitsForTheBroker(t *testing.T) { + t.Parallel() + + // The caller has just created the task, and containerd does not wait for + // the logging process it spawns, so the socket may not be there yet. + path := filepath.Join(t.TempDir(), "attach.sock") + b := NewBroker(true, nil) + + go func() { + time.Sleep(200 * time.Millisecond) + l, err := Listen(path) + if err != nil { + return + } + go b.Serve(context.Background(), l) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + session, err := DialStarting(ctx, path) + assert.NilError(t, err) + t.Cleanup(func() { session.Close() }) +} + +func TestDialStartingGivesUpEventually(t *testing.T) { t.Parallel() - // Dial retries while waiting for the broker to bind, so it gives up only - // when the context is done. Keep that short here. ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) t.Cleanup(cancel) - _, err := Dial(ctx, filepath.Join(t.TempDir(), "absent.sock")) + _, err := DialStarting(ctx, filepath.Join(t.TempDir(), "absent.sock")) assert.Assert(t, err != nil) } diff --git a/pkg/attachmux/socket_unsupported.go b/pkg/attachmux/socket_unsupported.go index ab335d23c79..2f3189cd707 100644 --- a/pkg/attachmux/socket_unsupported.go +++ b/pkg/attachmux/socket_unsupported.go @@ -32,5 +32,8 @@ func Listen(string) (net.Listener, error) { return nil, ErrUnsupported } // Dial is not implemented on this platform. func Dial(context.Context, string) (*Session, error) { return nil, ErrUnsupported } +// DialStarting is not implemented on this platform. +func DialStarting(context.Context, string) (*Session, error) { return nil, ErrUnsupported } + // RemoveSocket is not implemented on this platform. func RemoveSocket(string) error { return nil }