containerd-shim API Exposed to Host Network Containers

@@ -71,7 +71,7 @@ var (
 func init() {
 	flag.BoolVar(&debugFlag, "debug", false, "enable debug output in logs")
 	flag.StringVar(&namespaceFlag, "namespace", "", "namespace that owns the shim")
-	flag.StringVar(&socketFlag, "socket", "", "abstract socket path to serve")
+	flag.StringVar(&socketFlag, "socket", "", "socket path to serve")
 	flag.StringVar(&addressFlag, "address", "", "grpc address back to main containerd")
 	flag.StringVar(&workdirFlag, "workdir", "", "path used to storge large temporary data")
 	flag.StringVar(&runtimeRootFlag, "runtime-root", process.RuncRoot, "root directory for the runtime")
@@ -202,10 +202,18 @@ func serve(ctx context.Context, server *ttrpc.Server, path string) error {
 		f.Close()
 		path = "[inherited from parent]"
 	} else {
-		if len(path) > 106 {
-			return errors.Errorf("%q: unix socket path too long (> 106)", path)
+		const (
+			abstractSocketPrefix = "\x00"
+			socketPathLimit      = 106
+		)
+		p := strings.TrimPrefix(path, "unix://")
+		if len(p) == len(path) {
+			p = abstractSocketPrefix + p
 		}
-		l, err = net.Listen("unix", "\x00"+path)
+		if len(p) > socketPathLimit {
+			return errors.Errorf("%q: unix socket path too long (> %d)", p, socketPathLimit)
+		}
+		l, err = net.Listen("unix", p)
 	}
 	if err != nil {
 		return err

@@ -24,6 +24,7 @@ import (
 	"io/ioutil"
 	"net"
 	"path/filepath"
+	"strings"
 
 	"github.com/containerd/console"
 	"github.com/containerd/containerd/cmd/ctr/commands"
@@ -240,10 +241,11 @@ func getTaskService(context *cli.Context) (task.TaskService, error) {
 	s1 := filepath.Join(string(filepath.Separator), "containerd-shim", ns, id, "shim.sock")
 	// this should not error, ctr always get a default ns
 	ctx := namespaces.WithNamespace(gocontext.Background(), ns)
-	s2, _ := shim.SocketAddress(ctx, id)
+	s2, _ := shim.SocketAddress(ctx, context.GlobalString("address"), id)
+	s2 = strings.TrimPrefix(s2, "unix://")
 
-	for _, socket := range []string{s1, s2} {
-		conn, err := net.Dial("unix", "\x00"+socket)
+	for _, socket := range []string{s2, "\x00" + s1} {
+		conn, err := net.Dial("unix", socket)
 		if err == nil {
 			client := ttrpc.NewClient(conn)
 

@@ -91,7 +91,7 @@ func ShimRemote(c *Config, daemonAddress, cgroup string, exitHandler func()) Shi
 	return func(b *bundle, ns string, ropts *runctypes.RuncOptions) (shim.Config, client.Opt) {
 		config := b.shimConfig(ns, c, ropts)
 		return config,
-			client.WithStart(c.Shim, b.shimAddress(ns), daemonAddress, cgroup, c.ShimDebug, exitHandler)
+			client.WithStart(c.Shim, b.shimAddress(ns, daemonAddress), daemonAddress, cgroup, c.ShimDebug, exitHandler)
 	}
 }
 
@@ -117,6 +117,11 @@ func (b *bundle) NewShimClient(ctx context.Context, namespace string, getClientO
 
 // Delete deletes the bundle from disk
 func (b *bundle) Delete() error {
+	address, _ := b.loadAddress()
+	if address != "" {
+		// we don't care about errors here
+		client.RemoveSocket(address)
+	}
 	err := atomicDelete(b.path)
 	if err == nil {
 		return atomicDelete(b.workDir)
@@ -133,9 +138,11 @@ func (b *bundle) legacyShimAddress(namespace string) string {
 	return filepath.Join(string(filepath.Separator), "containerd-shim", namespace, b.id, "shim.sock")
 }
 
-func (b *bundle) shimAddress(namespace string) string {
-	d := sha256.Sum256([]byte(filepath.Join(namespace, b.id)))
-	return filepath.Join(string(filepath.Separator), "containerd-shim", fmt.Sprintf("%x.sock", d))
+const socketRoot = "/run/containerd"
+
+func (b *bundle) shimAddress(namespace, socketPath string) string {
+	d := sha256.Sum256([]byte(filepath.Join(socketPath, namespace, b.id)))
+	return fmt.Sprintf("unix://%s/%x", filepath.Join(socketRoot, "s"), d)
 }
 
 func (b *bundle) loadAddress() (string, error) {

@@ -59,9 +59,17 @@ func WithStart(binary, address, daemonAddress, cgroup string, debug bool, exitHa
 	return func(ctx context.Context, config shim.Config) (_ shimapi.ShimService, _ io.Closer, err error) {
 		socket, err := newSocket(address)
 		if err != nil {
-			return nil, nil, err
+			if !eaddrinuse(err) {
+				return nil, nil, err
+			}
+			if err := RemoveSocket(address); err != nil {
+				return nil, nil, errors.Wrap(err, "remove already used socket")
+			}
+			if socket, err = newSocket(address); err != nil {
+				return nil, nil, err
+			}
 		}
-		defer socket.Close()
+
 		f, err := socket.File()
 		if err != nil {
 			return nil, nil, errors.Wrapf(err, "failed to get fd for socket %s", address)
@@ -108,6 +116,8 @@ func WithStart(binary, address, daemonAddress, cgroup string, debug bool, exitHa
 			if stderrLog != nil {
 				stderrLog.Close()
 			}
+			socket.Close()
+			RemoveSocket(address)
 		}()
 		log.G(ctx).WithFields(logrus.Fields{
 			"pid":     cmd.Process.Pid,
@@ -142,6 +152,26 @@ func WithStart(binary, address, daemonAddress, cgroup string, debug bool, exitHa
 	}
 }
 
+func eaddrinuse(err error) bool {
+	cause := errors.Cause(err)
+	netErr, ok := cause.(*net.OpError)
+	if !ok {
+		return false
+	}
+	if netErr.Op != "listen" {
+		return false
+	}
+	syscallErr, ok := netErr.Err.(*os.SyscallError)
+	if !ok {
+		return false
+	}
+	errno, ok := syscallErr.Err.(syscall.Errno)
+	if !ok {
+		return false
+	}
+	return errno == syscall.EADDRINUSE
+}
+
 // setupOOMScore gets containerd's oom score and adds +1 to it
 // to ensure a shim has a lower* score than the daemons
 func setupOOMScore(shimPid int) error {
@@ -214,31 +244,73 @@ func writeFile(path, address string) error {
 	return os.Rename(tempPath, path)
 }
 
+const (
+	abstractSocketPrefix = "\x00"
+	socketPathLimit      = 106
+)
+
+type socket string
+
+func (s socket) isAbstract() bool {
+	return !strings.HasPrefix(string(s), "unix://")
+}
+
+func (s socket) path() string {
+	path := strings.TrimPrefix(string(s), "unix://")
+	// if there was no trim performed, we assume an abstract socket
+	if len(path) == len(s) {
+		path = abstractSocketPrefix + path
+	}
+	return path
+}
+
 func newSocket(address string) (*net.UnixListener, error) {
-	if len(address) > 106 {
-		return nil, errors.Errorf("%q: unix socket path too long (> 106)", address)
+	if len(address) > socketPathLimit {
+		return nil, errors.Errorf("%q: unix socket path too long (> %d)", address, socketPathLimit)
+	}
+	var (
+		sock = socket(address)
+		path = sock.path()
+	)
+	if !sock.isAbstract() {
+		if err := os.MkdirAll(filepath.Dir(path), 0600); err != nil {
+			return nil, errors.Wrapf(err, "%s", path)
+		}
 	}
-	l, err := net.Listen("unix", "\x00"+address)
+	l, err := net.Listen("unix", path)
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to listen to abstract unix socket %q", address)
+		return nil, errors.Wrapf(err, "failed to listen to unix socket %q (abstract: %t)", address, sock.isAbstract())
+	}
+	if err := os.Chmod(path, 0600); err != nil {
+		l.Close()
+		return nil, err
 	}
 
 	return l.(*net.UnixListener), nil
 }
 
+// RemoveSocket removes the socket at the specified address if
+// it exists on the filesystem
+func RemoveSocket(address string) error {
+	sock := socket(address)
+	if !sock.isAbstract() {
+		return os.Remove(sock.path())
+	}
+	return nil
+}
+
 func connect(address string, d func(string, time.Duration) (net.Conn, error)) (net.Conn, error) {
 	return d(address, 100*time.Second)
 }
 
-func annonDialer(address string, timeout time.Duration) (net.Conn, error) {
-	address = strings.TrimPrefix(address, "unix://")
-	return dialer.Dialer("\x00"+address, timeout)
+func anonDialer(address string, timeout time.Duration) (net.Conn, error) {
+	return dialer.Dialer(socket(address).path(), timeout)
 }
 
 // WithConnect connects to an existing shim
 func WithConnect(address string, onClose func()) Opt {
 	return func(ctx context.Context, config shim.Config) (shimapi.ShimService, io.Closer, error) {
-		conn, err := connect(address, annonDialer)
+		conn, err := connect(address, anonDialer)
 		if err != nil {
 			return nil, nil, err
 		}

@@ -131,20 +131,26 @@ func (s *service) StartShim(ctx context.Context, id, containerdBinary, container
 	if err != nil {
 		return "", err
 	}
-	address, err := shim.SocketAddress(ctx, id)
+	address, err := shim.SocketAddress(ctx, containerdAddress, id)
 	if err != nil {
 		return "", err
 	}
 	socket, err := shim.NewSocket(address)
 	if err != nil {
-		return "", err
+		if !shim.SocketEaddrinuse(err) {
+			return "", err
+		}
+		if err := shim.RemoveSocket(address); err != nil {
+			return "", errors.Wrap(err, "remove already used socket")
+		}
+		if socket, err = shim.NewSocket(address); err != nil {
+			return "", err
+		}
 	}
-	defer socket.Close()
 	f, err := socket.File()
 	if err != nil {
 		return "", err
 	}
-	defer f.Close()
 
 	cmd.ExtraFiles = append(cmd.ExtraFiles, f)
 
@@ -153,6 +159,7 @@ func (s *service) StartShim(ctx context.Context, id, containerdBinary, container
 	}
 	defer func() {
 		if err != nil {
+			_ = shim.RemoveSocket(address)
 			cmd.Process.Kill()
 		}
 	}()
@@ -551,6 +558,9 @@ func (s *service) Connect(ctx context.Context, r *taskAPI.ConnectRequest) (*task
 func (s *service) Shutdown(ctx context.Context, r *taskAPI.ShutdownRequest) (*ptypes.Empty, error) {
 	s.cancel()
 	close(s.events)
+	if address, err := shim.ReadAddress("address"); err == nil {
+		_ = shim.RemoveSocket(address)
+	}
 	return empty, nil
 }
 

@@ -25,7 +25,6 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
-	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -105,6 +104,10 @@ func New(ctx context.Context, id string, publisher shim.Publisher, shutdown func
 		return nil, errors.Wrap(err, "failed to initialized platform behavior")
 	}
 	go s.forward(ctx, publisher)
+
+	if address, err := shim.ReadAddress("address"); err == nil {
+		s.shimAddress = address
+	}
 	return s, nil
 }
 
@@ -124,7 +127,8 @@ type service struct {
 
 	containers map[string]*runc.Container
 
-	cancel func()
+	shimAddress string
+	cancel      func()
 }
 
 func newCommand(ctx context.Context, id, containerdBinary, containerdAddress, containerdTTRPCAddress string) (*exec.Cmd, error) {
@@ -183,30 +187,48 @@ func (s *service) StartShim(ctx context.Context, id, containerdBinary, container
 			break
 		}
 	}
-	address, err := shim.SocketAddress(ctx, grouping)
+	address, err := shim.SocketAddress(ctx, containerdAddress, grouping)
 	if err != nil {
 		return "", err
 	}
+
 	socket, err := shim.NewSocket(address)
 	if err != nil {
-		if strings.Contains(err.Error(), "address already in use") {
+		// the only time where this would happen is if there is a bug and the socket
+		// was not cleaned up in the cleanup method of the shim or we are using the
+		// grouping functionality where the new process should be run with the same
+		// shim as an existing container
+		if !shim.SocketEaddrinuse(err) {
+			return "", errors.Wrap(err, "create new shim socket")
+		}
+		if shim.CanConnect(address) {
 			if err := shim.WriteAddress("address", address); err != nil {
-				return "", err
+				return "", errors.Wrap(err, "write existing socket for shim")
 			}
 			return address, nil
 		}
-		return "", err
+		if err := shim.RemoveSocket(address); err != nil {
+			return "", errors.Wrap(err, "remove pre-existing socket")
+		}
+		if socket, err = shim.NewSocket(address); err != nil {
+			return "", errors.Wrap(err, "try create new shim socket 2x")
+		}
 	}
-	defer socket.Close()
+	defer func() {
+		if retErr != nil {
+			socket.Close()
+			_ = shim.RemoveSocket(address)
+		}
+	}()
 	f, err := socket.File()
 	if err != nil {
 		return "", err
 	}
-	defer f.Close()
 
 	cmd.ExtraFiles = append(cmd.ExtraFiles, f)
 
 	if err := cmd.Start(); err != nil {
+		f.Close()
 		return "", err
 	}
 	defer func() {
@@ -273,7 +295,6 @@ func (s *service) Cleanup(ctx context.Context) (*taskAPI.DeleteResponse, error)
 	if err != nil {
 		return nil, err
 	}
-
 	runtime, err := runc.ReadRuntime(path)
 	if err != nil {
 		return nil, err
@@ -652,7 +673,9 @@ func (s *service) Shutdown(ctx context.Context, r *taskAPI.ShutdownRequest) (*pt
 	if s.platform != nil {
 		s.platform.Close()
 	}
-
+	if s.shimAddress != "" {
+		_ = shim.RemoveSocket(s.shimAddress)
+	}
 	return empty, nil
 }
 

@@ -104,7 +104,7 @@ func parseFlags() {
 	flag.BoolVar(&versionFlag, "v", false, "show the shim version and exit")
 	flag.StringVar(&namespaceFlag, "namespace", "", "namespace that owns the shim")
 	flag.StringVar(&idFlag, "id", "", "id of the task")
-	flag.StringVar(&socketFlag, "socket", "", "abstract socket path to serve")
+	flag.StringVar(&socketFlag, "socket", "", "socket path to serve")
 	flag.StringVar(&bundlePath, "bundle", "", "path to the bundle if not workdir")
 
 	flag.StringVar(&addressFlag, "address", "", "grpc address back to main containerd")
@@ -195,7 +195,6 @@ func run(id string, initFunc Init, config Config) error {
 	ctx = context.WithValue(ctx, OptsKey{}, Opts{BundlePath: bundlePath, Debug: debugFlag})
 	ctx = log.WithLogger(ctx, log.G(ctx).WithField("runtime", id))
 	ctx, cancel := context.WithCancel(ctx)
-
 	service, err := initFunc(ctx, idFlag, publisher, cancel)
 	if err != nil {
 		return err
@@ -300,11 +299,15 @@ func serve(ctx context.Context, server *ttrpc.Server, path string) error {
 		return err
 	}
 	go func() {
-		defer l.Close()
 		if err := server.Serve(ctx, l); err != nil &&
 			!strings.Contains(err.Error(), "use of closed network connection") {
 			logrus.WithError(err).Fatal("containerd-shim: ttrpc server failure")
 		}
+		l.Close()
+		if address, err := ReadAddress("address"); err == nil {
+			_ = RemoveSocket(address)
+		}
+
 	}()
 	return nil
 }

@@ -58,15 +58,15 @@ func serveListener(path string) (net.Listener, error) {
 		l, err = net.FileListener(os.NewFile(3, "socket"))
 		path = "[inherited from parent]"
 	} else {
-		if len(path) > 106 {
-			return nil, errors.Errorf("%q: unix socket path too long (> 106)", path)
+		if len(path) > socketPathLimit {
+			return nil, errors.Errorf("%q: unix socket path too long (> %d)", path, socketPathLimit)
 		}
-		l, err = net.Listen("unix", "\x00"+path)
+		l, err = net.Listen("unix", path)
 	}
 	if err != nil {
 		return nil, err
 	}
-	logrus.WithField("socket", path).Debug("serving api on abstract socket")
+	logrus.WithField("socket", path).Debug("serving api on socket")
 	return l, nil
 }
 

@@ -169,7 +169,7 @@ func WriteAddress(path, address string) error {
 // ErrNoAddress is returned when the address file has no content
 var ErrNoAddress = errors.New("no shim address")
 
-// ReadAddress returns the shim's abstract socket address from the path
+// ReadAddress returns the shim's socket address from the path
 func ReadAddress(path string) (string, error) {
 	path, err := filepath.Abs(path)
 	if err != nil {

@@ -35,7 +35,10 @@ import (
 	"github.com/pkg/errors"
 )
 
-const shimBinaryFormat = "containerd-shim-%s-%s"
+const (
+	shimBinaryFormat = "containerd-shim-%s-%s"
+	socketPathLimit  = 106
+)
 
 func getSysProcAttr() *syscall.SysProcAttr {
 	return &syscall.SysProcAttr{
@@ -63,20 +66,21 @@ func AdjustOOMScore(pid int) error {
 	return nil
 }
 
-// SocketAddress returns an abstract socket address
-func SocketAddress(ctx context.Context, id string) (string, error) {
+const socketRoot = "/run/containerd"
+
+// SocketAddress returns a socket address
+func SocketAddress(ctx context.Context, socketPath, id string) (string, error) {
 	ns, err := namespaces.NamespaceRequired(ctx)
 	if err != nil {
 		return "", err
 	}
-	d := sha256.Sum256([]byte(filepath.Join(ns, id)))
-	return filepath.Join(string(filepath.Separator), "containerd-shim", fmt.Sprintf("%x.sock", d)), nil
+	d := sha256.Sum256([]byte(filepath.Join(socketPath, ns, id)))
+	return fmt.Sprintf("unix://%s/%x", filepath.Join(socketRoot, "s"), d), nil
 }
 
-// AnonDialer returns a dialer for an abstract socket
+// AnonDialer returns a dialer for a socket
 func AnonDialer(address string, timeout time.Duration) (net.Conn, error) {
-	address = strings.TrimPrefix(address, "unix://")
-	return dialer.Dialer("\x00"+address, timeout)
+	return dialer.Dialer(socket(address).path(), timeout)
 }
 
 func AnonReconnectDialer(address string, timeout time.Duration) (net.Conn, error) {
@@ -85,12 +89,82 @@ func AnonReconnectDialer(address string, timeout time.Duration) (net.Conn, error
 
 // NewSocket returns a new socket
 func NewSocket(address string) (*net.UnixListener, error) {
-	if len(address) > 106 {
-		return nil, errors.Errorf("%q: unix socket path too long (> 106)", address)
+	var (
+		sock = socket(address)
+		path = sock.path()
+	)
+	if !sock.isAbstract() {
+		if err := os.MkdirAll(filepath.Dir(path), 0600); err != nil {
+			return nil, errors.Wrapf(err, "%s", path)
+		}
 	}
-	l, err := net.Listen("unix", "\x00"+address)
+	l, err := net.Listen("unix", path)
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to listen to abstract unix socket %q", address)
+		return nil, err
+	}
+	if err := os.Chmod(path, 0600); err != nil {
+		os.Remove(sock.path())
+		l.Close()
+		return nil, err
 	}
 	return l.(*net.UnixListener), nil
 }
+
+const abstractSocketPrefix = "\x00"
+
+type socket string
+
+func (s socket) isAbstract() bool {
+	return !strings.HasPrefix(string(s), "unix://")
+}
+
+func (s socket) path() string {
+	path := strings.TrimPrefix(string(s), "unix://")
+	// if there was no trim performed, we assume an abstract socket
+	if len(path) == len(s) {
+		path = abstractSocketPrefix + path
+	}
+	return path
+}
+
+// RemoveSocket removes the socket at the specified address if
+// it exists on the filesystem
+func RemoveSocket(address string) error {
+	sock := socket(address)
+	if !sock.isAbstract() {
+		return os.Remove(sock.path())
+	}
+	return nil
+}
+
+// SocketEaddrinuse returns true if the provided error is caused by the
+// EADDRINUSE error number
+func SocketEaddrinuse(err error) bool {
+	netErr, ok := err.(*net.OpError)
+	if !ok {
+		return false
+	}
+	if netErr.Op != "listen" {
+		return false
+	}
+	syscallErr, ok := netErr.Err.(*os.SyscallError)
+	if !ok {
+		return false
+	}
+	errno, ok := syscallErr.Err.(syscall.Errno)
+	if !ok {
+		return false
+	}
+	return errno == syscall.EADDRINUSE
+}
+
+// CanConnect returns true if the socket provided at the address
+// is accepting new connections
+func CanConnect(address string) bool {
+	conn, err := AnonDialer(address, 100*time.Millisecond)
+	if err != nil {
+		return false
+	}
+	conn.Close()
+	return true
+}

@@ -79,3 +79,9 @@ func AnonDialer(address string, timeout time.Duration) (net.Conn, error) {
 		return c, nil
 	}
 }
+
+// RemoveSocket removes the socket at the specified address if
+// it exists on the filesystem
+func RemoveSocket(address string) error {
+	return nil
+}