updates to containerd executor, including basic CNI based network support

agent/exec/containerd/adapter.go

@@ -11,6 +11,9 @@ import (
 
 	"github.com/Sirupsen/logrus"
 	"github.com/containerd/containerd"
+	"github.com/containerd/containerd/errdefs"
+	"github.com/containernetworking/cni/libcni"
+	cnicurr "github.com/containernetworking/cni/pkg/types/current"
 	"github.com/docker/docker/pkg/signal"
 	"github.com/docker/swarmkit/agent/exec"
 	"github.com/docker/swarmkit/api"
@@ -20,8 +23,6 @@ import (
 	"github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
 	"golang.org/x/net/context"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
 )
 
 var (
@@ -35,8 +36,17 @@ var (
 		api.MountPropagationRSlave:   "slave",
 		api.MountPropagationSlave:    "rslave",
 	}
+
+	cniPath = []string{"/opt/cni/bin"}
 )
 
+type networkStatus struct {
+	attachment *api.NetworkAttachment
+	iface      string
+	cniErr     error
+	cni        *cnicurr.Result
+}
+
 // containerAdapter conducts remote operations for a container. All calls
 // are mostly naked calls to the client API, seeded with information from
 // containerConfig.
@@ -49,6 +59,7 @@ type containerAdapter struct {
 	container  containerd.Container
 	task       containerd.Task
 	exitStatus error
+	networks   []*networkStatus
 }
 
 func newContainerAdapter(client *containerd.Client, task *api.Task, secrets exec.SecretGetter) (*containerAdapter, error) {
@@ -64,6 +75,16 @@ func newContainerAdapter(client *containerd.Client, task *api.Task, secrets exec
 		name:    naming.Task(task),
 	}
 
+	for i, na := range task.Networks {
+		c.networks = append(c.networks, &networkStatus{
+			attachment: na,
+			// This becomes CNI_IFNAME which according to
+			// the spec is optional, but the
+			// implementation currently requires it.
+			iface: fmt.Sprintf("eth%d", i),
+		})
+	}
+
 	if err := c.reattach(context.Background()); err != nil {
 		return nil, err
 	}
@@ -77,7 +98,7 @@ func newContainerAdapter(client *containerd.Client, task *api.Task, secrets exec
 func (c *containerAdapter) reattach(ctx context.Context) error {
 	container, err := c.client.LoadContainer(ctx, c.name)
 	if err != nil {
-		if grpc.Code(err) == codes.NotFound {
+		if errdefs.IsNotFound(err) {
 			c.log(ctx).Debug("reattach: container not found")
 			return nil
 		}
@@ -97,7 +118,7 @@ func (c *containerAdapter) reattach(ctx context.Context) error {
 
 	task, err := container.Task(ctx, containerd.WithAttach(devNull, os.Stdout, os.Stderr))
 	if err != nil {
-		if err == containerd.ErrNoRunningTask {
+		if errdefs.IsNotFound(err) {
 			c.log(ctx).WithError(err).Info("reattach: no running task")
 			return nil
 		}
@@ -193,6 +214,29 @@ func (c *containerAdapter) isPrepared() bool {
 	return c.container != nil && c.task != nil
 }
 
+func cniConfig(n *api.Network) (*libcni.NetworkConfig, error) {
+	if n.DriverState == nil || n.DriverState.Name != "cni" {
+		return nil, errors.New("containerd executor only supports CNI")
+	}
+
+	cniConfig, ok := n.DriverState.Options["config"]
+	if !ok {
+		return nil, errors.New("CNI network has no config")
+	}
+
+	// TODO(ijc) figure out how to cache this.
+	cninet, err := libcni.ConfFromBytes([]byte(cniConfig))
+	if err != nil {
+		return nil, errors.Wrap(err, "parsing CNI conf")
+	}
+
+	// TODO(ijc) could merge /etc/cni/net.d/<NAME>.{conf,json}, or
+	// even support networks with no Options["config"] and just go
+	// by the annotations name.
+
+	return cninet, nil
+}
+
 func (c *containerAdapter) prepare(ctx context.Context) error {
 	if c.isPrepared() {
 		return errors.New("adapter already prepared")
@@ -256,6 +300,40 @@ func (c *containerAdapter) prepare(ctx context.Context) error {
 		return errors.Wrap(err, "creating task")
 	}
 
+	cni := libcni.CNIConfig{Path: cniPath}
+	log.G(ctx).Infof("Adding %d networks", len(c.networks))
+	for _, na := range c.networks {
+		log.G(ctx).Infof("Network %T", na.attachment.Network.Spec)
+		cninet, err := cniConfig(na.attachment.Network)
+		if err != nil {
+			// XXX destroy container + task
+			return err
+		}
+
+		// Perhaps this should be in some state somewhere, e.g. populated by networkallocator etc in some cases?
+		rt := &libcni.RuntimeConf{
+			ContainerID: c.container.ID(),
+			NetNS:       fmt.Sprintf("/proc/%d/ns/net", c.task.Pid()),
+			IfName:      na.iface,
+		}
+
+		rawResult, err := cni.AddNetwork(cninet, rt)
+		if err != nil {
+			na.cniErr = err
+			log.G(ctx).Errorf("cni.AddNetwork failed: %s", err)
+			return errors.Wrap(err, "CNI add")
+		}
+		result, err := cnicurr.NewResultFromResult(rawResult)
+		if err != nil {
+			na.cniErr = err
+			// XXX destroy container + task
+			return errors.Wrap(err, "CNI add result conversion")
+		}
+		na.cni = result
+
+		log.G(ctx).Debugf("CNI add result: %v", result)
+	}
+
 	return nil
 }
 

agent/exec/containerd/controller.go

@@ -57,6 +57,58 @@ func (r *controller) ContainerStatus(ctx context.Context) (*api.ContainerStatus,
 		status.ExitCode = int32(ec.ExitCode())
 	}
 
+	for _, n := range r.adapter.networks {
+		l := log.G(ctx).WithField("network.id", n.attachment.Network.ID)
+
+		ns := &api.ContainerNetworkStatus{
+			Network: n.attachment.Network.ID,
+		}
+
+		if n.cniErr != nil {
+			ns.Error = n.cniErr.Error()
+		} else if n.cni == nil {
+			ns.Error = "No CNI information available"
+		} else {
+			for _, iface := range n.cni.Interfaces {
+				ns.Interfaces = append(ns.Interfaces, &api.ContainerNetworkStatus_NetworkInterface{
+					Name: iface.Name,
+				})
+			}
+
+			// Now len(ns.Interfaces) == len(n.cni.Interfaces) by construction.
+
+			// Create a dummy interface if nothing
+			// provided, this has been seen in the wild
+			// (with at least the weave-net plugin, along
+			// with n.cni.IPs containing Interface == -1
+			// (handled below too)
+			if len(ns.Interfaces) == 0 {
+				l.Debugf("No interfaces, creating dummy %s", n.iface)
+				ns.Interfaces = append(ns.Interfaces, &api.ContainerNetworkStatus_NetworkInterface{
+					Name: n.iface,
+				})
+			}
+
+			for _, ip := range n.cni.IPs {
+				ifnum := ip.Interface
+				if ifnum < 0 || ifnum >= len(ns.Interfaces) {
+					if len(ns.Interfaces) > 0 {
+						l.Debugf("Interface index %d out of range [0-%d) assuming first", ip.Interface, len(ns.Interfaces))
+						ifnum = 0
+					} else {
+						// This cannot happen given the workaround above, but keep this check for future use.
+						l.Debugf("Interface index %d out of range, no fallback available", ip.Interface)
+						continue
+					}
+				}
+				ni := ns.Interfaces[ifnum]
+				ni.IP = append(ni.IP, ip.Address.String())
+			}
+		}
+
+		status.Networks = append(status.Networks, ns)
+	}
+
 	return status, nil
 }