@@ -137,7 +137,7 @@ github.com/containerd/cgroups                       318312a373405e5e91134d8063d0

 github.com/containerd/console                       5d7e1412f07b502a01029ea20e20e0d2be31fa7c # v1.0.1

 github.com/containerd/go-runc                       16b287bc67d069a60fa48db15f330b790b74365b

 github.com/containerd/typeurl                       cd3ce7159eae562a4f60ceff37dada11a939d247 # v1.0.1

-github.com/containerd/ttrpc                         72bb1b21c5b0a4a107f59dd85f6ab58e564b68d6 # v1.0.1

+github.com/containerd/ttrpc                         bfba540dc45464586c106b1f31c8547933c1eb41 # v1.0.2

 github.com/gogo/googleapis                          01e0f9cca9b92166042241267ee2a5cdf5cff46c # v1.3.2

 github.com/cilium/ebpf                              1c8d4c9ef7759622653a1d319284a44652333b28

new file mode 100644

@@ -0,0 +1,53 @@

+/*

+   Copyright The containerd Authors.

+

+   Licensed under the Apache License, Version 2.0 (the "License");

+   you may not use this file except in compliance with the License.

+   You may obtain a copy of the License at

+

+       http://www.apache.org/licenses/LICENSE-2.0

+

+   Unless required by applicable law or agreed to in writing, software

+   distributed under the License is distributed on an "AS IS" BASIS,

+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

+   See the License for the specific language governing permissions and

+   limitations under the License.

+*/

+

+package runtime

+

+import (

+	"net/url"

+	"os"

+	"os/exec"

+)

+

+// NewBinaryCmd returns a Cmd to be used to start a logging binary.

+// The Cmd is generated from the provided uri, and the container ID and

+// namespace are appended to the Cmd environment.

+func NewBinaryCmd(binaryURI *url.URL, id, ns string) *exec.Cmd {

+	var args []string

+	for k, vs := range binaryURI.Query() {

+		args = append(args, k)

+		if len(vs) > 0 {

+			args = append(args, vs[0])

+		}

+	}

+

+	cmd := exec.Command(binaryURI.Path, args...)

+

+	cmd.Env = append(cmd.Env,

+		"CONTAINER_ID="+id,

+		"CONTAINER_NAMESPACE="+ns,

+	)

+

+	return cmd

+}

+

+// CloseFiles closes any files passed in.

+// It it used for cleanup in the event of unexpected errors.

+func CloseFiles(files ...*os.File) {

+	for _, file := range files {

+		file.Close()

+	}

+}

@@ -47,8 +47,9 @@ type Client struct {

 	ctx    context.Context

 	closed func()

-	closeOnce     sync.Once

-	userCloseFunc func()

+	closeOnce       sync.Once

+	userCloseFunc   func()

+	userCloseWaitCh chan struct{}

 	errOnce     sync.Once

 	err         error

@@ -75,14 +76,15 @@ func WithUnaryClientInterceptor(i UnaryClientInterceptor) ClientOpts {

 func NewClient(conn net.Conn, opts ...ClientOpts) *Client {

 	ctx, cancel := context.WithCancel(context.Background())

 	c := &Client{

-		codec:         codec{},

-		conn:          conn,

-		channel:       newChannel(conn),

-		calls:         make(chan *callRequest),

-		closed:        cancel,

-		ctx:           ctx,

-		userCloseFunc: func() {},

-		interceptor:   defaultClientInterceptor,

+		codec:           codec{},

+		conn:            conn,

+		channel:         newChannel(conn),

+		calls:           make(chan *callRequest),

+		closed:          cancel,

+		ctx:             ctx,

+		userCloseFunc:   func() {},

+		userCloseWaitCh: make(chan struct{}),

+		interceptor:     defaultClientInterceptor,

 	}

 	for _, o := range opts {

@@ -175,6 +177,17 @@ func (c *Client) Close() error {

 	return nil

 }

+// UserOnCloseWait is used to blocks untils the user's on-close callback

+// finishes.

+func (c *Client) UserOnCloseWait(ctx context.Context) error {

+	select {

+	case <-c.userCloseWaitCh:

+		return nil

+	case <-ctx.Done():

+		return ctx.Err()

+	}

+}

+

 type message struct {

 	messageHeader

 	p   []byte

@@ -251,6 +264,7 @@ func (c *Client) run() {

 	defer func() {

 		c.conn.Close()

 		c.userCloseFunc()

+		close(c.userCloseWaitCh)

 	}()

 	for {

@@ -339,7 +353,8 @@ func filterCloseErr(err error) error {

 		return ErrClosed

 	default:

 		// if we have an epipe on a write or econnreset on a read , we cast to errclosed

-		if oerr, ok := err.(*net.OpError); ok && (oerr.Op == "write" || oerr.Op == "read") {

+		var oerr *net.OpError

+		if errors.As(err, &oerr) && (oerr.Op == "write" || oerr.Op == "read") {

 			serr, sok := oerr.Err.(*os.SyscallError)

 			if sok && ((serr.Err == syscall.EPIPE && oerr.Op == "write") ||

 				(serr.Err == syscall.ECONNRESET && oerr.Op == "read")) {