From ea1a8e24320eb9cd12b73d3ea7446e35d6cd23a5 Mon Sep 17 00:00:00 2001 From: Matthew Heon Date: Thu, 26 May 2022 14:33:48 -0400 Subject: Move Attach under the OCI Runtime interface With conmon-rs on the horizon, we need to disentangle Libpod from legacy Conmon to the greatest extent possible. There are definitely opportunities for codesharing between the two, but we have to assume the implementations will be largely disjoint given the different architectures. Fortunately, most of the work has already been done in the past. The conmon-managed OCI runtime mostly sits behind an interface, with a few exceptions - the most notable of those being attach. This PR thus moves Attach behind the interface, to ensure that we can have attach implementations that don't use our existing unix socket streaming if necessary. Still to-do is conmon cleanup. There's a lot of code that removes Conmon-specific files, or kills the Conmon PID, and all of it will need to be refactored behind the interface. [NO NEW TESTS NEEDED] Just moving some things around. Signed-off-by: Matthew Heon --- libpod/container_api.go | 25 ++- libpod/oci.go | 30 +++- libpod/oci_attach_linux.go | 300 ------------------------------------ libpod/oci_conmon_attach_linux.go | 314 ++++++++++++++++++++++++++++++++++++++ libpod/oci_missing.go | 5 + 5 files changed, 369 insertions(+), 305 deletions(-) delete mode 100644 libpod/oci_attach_linux.go create mode 100644 libpod/oci_conmon_attach_linux.go diff --git a/libpod/container_api.go b/libpod/container_api.go index a6fcf709d..930b3e17b 100644 --- a/libpod/container_api.go +++ b/libpod/container_api.go @@ -123,7 +123,18 @@ func (c *Container) StartAndAttach(ctx context.Context, streams *define.AttachSt // Attach to the container before starting it go func() { - if err := c.attach(streams, keys, resize, true, startedChan, nil); err != nil { + // Start resizing + if c.LogDriver() != define.PassthroughLogging { + registerResizeFunc(resize, c.bundlePath()) + } + + opts := new(AttachOptions) + opts.Streams = streams + opts.DetachKeys = &keys + opts.Start = true + opts.Started = startedChan + + if err := c.ociRuntime.Attach(c, opts); err != nil { attachChan <- err } close(attachChan) @@ -261,8 +272,18 @@ func (c *Container) Attach(streams *define.AttachStreams, keys string, resize <- }() } + // Start resizing + if c.LogDriver() != define.PassthroughLogging { + registerResizeFunc(resize, c.bundlePath()) + } + + opts := new(AttachOptions) + opts.Streams = streams + opts.DetachKeys = &keys + opts.AttachReady = attachRdy + c.newContainerEvent(events.Attach) - return c.attach(streams, keys, resize, false, nil, attachRdy) + return c.ociRuntime.Attach(c, opts) } // HTTPAttach forwards an attach session over a hijacked HTTP session. diff --git a/libpod/oci.go b/libpod/oci.go index 09f856ac7..90862969c 100644 --- a/libpod/oci.go +++ b/libpod/oci.go @@ -12,9 +12,7 @@ import ( // management logic - e.g., we do not expect it to determine on its own that // calling 'UnpauseContainer()' on a container that is not paused is an error. // The code calling the OCIRuntime will manage this. -// TODO: May want to move the Attach() code under this umbrella. It's highly OCI -// runtime dependent. -// TODO: May want to move the conmon cleanup code here too - it depends on +// TODO: May want to move the conmon cleanup code here - it depends on // Conmon being in use. type OCIRuntime interface { // Name returns the name of the runtime. @@ -52,6 +50,8 @@ type OCIRuntime interface { // UnpauseContainer unpauses the given container. UnpauseContainer(ctr *Container) error + // Attach to a container. + Attach(ctr *Container, params *AttachOptions) error // HTTPAttach performs an attach intended to be transported over HTTP. // For terminal attach, the container's output will be directly streamed // to output; otherwise, STDOUT and STDERR will be multiplexed, with @@ -149,6 +149,30 @@ type OCIRuntime interface { RuntimeInfo() (*define.ConmonInfo, *define.OCIRuntimeInfo, error) } +// AttachOptions are options used when attached to a container or an exec +// session. +type AttachOptions struct { + // Streams are the streams to attach to. + Streams *define.AttachStreams + // DetachKeys containers the key combination that will detach from the + // attach session. Empty string is assumed as no detach keys - user + // detach is impossible. If unset, defaults from containers.conf will be + // used. + DetachKeys *string + // InitialSize is the initial size of the terminal. Set before the + // attach begins. + InitialSize *define.TerminalSize + // AttachReady signals when the attach has successfully completed and + // streaming has begun. + AttachReady chan<- bool + // Start indicates that the container should be started if it is not + // already running. + Start bool + // Started signals when the container has been successfully started. + // Required if Start is true, unused otherwise. + Started chan<- bool +} + // ExecOptions are options passed into ExecContainer. They control the command // that will be executed and how the exec will proceed. type ExecOptions struct { diff --git a/libpod/oci_attach_linux.go b/libpod/oci_attach_linux.go deleted file mode 100644 index 06f8f8719..000000000 --- a/libpod/oci_attach_linux.go +++ /dev/null @@ -1,300 +0,0 @@ -//go:build linux -// +build linux - -package libpod - -import ( - "fmt" - "io" - "net" - "os" - "path/filepath" - "syscall" - - "github.com/containers/common/pkg/config" - "github.com/containers/podman/v4/libpod/define" - "github.com/containers/podman/v4/pkg/errorhandling" - "github.com/containers/podman/v4/pkg/kubeutils" - "github.com/containers/podman/v4/utils" - "github.com/moby/term" - "github.com/pkg/errors" - "github.com/sirupsen/logrus" - "golang.org/x/sys/unix" -) - -/* Sync with stdpipe_t in conmon.c */ -const ( - AttachPipeStdin = 1 - AttachPipeStdout = 2 - AttachPipeStderr = 3 -) - -func openUnixSocket(path string) (*net.UnixConn, error) { - fd, err := unix.Open(path, unix.O_PATH, 0) - if err != nil { - return nil, err - } - defer unix.Close(fd) - return net.DialUnix("unixpacket", nil, &net.UnixAddr{Name: fmt.Sprintf("/proc/self/fd/%d", fd), Net: "unixpacket"}) -} - -// Attach to the given container -// Does not check if state is appropriate -// started is only required if startContainer is true -func (c *Container) attach(streams *define.AttachStreams, keys string, resize <-chan define.TerminalSize, startContainer bool, started chan bool, attachRdy chan<- bool) error { - passthrough := c.LogDriver() == define.PassthroughLogging - - if !streams.AttachOutput && !streams.AttachError && !streams.AttachInput && !passthrough { - return errors.Wrapf(define.ErrInvalidArg, "must provide at least one stream to attach to") - } - if startContainer && started == nil { - return errors.Wrapf(define.ErrInternal, "started chan not passed when startContainer set") - } - - detachKeys, err := processDetachKeys(keys) - if err != nil { - return err - } - - var conn *net.UnixConn - if !passthrough { - logrus.Debugf("Attaching to container %s", c.ID()) - - registerResizeFunc(resize, c.bundlePath()) - - attachSock, err := c.AttachSocketPath() - if err != nil { - return err - } - - conn, err = openUnixSocket(attachSock) - if err != nil { - return errors.Wrapf(err, "failed to connect to container's attach socket: %v", attachSock) - } - defer func() { - if err := conn.Close(); err != nil { - logrus.Errorf("unable to close socket: %q", err) - } - }() - } - - // If starting was requested, start the container and notify when that's - // done. - if startContainer { - if err := c.start(); err != nil { - return err - } - started <- true - } - - if passthrough { - return nil - } - - receiveStdoutError, stdinDone := setupStdioChannels(streams, conn, detachKeys) - if attachRdy != nil { - attachRdy <- true - } - return readStdio(conn, streams, receiveStdoutError, stdinDone) -} - -// Attach to the given container's exec session -// attachFd and startFd must be open file descriptors -// attachFd must be the output side of the fd. attachFd is used for two things: -// conmon will first send a nonce value across the pipe indicating it has set up its side of the console socket -// this ensures attachToExec gets all of the output of the called process -// conmon will then send the exit code of the exec process, or an error in the exec session -// startFd must be the input side of the fd. -// newSize resizes the tty to this size before the process is started, must be nil if the exec session has no tty -// conmon will wait to start the exec session until the parent process has setup the console socket. -// Once attachToExec successfully attaches to the console socket, the child conmon process responsible for calling runtime exec -// will read from the output side of start fd, thus learning to start the child process. -// Thus, the order goes as follow: -// 1. conmon parent process sets up its console socket. sends on attachFd -// 2. attachToExec attaches to the console socket after reading on attachFd and resizes the tty -// 3. child waits on startFd for attachToExec to attach to said console socket -// 4. attachToExec sends on startFd, signalling it has attached to the socket and child is ready to go -// 5. child receives on startFd, runs the runtime exec command -// attachToExec is responsible for closing startFd and attachFd -func (c *Container) attachToExec(streams *define.AttachStreams, keys *string, sessionID string, startFd, attachFd *os.File, newSize *define.TerminalSize) error { - if !streams.AttachOutput && !streams.AttachError && !streams.AttachInput { - return errors.Wrapf(define.ErrInvalidArg, "must provide at least one stream to attach to") - } - if startFd == nil || attachFd == nil { - return errors.Wrapf(define.ErrInvalidArg, "start sync pipe and attach sync pipe must be defined for exec attach") - } - - defer errorhandling.CloseQuiet(startFd) - defer errorhandling.CloseQuiet(attachFd) - - detachString := config.DefaultDetachKeys - if keys != nil { - detachString = *keys - } - detachKeys, err := processDetachKeys(detachString) - if err != nil { - return err - } - - logrus.Debugf("Attaching to container %s exec session %s", c.ID(), sessionID) - - // set up the socket path, such that it is the correct length and location for exec - sockPath, err := c.execAttachSocketPath(sessionID) - if err != nil { - return err - } - - // 2: read from attachFd that the parent process has set up the console socket - if _, err := readConmonPipeData(c.ociRuntime.Name(), attachFd, ""); err != nil { - return err - } - - // resize before we start the container process - if newSize != nil { - err = c.ociRuntime.ExecAttachResize(c, sessionID, *newSize) - if err != nil { - logrus.Warnf("Resize failed: %v", err) - } - } - - // 2: then attach - conn, err := openUnixSocket(sockPath) - if err != nil { - return errors.Wrapf(err, "failed to connect to container's attach socket: %v", sockPath) - } - defer func() { - if err := conn.Close(); err != nil { - logrus.Errorf("Unable to close socket: %q", err) - } - }() - - // start listening on stdio of the process - receiveStdoutError, stdinDone := setupStdioChannels(streams, conn, detachKeys) - - // 4: send start message to child - if err := writeConmonPipeData(startFd); err != nil { - return err - } - - return readStdio(conn, streams, receiveStdoutError, stdinDone) -} - -func processDetachKeys(keys string) ([]byte, error) { - // Check the validity of the provided keys first - if len(keys) == 0 { - return []byte{}, nil - } - detachKeys, err := term.ToBytes(keys) - if err != nil { - return nil, errors.Wrapf(err, "invalid detach keys") - } - return detachKeys, nil -} - -func registerResizeFunc(resize <-chan define.TerminalSize, bundlePath string) { - kubeutils.HandleResizing(resize, func(size define.TerminalSize) { - controlPath := filepath.Join(bundlePath, "ctl") - controlFile, err := os.OpenFile(controlPath, unix.O_WRONLY, 0) - if err != nil { - logrus.Debugf("Could not open ctl file: %v", err) - return - } - defer controlFile.Close() - - logrus.Debugf("Received a resize event: %+v", size) - if _, err = fmt.Fprintf(controlFile, "%d %d %d\n", 1, size.Height, size.Width); err != nil { - logrus.Warnf("Failed to write to control file to resize terminal: %v", err) - } - }) -} - -func setupStdioChannels(streams *define.AttachStreams, conn *net.UnixConn, detachKeys []byte) (chan error, chan error) { - receiveStdoutError := make(chan error) - go func() { - receiveStdoutError <- redirectResponseToOutputStreams(streams.OutputStream, streams.ErrorStream, streams.AttachOutput, streams.AttachError, conn) - }() - - stdinDone := make(chan error) - go func() { - var err error - if streams.AttachInput { - _, err = utils.CopyDetachable(conn, streams.InputStream, detachKeys) - } - stdinDone <- err - }() - - return receiveStdoutError, stdinDone -} - -func redirectResponseToOutputStreams(outputStream, errorStream io.Writer, writeOutput, writeError bool, conn io.Reader) error { - var err error - buf := make([]byte, 8192+1) /* Sync with conmon STDIO_BUF_SIZE */ - for { - nr, er := conn.Read(buf) - if nr > 0 { - var dst io.Writer - var doWrite bool - switch buf[0] { - case AttachPipeStdout: - dst = outputStream - doWrite = writeOutput - case AttachPipeStderr: - dst = errorStream - doWrite = writeError - default: - logrus.Infof("Received unexpected attach type %+d", buf[0]) - } - if dst == nil { - return errors.New("output destination cannot be nil") - } - - if doWrite { - nw, ew := dst.Write(buf[1:nr]) - if ew != nil { - err = ew - break - } - if nr != nw+1 { - err = io.ErrShortWrite - break - } - } - } - if errors.Is(er, io.EOF) || errors.Is(er, syscall.ECONNRESET) { - break - } - if er != nil { - err = er - break - } - } - return err -} - -func readStdio(conn *net.UnixConn, streams *define.AttachStreams, receiveStdoutError, stdinDone chan error) error { - var err error - select { - case err = <-receiveStdoutError: - if err := conn.CloseWrite(); err != nil { - logrus.Errorf("Failed to close stdin: %v", err) - } - return err - case err = <-stdinDone: - if err == define.ErrDetach { - if err := conn.CloseWrite(); err != nil { - logrus.Errorf("Failed to close stdin: %v", err) - } - return err - } - if err == nil { - // copy stdin is done, close it - if connErr := conn.CloseWrite(); connErr != nil { - logrus.Errorf("Unable to close conn: %v", connErr) - } - } - if streams.AttachOutput || streams.AttachError { - return <-receiveStdoutError - } - } - return nil -} diff --git a/libpod/oci_conmon_attach_linux.go b/libpod/oci_conmon_attach_linux.go new file mode 100644 index 000000000..155a8fbc3 --- /dev/null +++ b/libpod/oci_conmon_attach_linux.go @@ -0,0 +1,314 @@ +//go:build linux +// +build linux + +package libpod + +import ( + "fmt" + "io" + "net" + "os" + "path/filepath" + "syscall" + + "github.com/containers/common/pkg/config" + "github.com/containers/podman/v4/libpod/define" + "github.com/containers/podman/v4/pkg/errorhandling" + "github.com/containers/podman/v4/pkg/kubeutils" + "github.com/containers/podman/v4/utils" + "github.com/moby/term" + "github.com/pkg/errors" + "github.com/sirupsen/logrus" + "golang.org/x/sys/unix" +) + +/* Sync with stdpipe_t in conmon.c */ +const ( + AttachPipeStdin = 1 + AttachPipeStdout = 2 + AttachPipeStderr = 3 +) + +func openUnixSocket(path string) (*net.UnixConn, error) { + fd, err := unix.Open(path, unix.O_PATH, 0) + if err != nil { + return nil, err + } + defer unix.Close(fd) + return net.DialUnix("unixpacket", nil, &net.UnixAddr{Name: fmt.Sprintf("/proc/self/fd/%d", fd), Net: "unixpacket"}) +} + +// Attach to the given container. +// Does not check if state is appropriate. +// started is only required if startContainer is true. +func (r *ConmonOCIRuntime) Attach(c *Container, params *AttachOptions) error { + passthrough := c.LogDriver() == define.PassthroughLogging + + if params == nil || params.Streams == nil { + return errors.Wrapf(define.ErrInternal, "must provide parameters to Attach") + } + + if !params.Streams.AttachOutput && !params.Streams.AttachError && !params.Streams.AttachInput && !passthrough { + return errors.Wrapf(define.ErrInvalidArg, "must provide at least one stream to attach to") + } + if params.Start && params.Started == nil { + return errors.Wrapf(define.ErrInternal, "started chan not passed when startContainer set") + } + + keys := config.DefaultDetachKeys + if params.DetachKeys != nil { + keys = *params.DetachKeys + } + + detachKeys, err := processDetachKeys(keys) + if err != nil { + return err + } + + var conn *net.UnixConn + if !passthrough { + logrus.Debugf("Attaching to container %s", c.ID()) + + // If we have a resize, do it. + if params.InitialSize != nil { + if err := r.AttachResize(c, *params.InitialSize); err != nil { + return err + } + } + + attachSock, err := c.AttachSocketPath() + if err != nil { + return err + } + + conn, err = openUnixSocket(attachSock) + if err != nil { + return errors.Wrapf(err, "failed to connect to container's attach socket: %v", attachSock) + } + defer func() { + if err := conn.Close(); err != nil { + logrus.Errorf("unable to close socket: %q", err) + } + }() + } + + // If starting was requested, start the container and notify when that's + // done. + if params.Start { + if err := c.start(); err != nil { + return err + } + params.Started <- true + } + + if passthrough { + return nil + } + + receiveStdoutError, stdinDone := setupStdioChannels(params.Streams, conn, detachKeys) + if params.AttachReady != nil { + params.AttachReady <- true + } + return readStdio(conn, params.Streams, receiveStdoutError, stdinDone) +} + +// Attach to the given container's exec session +// attachFd and startFd must be open file descriptors +// attachFd must be the output side of the fd. attachFd is used for two things: +// conmon will first send a nonce value across the pipe indicating it has set up its side of the console socket +// this ensures attachToExec gets all of the output of the called process +// conmon will then send the exit code of the exec process, or an error in the exec session +// startFd must be the input side of the fd. +// newSize resizes the tty to this size before the process is started, must be nil if the exec session has no tty +// conmon will wait to start the exec session until the parent process has setup the console socket. +// Once attachToExec successfully attaches to the console socket, the child conmon process responsible for calling runtime exec +// will read from the output side of start fd, thus learning to start the child process. +// Thus, the order goes as follow: +// 1. conmon parent process sets up its console socket. sends on attachFd +// 2. attachToExec attaches to the console socket after reading on attachFd and resizes the tty +// 3. child waits on startFd for attachToExec to attach to said console socket +// 4. attachToExec sends on startFd, signalling it has attached to the socket and child is ready to go +// 5. child receives on startFd, runs the runtime exec command +// attachToExec is responsible for closing startFd and attachFd +func (c *Container) attachToExec(streams *define.AttachStreams, keys *string, sessionID string, startFd, attachFd *os.File, newSize *define.TerminalSize) error { + if !streams.AttachOutput && !streams.AttachError && !streams.AttachInput { + return errors.Wrapf(define.ErrInvalidArg, "must provide at least one stream to attach to") + } + if startFd == nil || attachFd == nil { + return errors.Wrapf(define.ErrInvalidArg, "start sync pipe and attach sync pipe must be defined for exec attach") + } + + defer errorhandling.CloseQuiet(startFd) + defer errorhandling.CloseQuiet(attachFd) + + detachString := config.DefaultDetachKeys + if keys != nil { + detachString = *keys + } + detachKeys, err := processDetachKeys(detachString) + if err != nil { + return err + } + + logrus.Debugf("Attaching to container %s exec session %s", c.ID(), sessionID) + + // set up the socket path, such that it is the correct length and location for exec + sockPath, err := c.execAttachSocketPath(sessionID) + if err != nil { + return err + } + + // 2: read from attachFd that the parent process has set up the console socket + if _, err := readConmonPipeData(c.ociRuntime.Name(), attachFd, ""); err != nil { + return err + } + + // resize before we start the container process + if newSize != nil { + err = c.ociRuntime.ExecAttachResize(c, sessionID, *newSize) + if err != nil { + logrus.Warnf("Resize failed: %v", err) + } + } + + // 2: then attach + conn, err := openUnixSocket(sockPath) + if err != nil { + return errors.Wrapf(err, "failed to connect to container's attach socket: %v", sockPath) + } + defer func() { + if err := conn.Close(); err != nil { + logrus.Errorf("Unable to close socket: %q", err) + } + }() + + // start listening on stdio of the process + receiveStdoutError, stdinDone := setupStdioChannels(streams, conn, detachKeys) + + // 4: send start message to child + if err := writeConmonPipeData(startFd); err != nil { + return err + } + + return readStdio(conn, streams, receiveStdoutError, stdinDone) +} + +func processDetachKeys(keys string) ([]byte, error) { + // Check the validity of the provided keys first + if len(keys) == 0 { + return []byte{}, nil + } + detachKeys, err := term.ToBytes(keys) + if err != nil { + return nil, errors.Wrapf(err, "invalid detach keys") + } + return detachKeys, nil +} + +func registerResizeFunc(resize <-chan define.TerminalSize, bundlePath string) { + kubeutils.HandleResizing(resize, func(size define.TerminalSize) { + controlPath := filepath.Join(bundlePath, "ctl") + controlFile, err := os.OpenFile(controlPath, unix.O_WRONLY, 0) + if err != nil { + logrus.Debugf("Could not open ctl file: %v", err) + return + } + defer controlFile.Close() + + logrus.Debugf("Received a resize event: %+v", size) + if _, err = fmt.Fprintf(controlFile, "%d %d %d\n", 1, size.Height, size.Width); err != nil { + logrus.Warnf("Failed to write to control file to resize terminal: %v", err) + } + }) +} + +func setupStdioChannels(streams *define.AttachStreams, conn *net.UnixConn, detachKeys []byte) (chan error, chan error) { + receiveStdoutError := make(chan error) + go func() { + receiveStdoutError <- redirectResponseToOutputStreams(streams.OutputStream, streams.ErrorStream, streams.AttachOutput, streams.AttachError, conn) + }() + + stdinDone := make(chan error) + go func() { + var err error + if streams.AttachInput { + _, err = utils.CopyDetachable(conn, streams.InputStream, detachKeys) + } + stdinDone <- err + }() + + return receiveStdoutError, stdinDone +} + +func redirectResponseToOutputStreams(outputStream, errorStream io.Writer, writeOutput, writeError bool, conn io.Reader) error { + var err error + buf := make([]byte, 8192+1) /* Sync with conmon STDIO_BUF_SIZE */ + for { + nr, er := conn.Read(buf) + if nr > 0 { + var dst io.Writer + var doWrite bool + switch buf[0] { + case AttachPipeStdout: + dst = outputStream + doWrite = writeOutput + case AttachPipeStderr: + dst = errorStream + doWrite = writeError + default: + logrus.Infof("Received unexpected attach type %+d", buf[0]) + } + if dst == nil { + return errors.New("output destination cannot be nil") + } + + if doWrite { + nw, ew := dst.Write(buf[1:nr]) + if ew != nil { + err = ew + break + } + if nr != nw+1 { + err = io.ErrShortWrite + break + } + } + } + if errors.Is(er, io.EOF) || errors.Is(er, syscall.ECONNRESET) { + break + } + if er != nil { + err = er + break + } + } + return err +} + +func readStdio(conn *net.UnixConn, streams *define.AttachStreams, receiveStdoutError, stdinDone chan error) error { + var err error + select { + case err = <-receiveStdoutError: + if err := conn.CloseWrite(); err != nil { + logrus.Errorf("Failed to close stdin: %v", err) + } + return err + case err = <-stdinDone: + if err == define.ErrDetach { + if err := conn.CloseWrite(); err != nil { + logrus.Errorf("Failed to close stdin: %v", err) + } + return err + } + if err == nil { + // copy stdin is done, close it + if connErr := conn.CloseWrite(); connErr != nil { + logrus.Errorf("Unable to close conn: %v", connErr) + } + } + if streams.AttachOutput || streams.AttachError { + return <-receiveStdoutError + } + } + return nil +} diff --git a/libpod/oci_missing.go b/libpod/oci_missing.go index 86f54c02e..fd8160830 100644 --- a/libpod/oci_missing.go +++ b/libpod/oci_missing.go @@ -108,6 +108,11 @@ func (r *MissingRuntime) UnpauseContainer(ctr *Container) error { return r.printError() } +// Attach is not available as the runtime is missing +func (r *MissingRuntime) Attach(ctr *Container, params *AttachOptions) error { + return r.printError() +} + // HTTPAttach is not available as the runtime is missing func (r *MissingRuntime) HTTPAttach(ctr *Container, req *http.Request, w http.ResponseWriter, streams *HTTPAttachStreams, detachKeys *string, cancel <-chan bool, hijackDone chan<- bool, streamAttach, streamLogs bool) error { return r.printError() -- cgit v1.2.3-54-g00ecf