Merge pull request #7421 from brendandburns/prestop

Implement PreStop hooks.

pkg/kubelet/dockertools/fake_docker_client.go

@@ -48,6 +48,7 @@ type FakeDockerClient struct {
 	VersionInfo         docker.Env
 	Information         docker.Env
 	ExecInspect         *docker.ExecInspect
+	execCmd             []string
 }
 
 func (f *FakeDockerClient) ClearCalls() {
@@ -281,11 +282,18 @@ func (f *FakeDockerClient) Info() (*docker.Env, error) {
 	return &f.Information, nil
 }
 
-func (f *FakeDockerClient) CreateExec(_ docker.CreateExecOptions) (*docker.Exec, error) {
+func (f *FakeDockerClient) CreateExec(opts docker.CreateExecOptions) (*docker.Exec, error) {
+	f.Lock()
+	defer f.Unlock()
+	f.execCmd = opts.Cmd
+	f.called = append(f.called, "create_exec")
 	return &docker.Exec{"12345678"}, nil
 }
 
 func (f *FakeDockerClient) StartExec(_ string, _ docker.StartExecOptions) error {
+	f.Lock()
+	defer f.Unlock()
+	f.called = append(f.called, "start_exec")
 	return nil
 }
 

pkg/kubelet/dockertools/manager.go

@@ -31,6 +31,7 @@ import (
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/record"
 	kubecontainer "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/container"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/lifecycle"
@@ -52,6 +53,9 @@ const (
 	podOomScoreAdj = -100
 
 	maxReasonCacheEntries = 200
+
+	kubernetesPodLabel       = "io.kubernetes.pod.data"
+	kubernetesContainerLabel = "io.kubernetes.container.name"
 )
 
 // DockerManager implements the Runtime interface.
@@ -499,6 +503,17 @@ func (dm *DockerManager) runContainer(
 	labels := map[string]string{
 		"io.kubernetes.pod.name": namespacedName.String(),
 	}
+	if container.Lifecycle != nil && container.Lifecycle.PreStop != nil {
+		glog.V(1).Infof("Setting preStop hook")
+		// TODO: This is kind of hacky, we should really just encode the bits we need.
+		data, err := latest.Codec.Encode(pod)
+		if err != nil {
+			glog.Errorf("Failed to encode pod: %s for prestop hook", pod.Name)
+		} else {
+			labels[kubernetesPodLabel] = string(data)
+			labels[kubernetesContainerLabel] = container.Name
+		}
+	}
 	dockerOpts := docker.CreateContainerOptions{
 		Name: BuildDockerName(dockerName, container),
 		Config: &docker.Config{
@@ -1099,9 +1114,41 @@ func (dm *DockerManager) KillContainer(containerID types.UID) error {
 func (dm *DockerManager) killContainer(containerID types.UID) error {
 	ID := string(containerID)
 	glog.V(2).Infof("Killing container with id %q", ID)
+	inspect, err := dm.client.InspectContainer(ID)
+	if err != nil {
+		return err
+	}
+	var found bool
+	var preStop string
+	if inspect != nil && inspect.Config != nil && inspect.Config.Labels != nil {
+		preStop, found = inspect.Config.Labels[kubernetesPodLabel]
+	}
+	if found {
+		var pod api.Pod
+		err := latest.Codec.DecodeInto([]byte(preStop), &pod)
+		if err != nil {
+			glog.Errorf("Failed to decode prestop: %s, %s", preStop, ID)
+		} else {
+			name := inspect.Config.Labels[kubernetesContainerLabel]
+			var container *api.Container
+			for ix := range pod.Spec.Containers {
+				if pod.Spec.Containers[ix].Name == name {
+					container = &pod.Spec.Containers[ix]
+					break
+				}
+			}
+			if container != nil {
+				glog.V(1).Infof("Running preStop hook")
+				if err := dm.runner.Run(ID, &pod, container, container.Lifecycle.PreStop); err != nil {
+					glog.Errorf("failed to run preStop hook: %v", err)
+				}
+			} else {
+				glog.Errorf("unable to find container %v, %s", pod, name)
+			}
+		}
+	}
 	dm.readinessManager.RemoveReadiness(ID)
-	err := dm.client.StopContainer(ID, 10)
-
+	err = dm.client.StopContainer(ID, 10)
 	ref, ok := dm.containerRefManager.GetRef(ID)
 	if !ok {
 		glog.Warningf("No ref for pod '%v'", ID)

pkg/kubelet/dockertools/manager_test.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/testapi"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/record"
 	kubecontainer "github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/container"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/network"
@@ -37,7 +38,7 @@ import (
 )
 
 func newTestDockerManager() (*DockerManager, *FakeDockerClient) {
-	fakeDocker := &FakeDockerClient{Errors: make(map[string]error), RemovedImages: util.StringSet{}}
+	fakeDocker := &FakeDockerClient{VersionInfo: docker.Env{"Version=1.1.3", "ApiVersion=1.15"}, Errors: make(map[string]error), RemovedImages: util.StringSet{}}
 	fakeRecorder := &record.FakeRecorder{}
 	readinessManager := kubecontainer.NewReadinessManager()
 	containerRefManager := kubecontainer.NewRefManager()
@@ -298,6 +299,75 @@ func TestKillContainerInPod(t *testing.T) {
 	}
 }
 
+func TestKillContainerInPodWithPreStop(t *testing.T) {
+	manager, fakeDocker := newTestDockerManager()
+	fakeDocker.ExecInspect = &docker.ExecInspect{
+		Running:  false,
+		ExitCode: 0,
+	}
+	expectedCmd := []string{"foo.sh", "bar"}
+	pod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			UID:       "12345678",
+			Name:      "qux",
+			Namespace: "new",
+		},
+		Spec: api.PodSpec{
+			Containers: []api.Container{
+				{
+					Name: "foo",
+					Lifecycle: &api.Lifecycle{
+						PreStop: &api.Handler{
+							Exec: &api.ExecAction{
+								Command: expectedCmd,
+							},
+						},
+					},
+				},
+				{Name: "bar"}}},
+	}
+	podString, err := testapi.Codec().Encode(pod)
+	if err != nil {
+		t.Errorf("unexpected error: %v")
+	}
+	containers := []docker.APIContainers{
+		{
+			ID:    "1111",
+			Names: []string{"/k8s_foo_qux_new_1234_42"},
+		},
+		{
+			ID:    "2222",
+			Names: []string{"/k8s_bar_qux_new_1234_42"},
+		},
+	}
+	containerToKill := &containers[0]
+	fakeDocker.ContainerList = containers
+	fakeDocker.Container = &docker.Container{
+		Config: &docker.Config{
+			Labels: map[string]string{
+				kubernetesPodLabel:       string(podString),
+				kubernetesContainerLabel: "foo",
+			},
+		},
+	}
+	// Set all containers to ready.
+	for _, c := range fakeDocker.ContainerList {
+		manager.readinessManager.SetReadiness(c.ID, true)
+	}
+
+	if err := manager.KillContainerInPod(pod.Spec.Containers[0], pod); err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	// Assert the container has been stopped.
+	if err := fakeDocker.AssertStopped([]string{containerToKill.ID}); err != nil {
+		t.Errorf("container was not stopped correctly: %v", err)
+	}
+	verifyCalls(t, fakeDocker, []string{"list", "inspect_container", "create_exec", "start_exec", "stop"})
+	if !reflect.DeepEqual(expectedCmd, fakeDocker.execCmd) {
+		t.Errorf("expected: %v, got %v", expectedCmd, fakeDocker.execCmd)
+	}
+}
+
 func TestKillContainerInPodWithError(t *testing.T) {
 	manager, fakeDocker := newTestDockerManager()
 

pkg/kubelet/kubelet_test.go

@@ -867,7 +867,7 @@ func TestSyncPodsDeletesWithNoPodInfraContainer(t *testing.T) {
 		// Get pod status.
 		"list", "inspect_container",
 		// Kill the container since pod infra container is not running.
-		"stop",
+		"inspect_container", "stop",
 		// Create pod infra container.
 		"create", "start", "inspect_container",
 		// Create container.
@@ -933,7 +933,7 @@ func TestSyncPodsDeletesWhenSourcesAreReady(t *testing.T) {
 	if err := kubelet.SyncPods([]*api.Pod{}, emptyPodUIDs, map[string]*api.Pod{}, time.Now()); err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
-	verifyCalls(t, fakeDocker, []string{"list", "stop", "stop", "list"})
+	verifyCalls(t, fakeDocker, []string{"list", "inspect_container", "stop", "inspect_container", "stop", "list"})
 
 	// A map iteration is used to delete containers, so must not depend on
 	// order here.
@@ -976,7 +976,7 @@ func TestSyncPodsDeletes(t *testing.T) {
 		t.Errorf("unexpected error: %v", err)
 	}
 
-	verifyCalls(t, fakeDocker, []string{"list", "stop", "stop", "list"})
+	verifyCalls(t, fakeDocker, []string{"list", "inspect_container", "stop", "inspect_container", "stop", "list"})
 
 	// A map iteration is used to delete containers, so must not depend on
 	// order here.
@@ -1062,7 +1062,7 @@ func TestSyncPodsDeletesDuplicate(t *testing.T) {
 		// Check the pod infra container.
 		"inspect_container",
 		// Kill the duplicated container.
-		"stop",
+		"inspect_container", "stop",
 		// Get pod status.
 		"list", "inspect_container", "inspect_container", "inspect_container",
 		// Get pods for deleting orphaned volumes.
@@ -1135,7 +1135,7 @@ func TestSyncPodsBadHash(t *testing.T) {
 		// Check the pod infra container.
 		"inspect_container",
 		// Kill and restart the bad hash container.
-		"stop", "create", "start",
+		"inspect_container", "stop", "create", "start",
 		// Get pod status.
 		"list", "inspect_container", "inspect_container", "inspect_container",
 		// Get pods for deleting orphaned volumes.
@@ -1211,7 +1211,7 @@ func TestSyncPodsUnhealthy(t *testing.T) {
 		// Check the pod infra container.
 		"inspect_container",
 		// Kill the unhealthy container.
-		"stop",
+		"inspect_container", "stop",
 		// Restart the unhealthy container.
 		"create", "start",
 		// Get pod status.
@@ -1736,7 +1736,7 @@ func TestSyncPodEventHandlerFails(t *testing.T) {
 		// Create the container.
 		"create", "start",
 		// Kill the container since event handler fails.
-		"stop",
+		"inspect_container", "stop",
 		// Get pod status.
 		"list", "inspect_container", "inspect_container",
 		// Get pods for deleting orphaned volumes.
@@ -3981,7 +3981,7 @@ func TestSyncPodsWithRestartPolicy(t *testing.T) {
 				// Check the pod infra container.
 				"inspect_container",
 				// Stop the last pod infra container.
-				"stop",
+				"inspect_container", "stop",
 				// Get pod status.
 				"list", "inspect_container", "inspect_container", "inspect_container",
 				// Get pods for deleting orphaned volumes.