Merge pull request #1470 from aojea/gocat

Get rid of socat for port forwarding

.github/workflows/ci.yml

@@ -107,8 +107,7 @@ jobs:
           sudo apt-get install -y \
             btrfs-tools \
             libseccomp2 \
-            libseccomp-dev \
-            socat
+            libseccomp-dev
           make install.deps
         working-directory: ${{github.workspace}}/src/github.com/containerd/cri
 

README.md

@@ -77,11 +77,10 @@ specifications as appropriate.
 (Fedora, CentOS, RHEL). On releases of Ubuntu <=Trusty and Debian <=jessie a
 backport version of `libseccomp-dev` is required. See [travis.yml](.travis.yml) for an example on trusty.
 * **btrfs development library.** Required by containerd btrfs support. `btrfs-tools`(Ubuntu, Debian) / `btrfs-progs-devel`(Fedora, CentOS, RHEL)
-2. Install **`socat`** (required by portforward).
-3. Install **`pkg-config`** (required for linking with `libseccomp`).
-4. Install and setup a Go 1.13.10 development environment.
-5. Make a local clone of this repository.
-6. Install binary dependencies by running the following command from your cloned `cri/` project directory:
+2. Install **`pkg-config`** (required for linking with `libseccomp`).
+3. Install and setup a Go 1.13.10 development environment.
+4. Make a local clone of this repository.
+5. Install binary dependencies by running the following command from your cloned `cri/` project directory:
 ```bash
 # Note: install.deps installs the above mentioned runc, containerd, and CNI
 # binary dependencies. install.deps is only provided for general use and ease of

contrib/ansible/tasks/bootstrap_centos.yaml

@@ -9,5 +9,4 @@
     - btrfs-progs
     - libseccomp
     - util-linux
-    - socat
     - libselinux-python

contrib/ansible/tasks/bootstrap_ubuntu.yaml

@@ -9,5 +9,4 @@
       - apt-transport-https
       - btrfs-tools
       - libseccomp2
-      - socat
       - util-linux

pkg/server/sandbox_portforward_unix.go

@@ -19,28 +19,27 @@
 package server
 
 import (
-	"bytes"
 	"fmt"
 	"io"
-	"os/exec"
-	"strings"
+	"net"
+	"time"
 
 	"github.com/containerd/containerd/log"
 	"github.com/containernetworking/plugins/pkg/ns"
 	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
+
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
 )
 
-// portForward requires `socat` on the node. It uses netns to enter the sandbox namespace,
-// and run `socat` inside the namespace to forward stream for a specific port. The `socat`
-// command keeps running until it exits or client disconnect.
-func (c *criService) portForward(ctx context.Context, id string, port int32, stream io.ReadWriter) error {
+// portForward uses netns to enter the sandbox namespace, and forwards a stream inside the
+// the namespace to a specific port. It keeps forwarding until it exits or client disconnect.
+func (c *criService) portForward(ctx context.Context, id string, port int32, stream io.ReadWriteCloser) error {
 	s, err := c.sandboxStore.Get(id)
 	if err != nil {
 		return errors.Wrapf(err, "failed to find sandbox %q in store", id)
 	}
+
 	var netNSDo func(func(ns.NetNS) error) error
 	// netNSPath is the network namespace path for logging.
 	var netNSPath string
@@ -62,48 +61,64 @@ func (c *criService) portForward(ctx context.Context, id string, port int32, str
 		netNSPath = "host"
 	}
 
-	socat, err := exec.LookPath("socat")
-	if err != nil {
-		return errors.Wrap(err, "failed to find socat")
-	}
-
-	// Check https://linux.die.net/man/1/socat for meaning of the options.
-	args := []string{socat, "-", fmt.Sprintf("TCP4:localhost:%d", port)}
-
-	log.G(ctx).Infof("Executing port forwarding command %q in network namespace %q", strings.Join(args, " "), netNSPath)
+	log.G(ctx).Infof("Executing port forwarding in network namespace %q", netNSPath)
 	err = netNSDo(func(_ ns.NetNS) error {
-		cmd := exec.Command(args[0], args[1:]...)
-		cmd.Stdout = stream
-
-		stderr := new(bytes.Buffer)
-		cmd.Stderr = stderr
-
-		// If we use Stdin, command.Run() won't return until the goroutine that's copying
-		// from stream finishes. Unfortunately, if you have a client like telnet connected
-		// via port forwarding, as long as the user's telnet client is connected to the user's
-		// local listener that port forwarding sets up, the telnet session never exits. This
-		// means that even if socat has finished running, command.Run() won't ever return
-		// (because the client still has the connection and stream open).
-		//
-		// The work around is to use StdinPipe(), as Wait() (called by Run()) closes the pipe
-		// when the command (socat) exits.
-		in, err := cmd.StdinPipe()
+		defer stream.Close()
+		// TODO: hardcoded to tcp4 because localhost resolves to ::1 by default if the system has IPv6 enabled.
+		// Theoretically happy eyeballs will try IPv6 first and fallback to IPv4
+		// but resolving localhost doesn't seem to return and IPv4 address, thus failing the connection.
+		conn, err := net.Dial("tcp4", fmt.Sprintf("localhost:%d", port))
 		if err != nil {
-			return errors.Wrap(err, "failed to create stdin pipe")
+			return errors.Wrapf(err, "failed to dial %d", port)
 		}
+		defer conn.Close()
+
+		errCh := make(chan error, 2)
+		// Copy from the the namespace port connection to the client stream
 		go func() {
-			if _, err := io.Copy(in, stream); err != nil {
-				logrus.WithError(err).Errorf("Failed to copy port forward input for %q port %d", id, port)
-			}
-			in.Close()
-			logrus.Debugf("Finish copying port forward input for %q port %d", id, port)
+			log.G(ctx).Debugf("PortForward copying data from namespace %q port %d to the client stream", id, port)
+			_, err := io.Copy(stream, conn)
+			errCh <- err
 		}()
 
-		if err := cmd.Run(); err != nil {
-			return errors.Errorf("socat command returns error: %v, stderr: %q", err, stderr.String())
+		// Copy from the client stream to the namespace port connection
+		go func() {
+			log.G(ctx).Debugf("PortForward copying data from client stream to namespace %q port %d", id, port)
+			_, err := io.Copy(conn, stream)
+			errCh <- err
+		}()
+
+		// Wait until the first error is returned by one of the connections
+		// we use errFwd to store the result of the port forwarding operation
+		// if the context is cancelled close everything and return
+		var errFwd error
+		select {
+		case errFwd = <-errCh:
+			log.G(ctx).Debugf("PortForward stop forwarding in one direction in network namespace %q port %d: %v", id, port, errFwd)
+		case <-ctx.Done():
+			log.G(ctx).Debugf("PortForward cancelled in network namespace %q port %d: %v", id, port, ctx.Err())
+			return ctx.Err()
 		}
-		return nil
+		// give a chance to terminate gracefully or timeout
+		// 0.5s is the default timeout used in socat
+		// https://linux.die.net/man/1/socat
+		timeout := time.Duration(500) * time.Millisecond
+		select {
+		case e := <-errCh:
+			if errFwd == nil {
+				errFwd = e
+			}
+			log.G(ctx).Debugf("PortForward stopped forwarding in both directions in network namespace %q port %d: %v", id, port, e)
+		case <-time.After(timeout):
+			log.G(ctx).Debugf("PortForward timed out waiting to close the connection in network namespace %q port %d", id, port)
+		case <-ctx.Done():
+			log.G(ctx).Debugf("PortForward cancelled in network namespace %q port %d: %v", id, port, ctx.Err())
+			errFwd = ctx.Err()
+		}
+
+		return errFwd
 	})
+
 	if err != nil {
 		return errors.Wrapf(err, "failed to execute portforward in network namespace %q", netNSPath)
 	}