Merge pull request #14525 from dcbw/cni

CNI network plugin

pkg/kubelet/dockertools/manager.go

@@ -74,6 +74,8 @@ const (
 	kubernetesPodLabel                    = "io.kubernetes.pod.data"
 	kubernetesTerminationGracePeriodLabel = "io.kubernetes.pod.terminationGracePeriod"
 	kubernetesContainerLabel              = "io.kubernetes.container.name"
+
+	DockerNetnsFmt = "/proc/%v/ns/net"
 )
 
 // DockerManager implements the Runtime interface.
@@ -1190,6 +1192,32 @@ func (dm *DockerManager) PortForward(pod *kubecontainer.Pod, port uint16, stream
 	return command.Run()
 }
 
+// Get the IP address of a container's interface using nsenter
+func (dm *DockerManager) GetContainerIP(containerID, interfaceName string) (string, error) {
+	_, lookupErr := exec.LookPath("nsenter")
+	if lookupErr != nil {
+		return "", fmt.Errorf("Unable to obtain IP address of container: missing nsenter.")
+	}
+	container, err := dm.client.InspectContainer(containerID)
+	if err != nil {
+		return "", err
+	}
+
+	if !container.State.Running {
+		return "", fmt.Errorf("container not running (%s)", container.ID)
+	}
+
+	containerPid := container.State.Pid
+	extractIPCmd := fmt.Sprintf("ip -4 addr show %s | grep inet | awk -F\" \" '{print $2}'", interfaceName)
+	args := []string{"-t", fmt.Sprintf("%d", containerPid), "-n", "--", "bash", "-c", extractIPCmd}
+	command := exec.Command("nsenter", args...)
+	out, err := command.CombinedOutput()
+	if err != nil {
+		return "", err
+	}
+	return string(out), nil
+}
+
 // Kills all containers in the specified pod
 func (dm *DockerManager) KillPod(pod *api.Pod, runningPod kubecontainer.Pod) error {
 	// Send the kills in parallel since they may take a long time. Len + 1 since there
@@ -1535,6 +1563,10 @@ func (dm *DockerManager) createPodInfraContainer(pod *api.Pod) (kubeletTypes.Doc
 	netNamespace := ""
 	var ports []api.ContainerPort
 
+	if dm.networkPlugin.Name() == "cni" {
+		netNamespace = "none"
+	}
+
 	if pod.Spec.HostNetwork {
 		netNamespace = "host"
 	} else {
@@ -1949,3 +1981,14 @@ func getIPCMode(pod *api.Pod) string {
 	}
 	return ipcMode
 }
+
+// GetNetNs returns the network namespace path for the given container
+func (dm *DockerManager) GetNetNs(containerID string) (string, error) {
+	inspectResult, err := dm.client.InspectContainer(string(containerID))
+	if err != nil {
+		glog.Errorf("Error inspecting container: '%v'", err)
+		return "", err
+	}
+	netnsPath := fmt.Sprintf(DockerNetnsFmt, inspectResult.State.Pid)
+	return netnsPath, nil
+}

pkg/kubelet/network/cni/cni.go

@@ -0,0 +1,204 @@
+/*
+Copyright 2014 The Kubernetes Authors All rights reserved.
+
+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 cni
+
+import (
+	"fmt"
+	"github.com/appc/cni/libcni"
+	cniTypes "github.com/appc/cni/pkg/types"
+	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
+	"k8s.io/kubernetes/pkg/kubelet/network"
+	kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
+	"net"
+	"sort"
+	"strings"
+)
+
+const (
+	CNIPluginName        = "cni"
+	DefaultNetDir        = "/etc/cni/net.d"
+	DefaultCNIDir        = "/opt/cni/bin"
+	DefaultInterfaceName = "eth0"
+	VendorCNIDirTemplate = "%s/opt/%s/bin"
+)
+
+type cniNetworkPlugin struct {
+	defaultNetwork *cniNetwork
+	host           network.Host
+}
+
+type cniNetwork struct {
+	name          string
+	NetworkConfig *libcni.NetworkConfig
+	CNIConfig     *libcni.CNIConfig
+}
+
+func probeNetworkPluginsWithVendorCNIDirPrefix(pluginDir, vendorCNIDirPrefix string) []network.NetworkPlugin {
+	configList := make([]network.NetworkPlugin, 0)
+	network, err := getDefaultCNINetwork(pluginDir, vendorCNIDirPrefix)
+	if err != nil {
+		return configList
+	}
+	return append(configList, &cniNetworkPlugin{defaultNetwork: network})
+}
+
+func ProbeNetworkPlugins(pluginDir string) []network.NetworkPlugin {
+	return probeNetworkPluginsWithVendorCNIDirPrefix(pluginDir, "")
+}
+
+func getDefaultCNINetwork(pluginDir, vendorCNIDirPrefix string) (*cniNetwork, error) {
+	if pluginDir == "" {
+		pluginDir = DefaultNetDir
+	}
+	files, err := libcni.ConfFiles(pluginDir)
+	switch {
+	case err != nil:
+		return nil, err
+	case len(files) == 0:
+		return nil, fmt.Errorf("No networks found in %s", pluginDir)
+	}
+
+	sort.Strings(files)
+	for _, confFile := range files {
+		conf, err := libcni.ConfFromFile(confFile)
+		if err != nil {
+			glog.Warningf("Error loading CNI config file %s: %v", confFile, err)
+			continue
+		}
+		// Search for vendor-specific plugins as well as default plugins in the CNI codebase.
+		vendorCNIDir := fmt.Sprintf(VendorCNIDirTemplate, vendorCNIDirPrefix, conf.Network.Type)
+		cninet := &libcni.CNIConfig{
+			Path: []string{DefaultCNIDir, vendorCNIDir},
+		}
+		network := &cniNetwork{name: conf.Network.Name, NetworkConfig: conf, CNIConfig: cninet}
+		return network, nil
+	}
+	return nil, fmt.Errorf("No valid networks found in %s", pluginDir)
+}
+
+func (plugin *cniNetworkPlugin) Init(host network.Host) error {
+	plugin.host = host
+	return nil
+}
+
+func (plugin *cniNetworkPlugin) Name() string {
+	return CNIPluginName
+}
+
+func (plugin *cniNetworkPlugin) SetUpPod(namespace string, name string, id kubeletTypes.DockerID) error {
+	runtime, ok := plugin.host.GetRuntime().(*dockertools.DockerManager)
+	if !ok {
+		return fmt.Errorf("CNI execution called on non-docker runtime")
+	}
+	netns, err := runtime.GetNetNs(string(id))
+	if err != nil {
+		return err
+	}
+
+	_, err = plugin.defaultNetwork.addToNetwork(name, namespace, string(id), netns)
+	if err != nil {
+		glog.Errorf("Error while adding to cni network: %s", err)
+		return err
+	}
+
+	return err
+}
+
+func (plugin *cniNetworkPlugin) TearDownPod(namespace string, name string, id kubeletTypes.DockerID) error {
+	runtime, ok := plugin.host.GetRuntime().(*dockertools.DockerManager)
+	if !ok {
+		return fmt.Errorf("CNI execution called on non-docker runtime")
+	}
+	netns, err := runtime.GetNetNs(string(id))
+	if err != nil {
+		return err
+	}
+
+	return plugin.defaultNetwork.deleteFromNetwork(name, namespace, string(id), netns)
+}
+
+// TODO: Use the addToNetwork function to obtain the IP of the Pod. That will assume idempotent ADD call to the plugin.
+// Also fix the runtime's call to Status function to be done only in the case that the IP is lost, no need to do periodic calls
+func (plugin *cniNetworkPlugin) Status(namespace string, name string, id kubeletTypes.DockerID) (*network.PodNetworkStatus, error) {
+	runtime, ok := plugin.host.GetRuntime().(*dockertools.DockerManager)
+	if !ok {
+		return nil, fmt.Errorf("CNI execution called on non-docker runtime")
+	}
+	ipStr, err := runtime.GetContainerIP(string(id), DefaultInterfaceName)
+	if err != nil {
+		return nil, err
+	}
+	ip, _, err := net.ParseCIDR(strings.Trim(ipStr, "\n"))
+	if err != nil {
+		return nil, err
+	}
+	return &network.PodNetworkStatus{IP: ip}, nil
+}
+
+func (network *cniNetwork) addToNetwork(podName string, podNamespace string, podInfraContainerID string, podNetnsPath string) (*cniTypes.Result, error) {
+	rt, err := buildCNIRuntimeConf(podName, podNamespace, podInfraContainerID, podNetnsPath)
+	if err != nil {
+		glog.Errorf("Error adding network: %v", err)
+		return nil, err
+	}
+
+	netconf, cninet := network.NetworkConfig, network.CNIConfig
+	glog.V(4).Infof("About to run with conf.Network.Type=%v, c.Path=%v", netconf.Network.Type, cninet.Path)
+	res, err := cninet.AddNetwork(netconf, rt)
+	if err != nil {
+		glog.Errorf("Error adding network: %v", err)
+		return nil, err
+	}
+
+	return res, nil
+}
+
+func (network *cniNetwork) deleteFromNetwork(podName string, podNamespace string, podInfraContainerID string, podNetnsPath string) error {
+	rt, err := buildCNIRuntimeConf(podName, podNamespace, podInfraContainerID, podNetnsPath)
+	if err != nil {
+		glog.Errorf("Error deleting network: %v", err)
+		return err
+	}
+
+	netconf, cninet := network.NetworkConfig, network.CNIConfig
+	glog.V(4).Infof("About to run with conf.Network.Type=%v, c.Path=%v", netconf.Network.Type, cninet.Path)
+	err = cninet.DelNetwork(netconf, rt)
+	if err != nil {
+		glog.Errorf("Error deleting network: %v", err)
+		return err
+	}
+	return nil
+}
+
+func buildCNIRuntimeConf(podName string, podNs string, podInfraContainerID string, podNetnsPath string) (*libcni.RuntimeConf, error) {
+	glog.V(4).Infof("Got netns path %v", podNetnsPath)
+	glog.V(4).Infof("Using netns path %v", podNs)
+
+	rt := &libcni.RuntimeConf{
+		ContainerID: podInfraContainerID,
+		NetNS:       podNetnsPath,
+		IfName:      DefaultInterfaceName,
+		Args: [][2]string{
+			{"K8S_POD_NAMESPACE", podNs},
+			{"K8S_POD_NAME", podName},
+			{"K8S_POD_INFRA_CONTAINER_ID", podInfraContainerID},
+		},
+	}
+
+	return rt, nil
+}

pkg/kubelet/network/cni/cni_test.go

@@ -0,0 +1,196 @@
+// +build linux
+
+/*
+Copyright 2014 The Kubernetes Authors All rights reserved.
+
+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 cni
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"os"
+	"path"
+	"testing"
+	"text/template"
+
+	docker "github.com/fsouza/go-dockerclient"
+	cadvisorApi "github.com/google/cadvisor/info/v1"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/client/record"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
+	"k8s.io/kubernetes/pkg/kubelet/network"
+	"k8s.io/kubernetes/pkg/util/sets"
+)
+
+// The temp dir where test plugins will be stored.
+const testNetworkConfigPath = "/tmp/fake/plugins/net/cni"
+const testVendorCNIDirPrefix = "/tmp"
+
+func installPluginUnderTest(t *testing.T, vendorName string, plugName string) {
+	pluginDir := path.Join(testNetworkConfigPath, plugName)
+	err := os.MkdirAll(pluginDir, 0777)
+	if err != nil {
+		t.Fatalf("Failed to create plugin config dir: %v", err)
+	}
+	pluginConfig := path.Join(pluginDir, plugName+".conf")
+	f, err := os.Create(pluginConfig)
+	if err != nil {
+		t.Fatalf("Failed to install plugin")
+	}
+	networkConfig := fmt.Sprintf("{ \"name\": \"%s\", \"type\": \"%s\" }", plugName, vendorName)
+
+	_, err = f.WriteString(networkConfig)
+	if err != nil {
+		t.Fatalf("Failed to write network config file (%v)", err)
+	}
+	f.Close()
+
+	vendorCNIDir := fmt.Sprintf(VendorCNIDirTemplate, testVendorCNIDirPrefix, vendorName)
+	err = os.MkdirAll(vendorCNIDir, 0777)
+	if err != nil {
+		t.Fatalf("Failed to create plugin dir: %v", err)
+	}
+	pluginExec := path.Join(vendorCNIDir, vendorName)
+	f, err = os.Create(pluginExec)
+
+	const execScriptTempl = `#!/bin/bash
+read ignore
+export $(echo ${CNI_ARGS} | sed 's/;/ /g') &> /dev/null
+mkdir -p {{.OutputDir}} &> /dev/null
+echo -n "$CNI_COMMAND $CNI_NETNS $K8S_POD_NAMESPACE $K8S_POD_NAME $K8S_POD_INFRA_CONTAINER_ID" >& {{.OutputFile}}
+echo -n "{ \"ip4\": { \"ip\": \"10.1.0.23/24\" } }"
+`
+	execTemplateData := &map[string]interface{}{
+		"OutputFile": path.Join(pluginDir, plugName+".out"),
+		"OutputDir":  pluginDir,
+	}
+
+	tObj := template.Must(template.New("test").Parse(execScriptTempl))
+	buf := &bytes.Buffer{}
+	if err := tObj.Execute(buf, *execTemplateData); err != nil {
+		t.Fatalf("Error in executing script template - %v", err)
+	}
+	execScript := buf.String()
+	_, err = f.WriteString(execScript)
+	if err != nil {
+		t.Fatalf("Failed to write plugin exec - %v", err)
+	}
+
+	err = f.Chmod(0777)
+	if err != nil {
+		t.Fatalf("Failed to set exec perms on plugin")
+	}
+
+	f.Close()
+}
+
+func tearDownPlugin(plugName string, vendorName string) {
+	err := os.RemoveAll(testNetworkConfigPath)
+	if err != nil {
+		fmt.Printf("Error in cleaning up test: %v", err)
+	}
+	vendorCNIDir := fmt.Sprintf(VendorCNIDirTemplate, testVendorCNIDirPrefix, vendorName)
+	err = os.RemoveAll(vendorCNIDir)
+	if err != nil {
+		fmt.Printf("Error in cleaning up test: %v", err)
+	}
+}
+
+type fakeNetworkHost struct {
+	kubeClient client.Interface
+}
+
+func NewFakeHost(kubeClient client.Interface) *fakeNetworkHost {
+	host := &fakeNetworkHost{kubeClient: kubeClient}
+	return host
+}
+
+func (fnh *fakeNetworkHost) GetPodByName(name, namespace string) (*api.Pod, bool) {
+	return nil, false
+}
+
+func (fnh *fakeNetworkHost) GetKubeClient() client.Interface {
+	return nil
+}
+
+func (nh *fakeNetworkHost) GetRuntime() kubecontainer.Runtime {
+	dm, fakeDockerClient := newTestDockerManager()
+	fakeDockerClient.Container = &docker.Container{
+		ID:    "foobar",
+		State: docker.State{Pid: 12345},
+	}
+	return dm
+}
+
+func newTestDockerManager() (*dockertools.DockerManager, *dockertools.FakeDockerClient) {
+	fakeDocker := &dockertools.FakeDockerClient{VersionInfo: docker.Env{"Version=1.1.3", "ApiVersion=1.15"}, Errors: make(map[string]error), RemovedImages: sets.String{}}
+	fakeRecorder := &record.FakeRecorder{}
+	readinessManager := kubecontainer.NewReadinessManager()
+	containerRefManager := kubecontainer.NewRefManager()
+	networkPlugin, _ := network.InitNetworkPlugin([]network.NetworkPlugin{}, "", network.NewFakeHost(nil))
+	dockerManager := dockertools.NewFakeDockerManager(
+		fakeDocker,
+		fakeRecorder,
+		readinessManager,
+		containerRefManager,
+		&cadvisorApi.MachineInfo{},
+		dockertools.PodInfraContainerImage,
+		0, 0, "",
+		kubecontainer.FakeOS{},
+		networkPlugin,
+		nil,
+		nil)
+
+	return dockerManager, fakeDocker
+}
+
+func TestCNIPlugin(t *testing.T) {
+	// install some random plugin
+	pluginName := fmt.Sprintf("test%d", rand.Intn(1000))
+	vendorName := fmt.Sprintf("test_vendor%d", rand.Intn(1000))
+	defer tearDownPlugin(pluginName, vendorName)
+	installPluginUnderTest(t, vendorName, pluginName)
+
+	np := probeNetworkPluginsWithVendorCNIDirPrefix(path.Join(testNetworkConfigPath, pluginName), testVendorCNIDirPrefix)
+	plug, err := network.InitNetworkPlugin(np, "cni", NewFakeHost(nil))
+	if err != nil {
+		t.Fatalf("Failed to select the desired plugin: %v", err)
+	}
+
+	err = plug.SetUpPod("podNamespace", "podName", "dockerid2345")
+	if err != nil {
+		t.Errorf("Expected nil: %v", err)
+	}
+	output, err := ioutil.ReadFile(path.Join(testNetworkConfigPath, pluginName, pluginName+".out"))
+	expectedOutput := "ADD /proc/12345/ns/net podNamespace podName dockerid2345"
+	if string(output) != expectedOutput {
+		t.Errorf("Mismatch in expected output for setup hook. Expected '%s', got '%s'", expectedOutput, string(output))
+	}
+	err = plug.TearDownPod("podNamespace", "podName", "dockerid4545454")
+	if err != nil {
+		t.Errorf("Expected nil: %v", err)
+	}
+	output, err = ioutil.ReadFile(path.Join(testNetworkConfigPath, pluginName, pluginName+".out"))
+	expectedOutput = "DEL /proc/12345/ns/net podNamespace podName dockerid4545454"
+	if string(output) != expectedOutput {
+		t.Errorf("Mismatch in expected output for setup hook. Expected '%s', got '%s'", expectedOutput, string(output))
+	}
+}

pkg/kubelet/network/plugins.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/util/errors"
 	"k8s.io/kubernetes/pkg/util/validation"
@@ -73,6 +74,9 @@ type Host interface {
 
 	// GetKubeClient returns a client interface
 	GetKubeClient() client.Interface
+
+	// GetContainerRuntime returns the container runtime that implements the containers (e.g. docker/rkt)
+	GetRuntime() kubecontainer.Runtime
 }
 
 // InitNetworkPlugin inits the plugin that matches networkPluginName. Plugins must have unique names.

pkg/kubelet/network/testing.go

@@ -22,6 +22,7 @@ package network
 import (
 	"k8s.io/kubernetes/pkg/api"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 )
 
 type fakeNetworkHost struct {
@@ -40,3 +41,7 @@ func (fnh *fakeNetworkHost) GetPodByName(name, namespace string) (*api.Pod, bool
 func (fnh *fakeNetworkHost) GetKubeClient() client.Interface {
 	return nil
 }
+
+func (nh *fakeNetworkHost) GetRuntime() kubecontainer.Runtime {
+	return &kubecontainer.FakeRuntime{}
+}

pkg/kubelet/networks.go

@@ -19,6 +19,7 @@ package kubelet
 import (
 	"k8s.io/kubernetes/pkg/api"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 )
 
 // This just exports required functions from kubelet proper, for use by network
@@ -34,3 +35,7 @@ func (nh *networkHost) GetPodByName(name, namespace string) (*api.Pod, bool) {
 func (nh *networkHost) GetKubeClient() client.Interface {
 	return nh.kubelet.kubeClient
 }
+
+func (nh *networkHost) GetRuntime() kubecontainer.Runtime {
+	return nh.kubelet.GetRuntime()
+}