Merge pull request #25062 from dcbw/kubenet-rkt

pkg/kubelet/container/runtime.go

@@ -97,6 +97,12 @@ type Runtime interface {
 	RemoveImage(image ImageSpec) error
 	// Returns Image statistics.
 	ImageStats() (*ImageStats, error)
+	// Returns the filesystem path of the pod's network namespace; if the
+	// runtime does not handle namespace creation itself, or cannot return
+	// the network namespace path, it should return an error.
+	// TODO: Change ContainerID to a Pod ID since the namespace is shared
+	// by all containers in the pod.
+	GetNetNS(containerID ContainerID) (string, error)
 	// TODO(vmarmol): Unify pod and containerID args.
 	// GetContainerLogs returns logs of a specific container. By
 	// default, it returns a snapshot of the container log. Set 'follow' to true to

pkg/kubelet/container/testing/fake_runtime.go

@@ -338,6 +338,14 @@ func (f *FakeRuntime) PortForward(pod *Pod, port uint16, stream io.ReadWriteClos
 	return f.Err
 }
 
+func (f *FakeRuntime) GetNetNS(containerID ContainerID) (string, error) {
+	f.Lock()
+	defer f.Unlock()
+
+	f.CalledFunctions = append(f.CalledFunctions, "GetNetNS")
+	return "", f.Err
+}
+
 func (f *FakeRuntime) GarbageCollect(gcPolicy ContainerGCPolicy) error {
 	f.Lock()
 	defer f.Unlock()

pkg/kubelet/container/testing/runtime_mock.go

@@ -128,6 +128,11 @@ func (r *Mock) PortForward(pod *Pod, port uint16, stream io.ReadWriteCloser) err
 	return args.Error(0)
 }
 
+func (r *Mock) GetNetNS(containerID ContainerID) (string, error) {
+	args := r.Called(containerID)
+	return "", args.Error(0)
+}
+
 func (r *Mock) GarbageCollect(gcPolicy ContainerGCPolicy) error {
 	args := r.Called(gcPolicy)
 	return args.Error(0)

pkg/kubelet/dockertools/manager.go

@@ -1959,7 +1959,7 @@ func (dm *DockerManager) SyncPod(pod *api.Pod, _ api.PodStatus, podStatus *kubec
 			}
 
 			if dm.configureHairpinMode {
-				if err = hairpin.SetUpContainer(podInfraContainer.State.Pid, network.DefaultInterfaceName); err != nil {
+				if err = hairpin.SetUpContainerPid(podInfraContainer.State.Pid, network.DefaultInterfaceName); err != nil {
 					glog.Warningf("Hairpin setup failed for pod %q: %v", format.Pod(pod), err)
 				}
 			}

pkg/kubelet/kubelet.go

@@ -439,6 +439,8 @@ func NewMainKubelet(
 			klet.livenessManager,
 			klet.volumeManager,
 			klet.httpClient,
+			klet.networkPlugin,
+			klet.hairpinMode == componentconfig.HairpinVeth,
 			utilexec.New(),
 			kubecontainer.RealOS{},
 			imageBackOff,

pkg/kubelet/network/hairpin/hairpin.go

@@ -40,13 +40,23 @@ var (
 	ethtoolOutputRegex = regexp.MustCompile("peer_ifindex: (\\d+)")
 )
 
-func SetUpContainer(containerPid int, containerInterfaceName string) error {
-	e := exec.New()
-	return setUpContainerInternal(e, containerPid, containerInterfaceName)
+func SetUpContainerPid(containerPid int, containerInterfaceName string) error {
+	pidStr := fmt.Sprintf("%d", containerPid)
+	nsenterArgs := []string{"-t", pidStr, "-n"}
+	return setUpContainerInternal(containerInterfaceName, pidStr, nsenterArgs)
+}
+
+func SetUpContainerPath(netnsPath string, containerInterfaceName string) error {
+	if netnsPath[0] != '/' {
+		return fmt.Errorf("netnsPath path '%s' was invalid", netnsPath)
+	}
+	nsenterArgs := []string{"-n", netnsPath}
+	return setUpContainerInternal(containerInterfaceName, netnsPath, nsenterArgs)
 }
 
-func setUpContainerInternal(e exec.Interface, containerPid int, containerInterfaceName string) error {
-	hostIfName, err := findPairInterfaceOfContainerInterface(e, containerPid, containerInterfaceName)
+func setUpContainerInternal(containerInterfaceName, containerDesc string, nsenterArgs []string) error {
+	e := exec.New()
+	hostIfName, err := findPairInterfaceOfContainerInterface(e, containerInterfaceName, containerDesc, nsenterArgs)
 	if err != nil {
 		glog.Infof("Unable to find pair interface, setting up all interfaces: %v", err)
 		return setUpAllInterfaces()
@@ -54,7 +64,7 @@ func setUpContainerInternal(e exec.Interface, containerPid int, containerInterfa
 	return setUpInterface(hostIfName)
 }
 
-func findPairInterfaceOfContainerInterface(e exec.Interface, containerPid int, containerInterfaceName string) (string, error) {
+func findPairInterfaceOfContainerInterface(e exec.Interface, containerInterfaceName, containerDesc string, nsenterArgs []string) (string, error) {
 	nsenterPath, err := e.LookPath("nsenter")
 	if err != nil {
 		return "", err
@@ -63,15 +73,16 @@ func findPairInterfaceOfContainerInterface(e exec.Interface, containerPid int, c
 	if err != nil {
 		return "", err
 	}
-	// Get container's interface index
-	output, err := e.Command(nsenterPath, "-t", fmt.Sprintf("%d", containerPid), "-n", "-F", "--", ethtoolPath, "--statistics", containerInterfaceName).CombinedOutput()
+
+	nsenterArgs = append(nsenterArgs, "-F", "--", ethtoolPath, "--statistics", containerInterfaceName)
+	output, err := e.Command(nsenterPath, nsenterArgs...).CombinedOutput()
 	if err != nil {
-		return "", fmt.Errorf("Unable to query interface %s of container %d: %v: %s", containerInterfaceName, containerPid, err, string(output))
+		return "", fmt.Errorf("Unable to query interface %s of container %s: %v: %s", containerInterfaceName, containerDesc, err, string(output))
 	}
 	// look for peer_ifindex
 	match := ethtoolOutputRegex.FindSubmatch(output)
 	if match == nil {
-		return "", fmt.Errorf("No peer_ifindex in interface statistics for %s of container %d", containerInterfaceName, containerPid)
+		return "", fmt.Errorf("No peer_ifindex in interface statistics for %s of container %s", containerInterfaceName, containerDesc)
 	}
 	peerIfIndex, err := strconv.Atoi(string(match[1]))
 	if err != nil { // seems impossible (\d+ not numeric)

pkg/kubelet/network/hairpin/hairpin_test.go

@@ -69,7 +69,8 @@ func TestFindPairInterfaceOfContainerInterface(t *testing.T) {
 				return fmt.Sprintf("/fake-bin/%s", file), nil
 			},
 		}
-		name, err := findPairInterfaceOfContainerInterface(&fexec, 123, "eth0")
+		nsenterArgs := []string{"-t", "123", "-n"}
+		name, err := findPairInterfaceOfContainerInterface(&fexec, "eth0", "123", nsenterArgs)
 		if test.expectErr {
 			if err == nil {
 				t.Errorf("unexpected non-error")

pkg/kubelet/network/kubenet/kubenet_linux.go

@@ -31,10 +31,10 @@ import (
 	"github.com/vishvananda/netlink/nl"
 
 	"github.com/appc/cni/libcni"
+	cnitypes "github.com/appc/cni/pkg/types"
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/apis/componentconfig"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
-	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	"k8s.io/kubernetes/pkg/util/bandwidth"
 	utilexec "k8s.io/kubernetes/pkg/util/exec"
@@ -55,6 +55,7 @@ type kubenetNetworkPlugin struct {
 
 	host        network.Host
 	netConfig   *libcni.NetworkConfig
+	loConfig    *libcni.NetworkConfig
 	cniConfig   *libcni.CNIConfig
 	shaper      bandwidth.BandwidthShaper
 	podCIDRs    map[kubecontainer.ContainerID]string
@@ -94,10 +95,20 @@ func (plugin *kubenetNetworkPlugin) Init(host network.Host, hairpinMode componen
 	// was built-in, we simply ignore the error here. A better thing to do is
 	// to check the kernel version in the future.
 	plugin.execer.Command("modprobe", "br-netfilter").CombinedOutput()
-	if err := utilsysctl.SetSysctl(sysctlBridgeCallIptables, 1); err != nil {
+	err := utilsysctl.SetSysctl(sysctlBridgeCallIptables, 1)
+	if err != nil {
 		glog.Warningf("can't set sysctl %s: %v", sysctlBridgeCallIptables, err)
 	}
 
+	plugin.loConfig, err = libcni.ConfFromBytes([]byte(`{
+  "cniVersion": "0.1.0",
+  "name": "kubenet-loopback",
+  "type": "loopback"
+}`))
+	if err != nil {
+		return fmt.Errorf("Failed to generate loopback config: %v", err)
+	}
+
 	return nil
 }
 
@@ -250,7 +261,7 @@ func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id k
 
 	start := time.Now()
 	defer func() {
-		glog.V(4).Infof("TearDownPod took %v for %s/%s", time.Since(start), namespace, name)
+		glog.V(4).Infof("SetUpPod took %v for %s/%s", time.Since(start), namespace, name)
 	}()
 
 	pod, ok := plugin.host.GetPodByName(namespace, name)
@@ -266,23 +277,20 @@ func (plugin *kubenetNetworkPlugin) SetUpPod(namespace string, name string, id k
 		return fmt.Errorf("Kubenet cannot SetUpPod: %v", err)
 	}
 
-	runtime, ok := plugin.host.GetRuntime().(*dockertools.DockerManager)
-	if !ok {
-		return fmt.Errorf("Kubenet execution called on non-docker runtime")
-	}
-	netnsPath, err := runtime.GetNetNS(id)
-	if err != nil {
-		return fmt.Errorf("Kubenet failed to retrieve network namespace path: %v", err)
+	// Bring up container loopback interface
+	if _, err := plugin.addContainerToNetwork(plugin.loConfig, "lo", namespace, name, id); err != nil {
+		return err
 	}
 
-	rt := buildCNIRuntimeConf(name, namespace, id, netnsPath)
+	// Hook container up with our bridge
+	res, err := plugin.addContainerToNetwork(plugin.netConfig, network.DefaultInterfaceName, namespace, name, id)
 	if err != nil {
-		return fmt.Errorf("Error building CNI config: %v", err)
-	}
-
-	if err = plugin.addContainerToNetwork(id, rt); err != nil {
 		return err
 	}
+	if res.IP4 == nil || res.IP4.IP.String() == "" {
+		return fmt.Errorf("CNI plugin reported no IPv4 address for container %v.", id)
+	}
+	plugin.podCIDRs[id] = res.IP4.IP.String()
 
 	// Put the container bridge into promiscuous mode to force it to accept hairpin packets.
 	// TODO: Remove this once the kernel bug (#20096) is fixed.
@@ -330,20 +338,6 @@ func (plugin *kubenetNetworkPlugin) TearDownPod(namespace string, name string, i
 		return fmt.Errorf("Kubenet needs a PodCIDR to tear down pods")
 	}
 
-	runtime, ok := plugin.host.GetRuntime().(*dockertools.DockerManager)
-	if !ok {
-		return fmt.Errorf("Kubenet execution called on non-docker runtime")
-	}
-	netnsPath, err := runtime.GetNetNS(id)
-	if err != nil {
-		return err
-	}
-
-	rt := buildCNIRuntimeConf(name, namespace, id, netnsPath)
-	if err != nil {
-		return fmt.Errorf("Error building CNI config: %v", err)
-	}
-
 	// no cached CIDR is Ok during teardown
 	if cidr, ok := plugin.podCIDRs[id]; ok {
 		glog.V(5).Infof("Removing pod CIDR %s from shaper", cidr)
@@ -354,9 +348,10 @@ func (plugin *kubenetNetworkPlugin) TearDownPod(namespace string, name string, i
 			}
 		}
 	}
-	if err = plugin.delContainerFromNetwork(id, rt); err != nil {
+	if err := plugin.delContainerFromNetwork(plugin.netConfig, network.DefaultInterfaceName, namespace, name, id); err != nil {
 		return err
 	}
+	delete(plugin.podCIDRs, id)
 
 	return nil
 }
@@ -373,12 +368,8 @@ func (plugin *kubenetNetworkPlugin) GetPodNetworkStatus(namespace string, name s
 			return &network.PodNetworkStatus{IP: ip}, nil
 		}
 	}
-	// TODO: remove type conversion once kubenet supports multiple runtime
-	runtime, ok := plugin.host.GetRuntime().(*dockertools.DockerManager)
-	if !ok {
-		return nil, fmt.Errorf("Kubenet execution called on non-docker runtime")
-	}
-	netnsPath, err := runtime.GetNetNS(id)
+
+	netnsPath, err := plugin.host.GetRuntime().GetNetNS(id)
 	if err != nil {
 		return nil, fmt.Errorf("Kubenet failed to retrieve network namespace path: %v", err)
 	}
@@ -412,37 +403,43 @@ func (plugin *kubenetNetworkPlugin) Status() error {
 	return nil
 }
 
-func buildCNIRuntimeConf(podName string, podNs string, podInfraContainerID kubecontainer.ContainerID, podNetnsPath string) *libcni.RuntimeConf {
-	glog.V(4).Infof("Kubenet: using netns path %v", podNetnsPath)
-	glog.V(4).Infof("Kubenet: using podns path %v", podNs)
+func (plugin *kubenetNetworkPlugin) buildCNIRuntimeConf(ifName string, id kubecontainer.ContainerID) (*libcni.RuntimeConf, error) {
+	netnsPath, err := plugin.host.GetRuntime().GetNetNS(id)
+	if err != nil {
+		return nil, fmt.Errorf("Kubenet failed to retrieve network namespace path: %v", err)
+	}
 
 	return &libcni.RuntimeConf{
-		ContainerID: podInfraContainerID.ID,
-		NetNS:       podNetnsPath,
-		IfName:      network.DefaultInterfaceName,
-	}
+		ContainerID: id.ID,
+		NetNS:       netnsPath,
+		IfName:      ifName,
+	}, nil
 }
 
-func (plugin *kubenetNetworkPlugin) addContainerToNetwork(id kubecontainer.ContainerID, rt *libcni.RuntimeConf) error {
-	glog.V(3).Infof("Calling cni plugins to add container to network with cni runtime: %+v", rt)
-	res, err := plugin.cniConfig.AddNetwork(plugin.netConfig, rt)
+func (plugin *kubenetNetworkPlugin) addContainerToNetwork(config *libcni.NetworkConfig, ifName, namespace, name string, id kubecontainer.ContainerID) (*cnitypes.Result, error) {
+	rt, err := plugin.buildCNIRuntimeConf(ifName, id)
 	if err != nil {
-		return fmt.Errorf("Error adding container to network: %v", err)
-	}
-	if res.IP4 == nil || res.IP4.IP.String() == "" {
-		return fmt.Errorf("CNI plugin reported no IPv4 address for container %v.", id)
+		return nil, fmt.Errorf("Error building CNI config: %v", err)
 	}
 
-	plugin.podCIDRs[id] = res.IP4.IP.String()
-	return nil
+	glog.V(3).Infof("Adding %s/%s to '%s' with CNI '%s' plugin and runtime: %+v", namespace, name, config.Network.Name, config.Network.Type, rt)
+	res, err := plugin.cniConfig.AddNetwork(config, rt)
+	if err != nil {
+		return nil, fmt.Errorf("Error adding container to network: %v", err)
+	}
+	return res, nil
 }
 
-func (plugin *kubenetNetworkPlugin) delContainerFromNetwork(id kubecontainer.ContainerID, rt *libcni.RuntimeConf) error {
-	glog.V(3).Infof("Calling cni plugins to remove container from network with cni runtime: %+v", rt)
-	if err := plugin.cniConfig.DelNetwork(plugin.netConfig, rt); err != nil {
+func (plugin *kubenetNetworkPlugin) delContainerFromNetwork(config *libcni.NetworkConfig, ifName, namespace, name string, id kubecontainer.ContainerID) error {
+	rt, err := plugin.buildCNIRuntimeConf(ifName, id)
+	if err != nil {
+		return fmt.Errorf("Error building CNI config: %v", err)
+	}
+
+	glog.V(3).Infof("Removing %s/%s from '%s' with CNI '%s' plugin and runtime: %+v", namespace, name, config.Network.Name, config.Network.Type, rt)
+	if err := plugin.cniConfig.DelNetwork(config, rt); err != nil {
 		return fmt.Errorf("Error removing container from network: %v", err)
 	}
-	delete(plugin.podCIDRs, id)
 	return nil
 }
 

pkg/kubelet/network/kubenet/kubenet_linux_test.go

@@ -17,6 +17,8 @@ limitations under the License.
 package kubenet
 
 import (
+	"fmt"
+
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	nettest "k8s.io/kubernetes/pkg/kubelet/network/testing"
@@ -38,9 +40,6 @@ func TestGetPodNetworkStatus(t *testing.T) {
 	podIPMap[kubecontainer.ContainerID{ID: "1"}] = "10.245.0.2/32"
 	podIPMap[kubecontainer.ContainerID{ID: "2"}] = "10.245.0.3/32"
 
-	fhost := nettest.NewFakeHost(nil)
-	fakeKubenet := newFakeKubenetPlugin(podIPMap, nil, fhost)
-
 	testCases := []struct {
 		id          string
 		expectError bool
@@ -66,6 +65,34 @@ func TestGetPodNetworkStatus(t *testing.T) {
 		//TODO: add test cases for retrieving ip inside container network namespace
 	}
 
+	fakeCmds := make([]exec.FakeCommandAction, 0)
+	for _, t := range testCases {
+		// the fake commands return the IP from the given index, or an error
+		fCmd := exec.FakeCmd{
+			CombinedOutputScript: []exec.FakeCombinedOutputAction{
+				func() ([]byte, error) {
+					ip, ok := podIPMap[kubecontainer.ContainerID{ID: t.id}]
+					if !ok {
+						return nil, fmt.Errorf("Pod IP %q not found", t.id)
+					}
+					return []byte(ip), nil
+				},
+			},
+		}
+		fakeCmds = append(fakeCmds, func(cmd string, args ...string) exec.Cmd {
+			return exec.InitFakeCmd(&fCmd, cmd, args...)
+		})
+	}
+	fexec := exec.FakeExec{
+		CommandScript: fakeCmds,
+		LookPathFunc: func(file string) (string, error) {
+			return fmt.Sprintf("/fake-bin/%s", file), nil
+		},
+	}
+
+	fhost := nettest.NewFakeHost(nil)
+	fakeKubenet := newFakeKubenetPlugin(podIPMap, &fexec, fhost)
+
 	for i, tc := range testCases {
 		out, err := fakeKubenet.GetPodNetworkStatus("", "", kubecontainer.ContainerID{ID: tc.id})
 		if tc.expectError {

pkg/kubelet/rkt/rkt_test.go

@@ -1078,6 +1078,7 @@ func TestGenerateRunCommand(t *testing.T) {
 	tests := []struct {
 		pod       *api.Pod
 		uuid      string
+		netnsName string
 
 		dnsServers  []string
 		dnsSearches []string
@@ -1095,6 +1096,7 @@ func TestGenerateRunCommand(t *testing.T) {
 				Spec: api.PodSpec{},
 			},
 			"rkt-uuid-foo",
+			"default",
 			[]string{},
 			[]string{},
 			"",
@@ -1109,11 +1111,12 @@ func TestGenerateRunCommand(t *testing.T) {
 				},
 			},
 			"rkt-uuid-foo",
+			"default",
 			[]string{},
 			[]string{},
 			"pod-hostname-foo",
 			nil,
-			"/bin/rkt/rkt --insecure-options=image,ondisk --local-config=/var/rkt/local/data --dir=/var/data run-prepared --net=rkt.kubernetes.io --hostname=pod-hostname-foo rkt-uuid-foo",
+			" --net=\"/var/run/netns/default\" -- /bin/rkt/rkt --insecure-options=image,ondisk --local-config=/var/rkt/local/data --dir=/var/data run-prepared --net=host --hostname=pod-hostname-foo rkt-uuid-foo",
 		},
 		// Case #2, returns no dns, with host-net.
 		{
@@ -1128,6 +1131,7 @@ func TestGenerateRunCommand(t *testing.T) {
 				},
 			},
 			"rkt-uuid-foo",
+			"",
 			[]string{},
 			[]string{},
 			"",
@@ -1147,11 +1151,12 @@ func TestGenerateRunCommand(t *testing.T) {
 				},
 			},
 			"rkt-uuid-foo",
+			"default",
 			[]string{"127.0.0.1"},
 			[]string{"."},
 			"pod-hostname-foo",
 			nil,
-			"/bin/rkt/rkt --insecure-options=image,ondisk --local-config=/var/rkt/local/data --dir=/var/data run-prepared --net=rkt.kubernetes.io --dns=127.0.0.1 --dns-search=. --dns-opt=ndots:5 --hostname=pod-hostname-foo rkt-uuid-foo",
+			" --net=\"/var/run/netns/default\" -- /bin/rkt/rkt --insecure-options=image,ondisk --local-config=/var/rkt/local/data --dir=/var/data run-prepared --net=host --dns=127.0.0.1 --dns-search=. --dns-opt=ndots:5 --hostname=pod-hostname-foo rkt-uuid-foo",
 		},
 		// Case #4, returns no dns, dns searches, with host-network.
 		{
@@ -1166,6 +1171,7 @@ func TestGenerateRunCommand(t *testing.T) {
 				},
 			},
 			"rkt-uuid-foo",
+			"",
 			[]string{"127.0.0.1"},
 			[]string{"."},
 			"pod-hostname-foo",
@@ -1189,7 +1195,7 @@ func TestGenerateRunCommand(t *testing.T) {
 		testCaseHint := fmt.Sprintf("test case #%d", i)
 		rkt.runtimeHelper = &fakeRuntimeHelper{tt.dnsServers, tt.dnsSearches, tt.hostName, "", tt.err}
 
-		result, err := rkt.generateRunCommand(tt.pod, tt.uuid)
+		result, err := rkt.generateRunCommand(tt.pod, tt.uuid, tt.netnsName)
 		assert.Equal(t, tt.err, err, testCaseHint)
 		assert.Equal(t, tt.expect, result, testCaseHint)
 	}