Fix IP affinity to be per-pod, not per-container.

pkg/api/types.go

@@ -57,7 +57,7 @@ type EnvVar struct {
 type Container struct {
 	Name         string        `yaml:"name,omitempty" json:"name,omitempty"`
 	Image        string        `yaml:"image,omitempty" json:"image,omitempty"`
-	Command      string        `yaml:"command,omitempty" json:"command,omitempty"`
+	Command      []string      `yaml:"command,omitempty" json:"command,omitempty"`
 	WorkingDir   string        `yaml:"workingDir,omitempty" json:"workingDir,omitempty"`
 	Ports        []Port        `yaml:"ports,omitempty" json:"ports,omitempty"`
 	Env          []EnvVar      `yaml:"env,omitempty" json:"env,omitempty"`

pkg/kubelet/kubelet.go

@@ -296,15 +296,7 @@ func makePortsAndBindings(container *api.Container) (map[docker.Port]struct{}, m
 	return exposedPorts, portBindings
 }
 
-func makeCommandLine(container *api.Container) []string {
-	var cmdList []string
-	if len(container.Command) > 0 {
-		cmdList = strings.Split(container.Command, " ")
-	}
-	return cmdList
-}
-
-func (kl *Kubelet) RunContainer(manifest *api.ContainerManifest, container *api.Container) (name string, err error) {
+func (kl *Kubelet) RunContainer(manifest *api.ContainerManifest, container *api.Container, netMode string) (name string, err error) {
 	name = manifestAndContainerToDockerName(manifest, container)
 
 	envVariables := makeEnvironmentVariables(container)
@@ -319,7 +311,7 @@ func (kl *Kubelet) RunContainer(manifest *api.ContainerManifest, container *api.
 			Env:          envVariables,
 			Volumes:      volumes,
 			WorkingDir:   container.WorkingDir,
-			Cmd:          makeCommandLine(container),
+			Cmd:          container.Command,
 		},
 	}
 	dockerContainer, err := kl.DockerClient.CreateContainer(opts)
@@ -329,6 +321,7 @@ func (kl *Kubelet) RunContainer(manifest *api.ContainerManifest, container *api.
 	return name, kl.DockerClient.StartContainer(dockerContainer.ID, &docker.HostConfig{
 		PortBindings: portBindings,
 		Binds:        binds,
+		NetworkMode:  netMode,
 	})
 }
 
@@ -544,12 +537,59 @@ func (kl *Kubelet) WatchEtcd(watchChannel <-chan *etcd.Response, changeChannel c
 	}
 }
 
+const networkContainerName = "k8snet"
+
+func (kl *Kubelet) networkContainerExists(manifest *api.ContainerManifest) (string, bool, error) {
+	pods, err := kl.ListContainers()
+	if err != nil {
+		return "", false, err
+	}
+	for _, name := range pods {
+		if strings.Contains(name, networkContainerName+"--"+manifest.Id+"--") {
+			return name, true, nil
+		}
+	}
+	return "", false, nil
+}
+
+func (kl *Kubelet) createNetworkContainer(manifest *api.ContainerManifest) (string, error) {
+	var ports []api.Port
+	// Docker only exports ports from the network container.  Let's
+	// collect all of the relevant ports an export them.
+	for _, container := range manifest.Containers {
+		ports = append(ports, container.Ports...)
+	}
+	container := &api.Container{
+		Name:    networkContainerName,
+		Image:   "busybox",
+		Command: []string{"sh", "-c", "rm -f nap && mkfifo nap && exec cat nap"},
+		Ports:   ports,
+	}
+	kl.pullImage("busybox")
+	return kl.RunContainer(manifest, container, "")
+}
+
 // Sync the configured list of containers (desired state) with the host current state
 func (kl *Kubelet) SyncManifests(config []api.ContainerManifest) error {
 	log.Printf("Desired:%#v", config)
 	var err error
 	desired := map[string]bool{}
 	for _, manifest := range config {
+		netName, exists, err := kl.networkContainerExists(&manifest)
+		if err != nil {
+			log.Printf("Failed to introspect network container. (%#v)  Skipping container %s", err, manifest.Id)
+			continue
+		}
+		if !exists {
+			log.Printf("Network container doesn't exit, creating")
+			netName, err = kl.createNetworkContainer(&manifest)
+			if err != nil {
+				log.Printf("Failed to create network container: %#v", err)
+			}
+			// Docker list prefixes '/' for some reason, so let's do that...
+			netName = "/" + netName
+		}
+		desired[netName] = true
 		for _, element := range manifest.Containers {
 			var exists bool
 			exists, actualName, err := kl.ContainerExists(&manifest, &element)
@@ -564,7 +604,9 @@ func (kl *Kubelet) SyncManifests(config []api.ContainerManifest) error {
 					log.Printf("Error pulling container: %#v", err)
 					continue
 				}
-				actualName, err = kl.RunContainer(&manifest, &element)
+				// netName has the '/' prefix, so slice it off
+				networkContainer := netName[1:]
+				actualName, err = kl.RunContainer(&manifest, &element, "container:"+networkContainer)
 				// For some reason, list gives back names that start with '/'
 				actualName = "/" + actualName
 

pkg/kubelet/kubelet_test.go

@@ -22,7 +22,6 @@ import (
 	"io/ioutil"
 	"net/http/httptest"
 	"reflect"
-	"strings"
 	"sync"
 	"testing"
 
@@ -90,7 +89,7 @@ type FakeDockerClient struct {
 	container     *docker.Container
 	err           error
 	called        []string
-	stopped       string
+	stopped       []string
 }
 
 func (f *FakeDockerClient) clearCalls() {
@@ -123,7 +122,7 @@ func (f *FakeDockerClient) StartContainer(id string, hostConfig *docker.HostConf
 
 func (f *FakeDockerClient) StopContainer(id string, timeout uint) error {
 	f.appendCall("stop")
-	f.stopped = id
+	f.stopped = append(f.stopped, id)
 	return nil
 }
 
@@ -498,6 +497,11 @@ func TestSyncManifestsDoesNothing(t *testing.T) {
 			Names: []string{"bar--foo"},
 			ID:    "1234",
 		},
+		{
+			// network container
+			Names: []string{"k8snet--foo--"},
+			ID:    "9876",
+		},
 	}
 	fakeDocker.container = &docker.Container{
 		ID: "1234",
@@ -514,11 +518,12 @@ func TestSyncManifestsDoesNothing(t *testing.T) {
 		},
 	})
 	expectNoError(t, err)
-	if len(fakeDocker.called) != 4 ||
+	if len(fakeDocker.called) != 5 ||
 		fakeDocker.called[0] != "list" ||
 		fakeDocker.called[1] != "list" ||
-		fakeDocker.called[2] != "inspect" ||
-		fakeDocker.called[3] != "list" {
+		fakeDocker.called[2] != "list" ||
+		fakeDocker.called[3] != "inspect" ||
+		fakeDocker.called[4] != "list" {
 		t.Errorf("Unexpected call sequence: %#v", fakeDocker.called)
 	}
 }
@@ -533,6 +538,11 @@ func TestSyncManifestsDeletes(t *testing.T) {
 			Names: []string{"foo--bar"},
 			ID:    "1234",
 		},
+		{
+			// network container
+			Names: []string{"k8snet--foo--"},
+			ID:    "9876",
+		},
 		{
 			Names: []string{"foo"},
 			ID:    "4567",
@@ -543,12 +553,15 @@ func TestSyncManifestsDeletes(t *testing.T) {
 	}
 	err := kubelet.SyncManifests([]api.ContainerManifest{})
 	expectNoError(t, err)
-	if len(fakeDocker.called) != 3 ||
+	if len(fakeDocker.called) != 5 ||
 		fakeDocker.called[0] != "list" ||
 		fakeDocker.called[1] != "list" ||
 		fakeDocker.called[2] != "stop" ||
-		fakeDocker.stopped != "1234" {
-		t.Errorf("Unexpected call sequence: %#v", fakeDocker.called)
+		fakeDocker.called[3] != "list" ||
+		fakeDocker.called[4] != "stop" ||
+		fakeDocker.stopped[0] != "1234" ||
+		fakeDocker.stopped[1] != "9876" {
+		t.Errorf("Unexpected call sequence: %#v %s", fakeDocker.called, fakeDocker.stopped)
 	}
 }
 
@@ -596,17 +609,6 @@ func TestEventWritingError(t *testing.T) {
 	}
 }
 
-func TestMakeCommandLine(t *testing.T) {
-	expected := []string{"echo", "hello", "world"}
-	container := api.Container{
-		Command: strings.Join(expected, " "),
-	}
-	cmdLine := makeCommandLine(&container)
-	if !reflect.DeepEqual(expected, cmdLine) {
-		t.Error("Unexpected command line.  Expected %#v, got %#v", expected, cmdLine)
-	}
-}
-
 func TestMakeEnvVariables(t *testing.T) {
 	container := api.Container{
 		Env: []api.EnvVar{