diff --git a/pkg/attachmux/attachmux.go b/pkg/attachmux/attachmux.go new file mode 100644 index 00000000000..9da356fad2e --- /dev/null +++ b/pkg/attachmux/attachmux.go @@ -0,0 +1,79 @@ +/* + 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" + "fmt" + "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. +// +// 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 +// 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/attachmux_test.go b/pkg/attachmux/attachmux_test.go new file mode 100644 index 00000000000..8eb1da55339 --- /dev/null +++ b/pkg/attachmux/attachmux_test.go @@ -0,0 +1,85 @@ +/* + 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. + 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")) +} diff --git a/pkg/attachmux/broker.go b/pkg/attachmux/broker.go new file mode 100644 index 00000000000..0d4b3d6e7ec --- /dev/null +++ b/pkg/attachmux/broker.go @@ -0,0 +1,495 @@ +/* + 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 ( + // 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 + // + // 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. + 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 + // 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 +// 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 + } + 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. +// +// 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 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) +} + +func (b *Broker) writeFrame(stream byte, p []byte) { + if len(p) == 0 { + 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) + if err != nil { + log.L.WithError(err).Warn("attachmux: dropping an output frame") + 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() + 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() + + 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.closeWith(dropped) + } +} + +// 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 { + done := make(chan struct{}) + defer close(done) + go func() { + select { + case <-ctx.Done(): + l.Close() + case <-done: + } + }() + + 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 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 +// 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 + } + _, err := s.conn.Write(frame) + s.dequeued(len(frame)) + if 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 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") + continue + } + } + } +} + +// 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..65e9982324c --- /dev/null +++ b/pkg/attachmux/broker_test.go @@ -0,0 +1,678 @@ +/* + 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" + "syscall" + "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 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() + + // 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()) +} + +// 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() + + // 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") + } +} + +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 new file mode 100644 index 00000000000..01e5f01c555 --- /dev/null +++ b/pkg/attachmux/client.go @@ -0,0 +1,229 @@ +/* + 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 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 +// 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) + } + // 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 +} + +// 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. +// 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() + 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 { + s.stopStdin = make(chan struct{}) + go s.pumpStdin(stdin) + } + + done := make(chan struct{}) + defer close(done) + go func() { + select { + case <-ctx.Done(): + s.conn.Close() + case <-done: + } + }() + + 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 || 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 + // 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 + } + 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") + } + } + } +} + +func (s *Session) pumpStdin(stdin io.Reader) { + buf := make([]byte, 32<<10) + 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 + } + 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..e73a56781d1 --- /dev/null +++ b/pkg/attachmux/client_test.go @@ -0,0 +1,305 @@ +/* + 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") + } +} + +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") + } +} + +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 new file mode 100644 index 00000000000..e582fd18d78 --- /dev/null +++ b/pkg/attachmux/proto.go @@ -0,0 +1,120 @@ +/* + 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 ( + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "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 + 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" + // 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 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 ( + 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]) +} diff --git a/pkg/attachmux/socket_supported.go b/pkg/attachmux/socket_supported.go new file mode 100644 index 00000000000..ad03dac166e --- /dev/null +++ b/pkg/attachmux/socket_supported.go @@ -0,0 +1,241 @@ +//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" + "syscall" + "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)) + 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") + + 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 + } + + // 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 + } + 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 +} + +// 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 ( + // dialTimeout bounds how long DialStarting waits for the broker to come up. + dialTimeout = 5 * time.Second + dialInterval = 20 * time.Millisecond +) + +// Dial connects to the attach socket at path and completes the handshake. +// +// 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) { + 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() + } + + var d net.Dialer + for { + conn, err := d.DialContext(ctx, "unix", path) + if err == nil { + return greet(ctx, conn) + } + if !retry { + 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) + case <-time.After(dialInterval): + } + } +} + +// 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 { + 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 new file mode 100644 index 00000000000..5a8929f34e0 --- /dev/null +++ b/pkg/attachmux/socket_test.go @@ -0,0 +1,182 @@ +//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 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 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() + + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + t.Cleanup(cancel) + + _, err := DialStarting(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") +} diff --git a/pkg/attachmux/socket_unsupported.go b/pkg/attachmux/socket_unsupported.go new file mode 100644 index 00000000000..2f3189cd707 --- /dev/null +++ b/pkg/attachmux/socket_unsupported.go @@ -0,0 +1,39 @@ +//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 } + +// 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 } 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")) +}