Merge pull request #3659 from katiewasnothere/shimreconnectupstream

fail on file not found for shim reconnect on containerd restart

runtime/v2/binary.go

@@ -74,7 +74,7 @@ func (b *binary) Start(ctx context.Context, opts *types.Any, onClose func()) (_
 	if err != nil {
 		return nil, err
 	}
-	f, err := openShimLog(ctx, b.bundle)
+	f, err := openShimLog(ctx, b.bundle, client.AnonDialer)
 	if err != nil {
 		return nil, errors.Wrap(err, "open shim log pipe")
 	}

runtime/v2/shim.go

@@ -67,7 +67,7 @@ func loadShim(ctx context.Context, bundle *Bundle, events *exchange.Exchange, rt
 	if err != nil {
 		return nil, err
 	}
-	conn, err := client.Connect(address, client.AnonDialer)
+	conn, err := client.Connect(address, client.AnonReconnectDialer)
 	if err != nil {
 		return nil, err
 	}
@@ -76,7 +76,7 @@ func loadShim(ctx context.Context, bundle *Bundle, events *exchange.Exchange, rt
 			conn.Close()
 		}
 	}()
-	f, err := openShimLog(ctx, bundle)
+	f, err := openShimLog(ctx, bundle, client.AnonReconnectDialer)
 	if err != nil {
 		return nil, errors.Wrap(err, "open shim log pipe")
 	}

runtime/v2/shim/util_unix.go

@@ -78,6 +78,10 @@ func AnonDialer(address string, timeout time.Duration) (net.Conn, error) {
 	return net.DialTimeout("unix", "\x00"+address, timeout)
 }
 
+func AnonReconnectDialer(address string, timeout time.Duration) (net.Conn, error) {
+	return AnonDialer(address, timeout)
+}
+
 // NewSocket returns a new socket
 func NewSocket(address string) (*net.UnixListener, error) {
 	if len(address) > 106 {

runtime/v2/shim/util_windows.go

@@ -33,6 +33,22 @@ func getSysProcAttr() *syscall.SysProcAttr {
 	return nil
 }
 
+// AnonReconnectDialer returns a dialer for an existing npipe on containerd reconnection
+func AnonReconnectDialer(address string, timeout time.Duration) (net.Conn, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	c, err := winio.DialPipeContext(ctx, address)
+	if os.IsNotExist(err) {
+		return nil, errors.Wrap(os.ErrNotExist, "npipe not found on reconnect")
+	} else if err == context.DeadlineExceeded {
+		return nil, errors.Wrapf(err, "timed out waiting for npipe %s", address)
+	} else if err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
 // AnonDialer returns a dialer for a npipe
 func AnonDialer(address string, timeout time.Duration) (net.Conn, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)

runtime/v2/shim_unix.go

@@ -21,13 +21,15 @@ package v2
 import (
 	"context"
 	"io"
+	"net"
 	"path/filepath"
+	"time"
 
 	"github.com/containerd/fifo"
 	"golang.org/x/sys/unix"
 )
 
-func openShimLog(ctx context.Context, bundle *Bundle) (io.ReadCloser, error) {
+func openShimLog(ctx context.Context, bundle *Bundle, _ func(string, time.Duration) (net.Conn, error)) (io.ReadCloser, error) {
 	return fifo.OpenFifo(ctx, filepath.Join(bundle.Path, "log"), unix.O_RDONLY|unix.O_CREAT|unix.O_NONBLOCK, 0700)
 }
 

runtime/v2/shim_windows.go

@@ -26,7 +26,6 @@ import (
 	"time"
 
 	"github.com/containerd/containerd/namespaces"
-	client "github.com/containerd/containerd/runtime/v2/shim"
 	"github.com/pkg/errors"
 )
 
@@ -64,7 +63,7 @@ func (dpc *deferredPipeConnection) Close() error {
 
 // openShimLog on Windows acts as the client of the log pipe. In this way the
 // containerd daemon can reconnect to the shim log stream if it is restarted.
-func openShimLog(ctx context.Context, bundle *Bundle) (io.ReadCloser, error) {
+func openShimLog(ctx context.Context, bundle *Bundle, dialer func(string, time.Duration) (net.Conn, error)) (io.ReadCloser, error) {
 	ns, err := namespaces.NamespaceRequired(ctx)
 	if err != nil {
 		return nil, err
@@ -74,7 +73,7 @@ func openShimLog(ctx context.Context, bundle *Bundle) (io.ReadCloser, error) {
 	}
 	dpc.wg.Add(1)
 	go func() {
-		c, conerr := client.AnonDialer(
+		c, conerr := dialer(
 			fmt.Sprintf("\\\\.\\pipe\\containerd-shim-%s-%s-log", ns, bundle.ID),
 			time.Second*10,
 		)