Merge pull request #3659 from katiewasnothere/shimreconnectupstream

crosbymichael · web-flow · commit 324a94790d7c · 2019-09-18T12:50:08.000-04:00
fail on file not found for shim reconnect on containerd restart
diff --git a/runtime/v2/binary.go b/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")
 	}
diff --git a/runtime/v2/shim.go b/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")
 	}
diff --git a/runtime/v2/shim/util_unix.go b/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 {
diff --git a/runtime/v2/shim/util_windows.go b/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)
diff --git a/runtime/v2/shim_unix.go b/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)
 }
 
diff --git a/runtime/v2/shim_windows.go b/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,
 		)

Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,7 @@ func (b binary) Start(ctx context.Context, opts types.Any, onClose func()) (_`
`74`	`74`	`if err != nil {`
`75`	`75`	`return nil, err`
`76`	`76`	`}`
`77`		`- f, err := openShimLog(ctx, b.bundle)`
	`77`	`+ f, err := openShimLog(ctx, b.bundle, client.AnonDialer)`
`78`	`78`	`if err != nil {`
`79`	`79`	`return nil, errors.Wrap(err, "open shim log pipe")`
`80`	`80`	`}`
Original file line number	Diff line number	Diff line change
`@@ -67,7 +67,7 @@ func loadShim(ctx context.Context, bundle Bundle, events exchange.Exchange, rt`
`67`	`67`	`if err != nil {`
`68`	`68`	`return nil, err`
`69`	`69`	`}`
`70`		`- conn, err := client.Connect(address, client.AnonDialer)`
	`70`	`+ conn, err := client.Connect(address, client.AnonReconnectDialer)`
`71`	`71`	`if err != nil {`
`72`	`72`	`return nil, err`
`73`	`73`	`}`
`@@ -76,7 +76,7 @@ func loadShim(ctx context.Context, bundle Bundle, events exchange.Exchange, rt`
`76`	`76`	`conn.Close()`
`77`	`77`	`}`
`78`	`78`	`}()`
`79`		`- f, err := openShimLog(ctx, bundle)`
	`79`	`+ f, err := openShimLog(ctx, bundle, client.AnonReconnectDialer)`
`80`	`80`	`if err != nil {`
`81`	`81`	`return nil, errors.Wrap(err, "open shim log pipe")`
`82`	`82`	`}`
Original file line number	Diff line number	Diff line change
`@@ -21,13 +21,15 @@ package v2`
`21`	`21`	`import (`
`22`	`22`	`"context"`
`23`	`23`	`"io"`
	`24`	`+ "net"`
`24`	`25`	`"path/filepath"`
	`26`	`+ "time"`
`25`	`27`
`26`	`28`	`"github.com/containerd/fifo"`
`27`	`29`	`"golang.org/x/sys/unix"`
`28`	`30`	`)`
`29`	`31`
`30`		`-func openShimLog(ctx context.Context, bundle *Bundle) (io.ReadCloser, error) {`
	`32`	`+func openShimLog(ctx context.Context, bundle *Bundle, _ func(string, time.Duration) (net.Conn, error)) (io.ReadCloser, error) {`
`31`	`33`	`return fifo.OpenFifo(ctx, filepath.Join(bundle.Path, "log"), unix.O_RDONLY\|unix.O_CREAT\|unix.O_NONBLOCK, 0700)`
`32`	`34`	`}`
`33`	`35`