hotplug feature: add pod networks annotation controller

cmd/multus-daemon/main.go

@@ -366,12 +366,13 @@ func handleSignals(stopChannel chan struct{}, signals ...os.Signal) {
 func newRuntime(config types.ControllerNetConf) (containerruntimes.ContainerRuntime, error) {
 	var runtime containerruntimes.ContainerRuntime
 
+	timeout := time.Second
 	rtType, err := parseRuntimeType(config.CriType)
 	switch rtType {
 	case containerruntimes.Crio:
-		runtime, err = crio.NewCrioRuntime(config.CriSocketPath, 5*time.Second)
+		runtime, err = crio.NewCrioRuntime(config.CriSocketPath, timeout)
 	case containerruntimes.Containerd:
-		runtime, err = containerd.NewContainerdRuntime(config.CriSocketPath, time.Second)
+		runtime, err = containerd.NewContainerdRuntime(config.CriSocketPath, timeout)
 	}
 
 	if err != nil {

pkg/containerruntimes/crio/crio.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/pkg/errors"
 
+	"gopkg.in/k8snetworkplumbingwg/multus-cni.v3/pkg/containerruntimes/crio/types"
 	crioruntime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
 	"k8s.io/kubernetes/pkg/kubelet/util"
 )
@@ -21,28 +22,6 @@ type CrioRuntime struct {
 	context    context.Context
 }
 
-// PodStatusResponseInfo represents the container status reply - crictl ps <containerID>
-type PodStatusResponseInfo struct {
-	SandboxID   string
-	RunTimeSpec RunTimeSpecInfo
-}
-
-// RunTimeSpecInfo represents the relevant part of the container status spec
-type RunTimeSpecInfo struct {
-	Linux NamespacesInfo
-}
-
-// NamespacesInfo represents the container status namespaces
-type NamespacesInfo struct {
-	Namespaces []NameSpaceInfo
-}
-
-// NameSpaceInfo represents the ns info
-type NameSpaceInfo struct {
-	Type string
-	Path string
-}
-
 // NewCrioRuntime returns a connection to the CRI-O runtime
 func NewCrioRuntime(socketPath string, timeout time.Duration) (*CrioRuntime, error) {
 	if socketPath == "" {
@@ -98,17 +77,11 @@ func (cr *CrioRuntime) NetNS(containerID string) (string, error) {
 		Verbose:     true,
 	})
 	if err != nil {
-		return "", errors.Wrap(err, "failed to get pod sandbox info")
+		return "", fmt.Errorf("failed to get pod sandbox info: %w", err) //errors.Wrap(err, "failed to get pod sandbox info")
 	}
 
-	mapInfo := reply.GetInfo()
-	var podStatusResponseInfo PodStatusResponseInfo
-	info := mapInfo["info"]
-	err = json.Unmarshal([]byte(info), &podStatusResponseInfo)
+	podStatusResponseInfo, err := ContainerStatus(reply)
 	if err != nil {
-		if e, ok := err.(*json.SyntaxError); ok {
-			return "", fmt.Errorf("error unmarshalling cri-o's response: syntax error at byte offset %d. Error: %w", e.Offset, e)
-		}
 		return "", err
 	}
 
@@ -120,3 +93,15 @@ func (cr *CrioRuntime) NetNS(containerID string) (string, error) {
 	}
 	return "", nil
 }
+
+func ContainerStatus(reply *crioruntime.ContainerStatusResponse) (types.PodStatusResponseInfo, error) {
+	var podStatusResponseInfo types.PodStatusResponseInfo
+	info := reply.GetInfo()["info"]
+	if err := json.Unmarshal([]byte(info), &podStatusResponseInfo); err != nil {
+		if e, ok := err.(*json.SyntaxError); ok {
+			return types.PodStatusResponseInfo{}, fmt.Errorf("error unmarshalling cri-o's response: syntax error at byte offset %d. Error: %w", e.Offset, e)
+		}
+		return types.PodStatusResponseInfo{}, err
+	}
+	return podStatusResponseInfo, nil
+}

pkg/containerruntimes/crio/crio_test.go

@@ -0,0 +1,61 @@
+package crio
+
+import (
+	"context"
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"gopkg.in/k8snetworkplumbingwg/multus-cni.v3/pkg/containerruntimes/crio/fake"
+)
+
+func TestController(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Dynamic network attachment controller suite")
+}
+
+var _ = Describe("CRI-O runtime", func() {
+	var runtime *CrioRuntime
+
+	When("the runtime *does not* feature any containers", func() {
+		BeforeEach(func() {
+			runtime = newDummyCrioRuntime()
+		})
+
+		It("cannot extract the network namespace of a container", func() {
+			_, err := runtime.NetNS("1234")
+			Expect(err).To(MatchError("failed to get pod sandbox info: container 1234 not found"))
+		})
+	})
+
+	When("a live container is provisioned in the runtime", func() {
+		const (
+			containerID = "1234"
+			netnsPath   = "bottom-drawer"
+		)
+		BeforeEach(func() {
+			runtime = newDummyCrioRuntime(fake.WithCachedContainer(containerID, netnsPath))
+		})
+
+		It("cannot extract the network namespace of a container", func() {
+			Expect(runtime.NetNS(containerID)).To(Equal(netnsPath))
+		})
+	})
+})
+
+func newDummyCrioRuntime(opts ...fake.ClientOpt) *CrioRuntime {
+	runtimeClient := fake.NewFakeClient()
+
+	for _, opt := range opts {
+		opt(runtimeClient)
+	}
+
+	ctx := context.TODO()
+	ctxWithCancel, cancelFunc := context.WithCancel(ctx)
+	return &CrioRuntime{
+		client:     runtimeClient,
+		context:    ctxWithCancel,
+		cancelFunc: cancelFunc,
+	}
+}

pkg/containerruntimes/crio/fake/crio.go

@@ -0,0 +1,147 @@
+package fake
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"google.golang.org/grpc"
+
+	crioruntime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
+
+	"gopkg.in/k8snetworkplumbingwg/multus-cni.v3/pkg/containerruntimes/crio/types"
+)
+
+type CrioClient struct {
+	cache map[string]string
+}
+
+type ClientOpt func(client *CrioClient)
+
+func NewFakeClient(opts ...ClientOpt) *CrioClient {
+	client := &CrioClient{cache: map[string]string{}}
+	for _, opt := range opts {
+		opt(client)
+	}
+	return client
+}
+
+func WithCachedContainer(containerID string, netnsPath string) ClientOpt {
+	return func(client *CrioClient) {
+		client.cache[containerID] = netnsPath
+	}
+}
+
+func (CrioClient) Version(ctx context.Context, in *crioruntime.VersionRequest, opts ...grpc.CallOption) (*crioruntime.VersionResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) RunPodSandbox(ctx context.Context, in *crioruntime.RunPodSandboxRequest, opts ...grpc.CallOption) (*crioruntime.RunPodSandboxResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) StopPodSandbox(ctx context.Context, in *crioruntime.StopPodSandboxRequest, opts ...grpc.CallOption) (*crioruntime.StopPodSandboxResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) RemovePodSandbox(ctx context.Context, in *crioruntime.RemovePodSandboxRequest, opts ...grpc.CallOption) (*crioruntime.RemovePodSandboxResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) PodSandboxStatus(ctx context.Context, in *crioruntime.PodSandboxStatusRequest, opts ...grpc.CallOption) (*crioruntime.PodSandboxStatusResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) ListPodSandbox(ctx context.Context, in *crioruntime.ListPodSandboxRequest, opts ...grpc.CallOption) (*crioruntime.ListPodSandboxResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) CreateContainer(ctx context.Context, in *crioruntime.CreateContainerRequest, opts ...grpc.CallOption) (*crioruntime.CreateContainerResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) StartContainer(ctx context.Context, in *crioruntime.StartContainerRequest, opts ...grpc.CallOption) (*crioruntime.StartContainerResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) StopContainer(ctx context.Context, in *crioruntime.StopContainerRequest, opts ...grpc.CallOption) (*crioruntime.StopContainerResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) RemoveContainer(ctx context.Context, in *crioruntime.RemoveContainerRequest, opts ...grpc.CallOption) (*crioruntime.RemoveContainerResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) ListContainers(ctx context.Context, in *crioruntime.ListContainersRequest, opts ...grpc.CallOption) (*crioruntime.ListContainersResponse, error) {
+	return nil, nil
+}
+
+func (cc CrioClient) ContainerStatus(ctx context.Context, in *crioruntime.ContainerStatusRequest, opts ...grpc.CallOption) (*crioruntime.ContainerStatusResponse, error) {
+	containerId := in.ContainerId
+	netnsPath, wasFound := cc.cache[containerId]
+	if !wasFound {
+		return nil, fmt.Errorf("container %s not found", containerId)
+	}
+
+	containerStatus := types.PodStatusResponseInfo{
+		RunTimeSpec: types.RunTimeSpecInfo{
+			Linux: types.NamespacesInfo{
+				Namespaces: []types.NameSpaceInfo{
+					{
+						Type: "network",
+						Path: netnsPath,
+					},
+				},
+			},
+		},
+	}
+
+	marshalledContainerStatus, err := json.Marshal(&containerStatus)
+	if err != nil {
+		return nil, fmt.Errorf("error marshalling the container status: %v", err)
+	}
+
+	return &crioruntime.ContainerStatusResponse{
+		Info: map[string]string{"info": string(marshalledContainerStatus)},
+	}, nil
+}
+
+func (CrioClient) UpdateContainerResources(ctx context.Context, in *crioruntime.UpdateContainerResourcesRequest, opts ...grpc.CallOption) (*crioruntime.UpdateContainerResourcesResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) ReopenContainerLog(ctx context.Context, in *crioruntime.ReopenContainerLogRequest, opts ...grpc.CallOption) (*crioruntime.ReopenContainerLogResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) ExecSync(ctx context.Context, in *crioruntime.ExecSyncRequest, opts ...grpc.CallOption) (*crioruntime.ExecSyncResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) Exec(ctx context.Context, in *crioruntime.ExecRequest, opts ...grpc.CallOption) (*crioruntime.ExecResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) Attach(ctx context.Context, in *crioruntime.AttachRequest, opts ...grpc.CallOption) (*crioruntime.AttachResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) PortForward(ctx context.Context, in *crioruntime.PortForwardRequest, opts ...grpc.CallOption) (*crioruntime.PortForwardResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) ContainerStats(ctx context.Context, in *crioruntime.ContainerStatsRequest, opts ...grpc.CallOption) (*crioruntime.ContainerStatsResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) ListContainerStats(ctx context.Context, in *crioruntime.ListContainerStatsRequest, opts ...grpc.CallOption) (*crioruntime.ListContainerStatsResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) UpdateRuntimeConfig(ctx context.Context, in *crioruntime.UpdateRuntimeConfigRequest, opts ...grpc.CallOption) (*crioruntime.UpdateRuntimeConfigResponse, error) {
+	return nil, nil
+}
+
+func (CrioClient) Status(ctx context.Context, in *crioruntime.StatusRequest, opts ...grpc.CallOption) (*crioruntime.StatusResponse, error) {
+	return nil, nil
+}

pkg/containerruntimes/crio/types/types.go

@@ -0,0 +1,23 @@
+package types
+
+// PodStatusResponseInfo represents the container status reply - crictl ps <containerID>
+type PodStatusResponseInfo struct {
+	SandboxID   string
+	RunTimeSpec RunTimeSpecInfo
+}
+
+// RunTimeSpecInfo represents the relevant part of the container status spec
+type RunTimeSpecInfo struct {
+	Linux NamespacesInfo
+}
+
+// NamespacesInfo represents the container status namespaces
+type NamespacesInfo struct {
+	Namespaces []NameSpaceInfo
+}
+
+// NameSpaceInfo represents the ns info
+type NameSpaceInfo struct {
+	Type string
+	Path string
+}