Kubelet: add SyncPod for new runtime API

pkg/kubelet/container/sync_result.go

@@ -40,6 +40,7 @@ var (
 	ErrVerifyNonRoot    = errors.New("VerifyNonRootError")
 	ErrRunInitContainer = errors.New("RunInitContainerError")
 	ErrCreatePodSandbox = errors.New("CreatePodSandboxError")
+	ErrConfigPodSandbox = errors.New("ConfigPodSandboxError")
 	ErrKillPodSandbox   = errors.New("KillPodSandboxError")
 )
 
@@ -59,6 +60,7 @@ const (
 	TeardownNetwork  SyncAction = "TeardownNetwork"
 	InitContainer    SyncAction = "InitContainer"
 	CreatePodSandbox SyncAction = "CreatePodSandbox"
+	ConfigPodSandbox SyncAction = "ConfigPodSandbox"
 	KillPodSandbox   SyncAction = "KillPodSandbox"
 )
 

pkg/kubelet/kuberuntime/kuberuntime_container.go

@@ -33,11 +33,84 @@ import (
 	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/events"
+	"k8s.io/kubernetes/pkg/kubelet/util/format"
 	"k8s.io/kubernetes/pkg/types"
 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
 	"k8s.io/kubernetes/pkg/util/term"
 )
 
+// startContainer starts a container and returns a message indicates why it is failed on error.
+// It starts the container through the following steps:
+// * pull the image
+// * create the container
+// * start the container
+// * run the post start lifecycle hooks (if applicable)
+func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandboxConfig *runtimeApi.PodSandboxConfig, container *api.Container, pod *api.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []api.Secret, podIP string) (string, error) {
+	// Step 1: pull the image.
+	err, msg := m.imagePuller.EnsureImageExists(pod, container, pullSecrets)
+	if err != nil {
+		return msg, err
+	}
+
+	// Step 2: create the container.
+	ref, err := kubecontainer.GenerateContainerRef(pod, container)
+	if err != nil {
+		glog.Errorf("Can't make a ref to pod %q, container %v: %v", format.Pod(pod), container.Name, err)
+	}
+	glog.V(4).Infof("Generating ref for container %s: %#v", container.Name, ref)
+
+	// For a new container, the RestartCount should be 0
+	restartCount := 0
+	containerStatus := podStatus.FindContainerStatusByName(container.Name)
+	if containerStatus != nil {
+		restartCount = containerStatus.RestartCount + 1
+	}
+
+	containerConfig, err := m.generateContainerConfig(container, pod, restartCount, podIP)
+	if err != nil {
+		m.recorder.Eventf(ref, api.EventTypeWarning, events.FailedToCreateContainer, "Failed to create container with error: %v", err)
+		return "Generate Container Config Failed", err
+	}
+	containerID, err := m.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)
+	if err != nil {
+		m.recorder.Eventf(ref, api.EventTypeWarning, events.FailedToCreateContainer, "Failed to create container with error: %v", err)
+		return "Create Container Failed", err
+	}
+	m.recorder.Eventf(ref, api.EventTypeNormal, events.CreatedContainer, "Created container with id %v", containerID)
+	if ref != nil {
+		m.containerRefManager.SetRef(kubecontainer.ContainerID{
+			Type: m.runtimeName,
+			ID:   containerID,
+		}, ref)
+	}
+
+	// Step 3: start the container.
+	err = m.runtimeService.StartContainer(containerID)
+	if err != nil {
+		m.recorder.Eventf(ref, api.EventTypeWarning, events.FailedToStartContainer,
+			"Failed to start container with id %v with error: %v", containerID, err)
+		return "Start Container Failed", err
+	}
+	m.recorder.Eventf(ref, api.EventTypeNormal, events.StartedContainer, "Started container with id %v", containerID)
+
+	// Step 4: execute the post start hook.
+	if container.Lifecycle != nil && container.Lifecycle.PostStart != nil {
+		kubeContainerID := kubecontainer.ContainerID{
+			Type: m.runtimeName,
+			ID:   containerID,
+		}
+		msg, handlerErr := m.runner.Run(kubeContainerID, pod, container, container.Lifecycle.PostStart)
+		if handlerErr != nil {
+			err := fmt.Errorf("PostStart handler: %v", handlerErr)
+			m.generateContainerEvent(kubeContainerID, api.EventTypeWarning, events.FailedPostStartHook, msg)
+			m.killContainer(pod, kubeContainerID, container, "FailedPostStartHook", nil)
+			return "PostStart Hook Failed", err
+		}
+	}
+
+	return "", nil
+}
+
 // getContainerLogsPath gets log path for container.
 func getContainerLogsPath(containerName string, podUID types.UID) string {
 	return path.Join(podLogsRootDirectory, string(podUID), fmt.Sprintf("%s.log", containerName))
@@ -370,12 +443,11 @@ func (m *kubeGenericRuntimeManager) killContainer(pod *api.Pod, containerID kube
 	if pod != nil && containerSpec != nil && containerSpec.Lifecycle != nil && containerSpec.Lifecycle.PreStop != nil {
 		gracePeriod = gracePeriod - m.executePreStopHook(pod, containerID, containerSpec, gracePeriod)
 	}
-	if gracePeriodOverride == nil {
-		// always give containers a minimal shutdown window to avoid unnecessary SIGKILLs
-		if gracePeriod < minimumGracePeriodInSeconds {
-			gracePeriod = minimumGracePeriodInSeconds
-		}
-	} else {
+	// always give containers a minimal shutdown window to avoid unnecessary SIGKILLs
+	if gracePeriod < minimumGracePeriodInSeconds {
+		gracePeriod = minimumGracePeriodInSeconds
+	}
+	if gracePeriodOverride != nil {
 		gracePeriod = *gracePeriodOverride
 		glog.V(3).Infof("Killing container %q, but using %d second grace period override", containerID, gracePeriod)
 	}

pkg/kubelet/kuberuntime/kuberuntime_manager_test.go

@@ -20,6 +20,7 @@ import (
 	"reflect"
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"k8s.io/kubernetes/pkg/api"
@@ -31,6 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	nettest "k8s.io/kubernetes/pkg/kubelet/network/testing"
 	kubetypes "k8s.io/kubernetes/pkg/types"
+	"k8s.io/kubernetes/pkg/util/flowcontrol"
 )
 
 var (
@@ -70,7 +72,7 @@ func makeAndSetFakePod(m *kubeGenericRuntimeManager, fakeRuntime *apitest.FakeRu
 }
 
 func makeFakePodSandbox(m *kubeGenericRuntimeManager, pod *api.Pod, createdAt int64) (*apitest.FakePodSandbox, error) {
-	config, err := m.generatePodSandboxConfig(pod, "", 0)
+	config, err := m.generatePodSandboxConfig(pod, 0)
 	if err != nil {
 		return nil, err
 	}
@@ -114,7 +116,7 @@ func makeFakeContainer(m *kubeGenericRuntimeManager, pod *api.Pod, container api
 }
 
 func makeFakeContainers(m *kubeGenericRuntimeManager, pod *api.Pod, containers []api.Container, createdAt int64) ([]*apitest.FakeContainer, error) {
-	sandboxConfig, err := m.generatePodSandboxConfig(pod, "", 0)
+	sandboxConfig, err := m.generatePodSandboxConfig(pod, 0)
 	if err != nil {
 		return nil, err
 	}
@@ -441,3 +443,44 @@ func TestKillPod(t *testing.T) {
 		assert.Equal(t, runtimeApi.ContainerState_EXITED, c.GetState())
 	}
 }
+
+func TestSyncPod(t *testing.T) {
+	fakeRuntime, fakeImage, m, err := createTestRuntimeManager()
+	assert.NoError(t, err)
+
+	containers := []api.Container{
+		{
+			Name:            "foo1",
+			Image:           "busybox",
+			ImagePullPolicy: api.PullIfNotPresent,
+		},
+		{
+			Name:            "foo2",
+			Image:           "alpine",
+			ImagePullPolicy: api.PullIfNotPresent,
+		},
+	}
+	pod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			UID:       "12345678",
+			Name:      "foo",
+			Namespace: "new",
+		},
+		Spec: api.PodSpec{
+			Containers: containers,
+		},
+	}
+
+	backOff := flowcontrol.NewBackOff(time.Second, time.Minute)
+	result := m.SyncPod(pod, api.PodStatus{}, &kubecontainer.PodStatus{}, []api.Secret{}, backOff)
+	assert.NoError(t, result.Error())
+	assert.Equal(t, 2, len(fakeRuntime.Containers))
+	assert.Equal(t, 2, len(fakeImage.Images))
+	assert.Equal(t, 1, len(fakeRuntime.Sandboxes))
+	for _, sandbox := range fakeRuntime.Sandboxes {
+		assert.Equal(t, runtimeApi.PodSandBoxState_READY, sandbox.GetState())
+	}
+	for _, c := range fakeRuntime.Containers {
+		assert.Equal(t, runtimeApi.ContainerState_RUNNING, c.GetState())
+	}
+}

pkg/kubelet/kuberuntime/kuberuntime_sandbox.go

@@ -17,6 +17,7 @@ limitations under the License.
 package kuberuntime
 
 import (
+	"fmt"
 	"sort"
 
 	"github.com/golang/glog"
@@ -25,10 +26,30 @@ import (
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	"k8s.io/kubernetes/pkg/kubelet/types"
+	"k8s.io/kubernetes/pkg/kubelet/util/format"
 )
 
+// createPodSandbox creates a pod sandbox and returns (podSandBoxID, message, error).
+func (m *kubeGenericRuntimeManager) createPodSandbox(pod *api.Pod, attempt uint32) (string, string, error) {
+	podSandboxConfig, err := m.generatePodSandboxConfig(pod, attempt)
+	if err != nil {
+		message := fmt.Sprintf("GeneratePodSandboxConfig for pod %q failed: %v", format.Pod(pod), err)
+		glog.Error(message)
+		return "", message, err
+	}
+
+	podSandBoxID, err := m.runtimeService.RunPodSandbox(podSandboxConfig)
+	if err != nil {
+		message := fmt.Sprintf("CreatePodSandbox for pod %q failed: %v", format.Pod(pod), err)
+		glog.Error(message)
+		return "", message, err
+	}
+
+	return podSandBoxID, "", nil
+}
+
 // generatePodSandboxConfig generates pod sandbox config from api.Pod.
-func (m *kubeGenericRuntimeManager) generatePodSandboxConfig(pod *api.Pod, podIP string, attempt uint32) (*runtimeApi.PodSandboxConfig, error) {
+func (m *kubeGenericRuntimeManager) generatePodSandboxConfig(pod *api.Pod, attempt uint32) (*runtimeApi.PodSandboxConfig, error) {
 	// TODO: deprecating podsandbox resource requirements in favor of the pod level cgroup
 	// Refer https://github.com/kubernetes/kubernetes/issues/29871
 	podUID := string(pod.UID)
@@ -63,7 +84,8 @@ func (m *kubeGenericRuntimeManager) generatePodSandboxConfig(pod *api.Pod, podIP
 	cgroupParent := ""
 	portMappings := []*runtimeApi.PortMapping{}
 	for _, c := range pod.Spec.Containers {
-		opts, err := m.runtimeHelper.GenerateRunContainerOptions(pod, &c, podIP)
+		// TODO: use a separate interface to only generate portmappings
+		opts, err := m.runtimeHelper.GenerateRunContainerOptions(pod, &c, "")
 		if err != nil {
 			return nil, err
 		}