Merge pull request #30267 from feiskyer/kuberuntime-podstatus

Automatic merge from submit-queue

Kubelet: implement GetPodStatus for new runtime API

Implement `GetPodStatus()` for new runtime API.  Part of #28789 .

CC @yujuhong @Random-Liu @dchen1107

pkg/kubelet/api/testing/fake_runtime_service.go

@@ -28,8 +28,8 @@ import (
 var (
 	version = "0.1.0"
 
-	fakeRuntimeName  = "fakeRuntime"
-	fakePodSandboxIP = "192.168.192.168"
+	FakeRuntimeName  = "fakeRuntime"
+	FakePodSandboxIP = "192.168.192.168"
 )
 
 type FakePodSandbox struct {
@@ -93,7 +93,7 @@ func (r *FakeRuntimeService) Version(apiVersion string) (*runtimeApi.VersionResp
 
 	return &runtimeApi.VersionResponse{
 		Version:           &version,
-		RuntimeName:       &fakeRuntimeName,
+		RuntimeName:       &FakeRuntimeName,
 		RuntimeVersion:    &version,
 		RuntimeApiVersion: &version,
 	}, nil
@@ -169,7 +169,7 @@ func (r *FakeRuntimeService) PodSandboxStatus(podSandboxID string) (*runtimeApi.
 		CreatedAt: s.CreatedAt,
 		State:     s.State,
 		Network: &runtimeApi.PodSandboxNetworkStatus{
-			Ip: &fakePodSandboxIP,
+			Ip: &FakePodSandboxIP,
 		},
 		Labels:      s.Labels,
 		Annotations: s.Annotations,

pkg/kubelet/container/runtime.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
+	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util/flowcontrol"
 	"k8s.io/kubernetes/pkg/util/term"
@@ -225,12 +226,11 @@ func (id DockerID) ContainerID() ContainerID {
 type ContainerState string
 
 const (
+	ContainerStateCreated ContainerState = "created"
 	ContainerStateRunning ContainerState = "running"
 	ContainerStateExited  ContainerState = "exited"
 	// This unknown encompasses all the states that we currently don't care.
 	ContainerStateUnknown ContainerState = "unknown"
-	// Not in use yet.
-	ContainerStateCreated ContainerState = "created"
 )
 
 // Container provides the runtime information for a container, such as ID, hash,
@@ -267,6 +267,9 @@ type PodStatus struct {
 	IP string
 	// Status of containers in the pod.
 	ContainerStatuses []*ContainerStatus
+	// Status of the pod sandbox.
+	// Only for kuberuntime now, other runtime may keep it nil.
+	SandboxStatuses []*runtimeApi.PodSandboxStatus
 }
 
 // ContainerStatus represents the status of a container.

pkg/kubelet/kuberuntime/doc.go

@@ -0,0 +1,19 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+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 kuberuntime contains an implementation of kubecontainer.Runtime using
+// the interface in pkg/kubelet/api/services.go.
+package kuberuntime

pkg/kubelet/kuberuntime/helpers.go

@@ -48,6 +48,19 @@ func (b containersByID) Len() int           { return len(b) }
 func (b containersByID) Swap(i, j int)      { b[i], b[j] = b[j], b[i] }
 func (b containersByID) Less(i, j int) bool { return b[i].ID.ID < b[j].ID.ID }
 
+// Newest first.
+type podSandboxByCreated []*runtimeApi.PodSandbox
+
+func (p podSandboxByCreated) Len() int           { return len(p) }
+func (p podSandboxByCreated) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
+func (p podSandboxByCreated) Less(i, j int) bool { return p[i].GetCreatedAt() > p[j].GetCreatedAt() }
+
+type containerStatusByCreated []*kubecontainer.ContainerStatus
+
+func (c containerStatusByCreated) Len() int           { return len(c) }
+func (c containerStatusByCreated) Swap(i, j int)      { c[i], c[j] = c[j], c[i] }
+func (c containerStatusByCreated) Less(i, j int) bool { return c[i].CreatedAt.After(c[j].CreatedAt) }
+
 // toKubeContainerState converts runtimeApi.ContainerState to kubecontainer.ContainerState.
 func toKubeContainerState(state runtimeApi.ContainerState) kubecontainer.ContainerState {
 	switch state {

pkg/kubelet/kuberuntime/kuberuntime_container.go

@@ -19,9 +19,12 @@ package kuberuntime
 import (
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math/rand"
 	"os"
 	"path"
+	"sort"
+	"time"
 
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
@@ -232,6 +235,73 @@ func makeUID() string {
 	return fmt.Sprintf("%08x", rand.Uint32())
 }
 
+// getKubeletContainerStatuses gets all containers' status for the pod sandbox.
+func (m *kubeGenericRuntimeManager) getKubeletContainerStatuses(podSandboxID string) ([]*kubecontainer.ContainerStatus, error) {
+	containers, err := m.runtimeService.ListContainers(&runtimeApi.ContainerFilter{
+		PodSandboxId: &podSandboxID,
+	})
+	if err != nil {
+		glog.Errorf("ListContainers error: %v", err)
+		return nil, err
+	}
+
+	statuses := make([]*kubecontainer.ContainerStatus, len(containers))
+	// TODO: optimization: set maximum number of containers per container name to examine.
+	for i, c := range containers {
+		status, err := m.runtimeService.ContainerStatus(c.GetId())
+		if err != nil {
+			glog.Errorf("ContainerStatus for %s error: %v", c.GetId(), err)
+			return nil, err
+		}
+
+		annotatedInfo := getContainerInfoFromAnnotations(c.Annotations)
+		labeledInfo := getContainerInfoFromLabels(c.Labels)
+		cStatus := &kubecontainer.ContainerStatus{
+			ID: kubecontainer.ContainerID{
+				Type: m.runtimeName,
+				ID:   c.GetId(),
+			},
+			Name:         labeledInfo.ContainerName,
+			Image:        status.Image.GetImage(),
+			ImageID:      status.GetImageRef(),
+			Hash:         annotatedInfo.Hash,
+			RestartCount: annotatedInfo.RestartCount,
+			State:        toKubeContainerState(c.GetState()),
+			CreatedAt:    time.Unix(status.GetCreatedAt(), 0),
+		}
+
+		if c.GetState() == runtimeApi.ContainerState_RUNNING {
+			cStatus.StartedAt = time.Unix(status.GetStartedAt(), 0)
+		} else {
+			cStatus.Reason = status.GetReason()
+			cStatus.ExitCode = int(status.GetExitCode())
+			cStatus.FinishedAt = time.Unix(status.GetFinishedAt(), 0)
+		}
+
+		message := ""
+		if !cStatus.FinishedAt.IsZero() || cStatus.ExitCode != 0 {
+			if annotatedInfo.TerminationMessagePath != "" {
+				for _, mount := range status.Mounts {
+					if mount.GetContainerPath() == annotatedInfo.TerminationMessagePath {
+						path := mount.GetHostPath()
+						if data, err := ioutil.ReadFile(path); err != nil {
+							message = fmt.Sprintf("Error on reading termination-log %s: %v", path, err)
+						} else {
+							message = string(data)
+						}
+						break
+					}
+				}
+			}
+		}
+		cStatus.Message = message
+		statuses[i] = cStatus
+	}
+
+	sort.Sort(containerStatusByCreated(statuses))
+	return statuses, nil
+}
+
 // AttachContainer attaches to the container's console
 func (m *kubeGenericRuntimeManager) AttachContainer(id kubecontainer.ContainerID, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan term.Size) (err error) {
 	return fmt.Errorf("not implemented")

pkg/kubelet/kuberuntime/kuberuntime_manager.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/coreos/go-semver/semver"
 	"github.com/golang/glog"
+
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/client/record"
 	"k8s.io/kubernetes/pkg/credentialprovider"
@@ -292,7 +293,58 @@ func (m *kubeGenericRuntimeManager) KillPod(pod *api.Pod, runningPod kubecontain
 // GetPodStatus retrieves the status of the pod, including the
 // information of all containers in the pod that are visble in Runtime.
 func (m *kubeGenericRuntimeManager) GetPodStatus(uid kubetypes.UID, name, namespace string) (*kubecontainer.PodStatus, error) {
-	return nil, fmt.Errorf("not implemented")
+	// Now we retain restart count of container as a container label. Each time a container
+	// restarts, pod will read the restart count from the registered dead container, increment
+	// it to get the new restart count, and then add a label with the new restart count on
+	// the newly started container.
+	// However, there are some limitations of this method:
+	//	1. When all dead containers were garbage collected, the container status could
+	//	not get the historical value and would be *inaccurate*. Fortunately, the chance
+	//	is really slim.
+	//	2. When working with old version containers which have no restart count label,
+	//	we can only assume their restart count is 0.
+	// Anyhow, we only promised "best-effort" restart count reporting, we can just ignore
+	// these limitations now.
+	// TODO: move this comment to SyncPod.
+	podFullName := kubecontainer.BuildPodFullName(name, namespace)
+	podSandboxIDs, err := m.getSandboxIDByPodUID(string(uid), nil)
+	if err != nil {
+		return nil, err
+	}
+	glog.V(4).Infof("getSandboxIDByPodUID got sandbox IDs %q for pod %q(UID:%q)", podSandboxIDs, podFullName, string(uid))
+
+	sandboxStatuses := make([]*runtimeApi.PodSandboxStatus, len(podSandboxIDs))
+	containerStatuses := []*kubecontainer.ContainerStatus{}
+	podIP := ""
+	for idx, podSandboxID := range podSandboxIDs {
+		podSandboxStatus, err := m.runtimeService.PodSandboxStatus(podSandboxID)
+		if err != nil {
+			glog.Errorf("PodSandboxStatus for pod (uid:%v, name:%s, namespace:%s) error: %v", uid, name, namespace, err)
+			return nil, err
+		}
+		sandboxStatuses[idx] = podSandboxStatus
+
+		// Only get pod IP from latest sandbox
+		if idx == 0 && podSandboxStatus.GetState() == runtimeApi.PodSandBoxState_READY {
+			podIP = m.determinePodSandboxIP(namespace, name, podSandboxStatus)
+		}
+
+		containerStatus, err := m.getKubeletContainerStatuses(podSandboxID)
+		if err != nil {
+			glog.Errorf("getKubeletContainerStatuses for sandbox %s failed: %v", podSandboxID, err)
+			return nil, err
+		}
+		containerStatuses = append(containerStatuses, containerStatus...)
+	}
+
+	return &kubecontainer.PodStatus{
+		ID:                uid,
+		Name:              name,
+		Namespace:         namespace,
+		IP:                podIP,
+		SandboxStatuses:   sandboxStatuses,
+		ContainerStatuses: containerStatuses,
+	}, nil
 }
 
 // Returns the filesystem path of the pod's network namespace; if the

pkg/kubelet/kuberuntime/kuberuntime_manager_test.go

@@ -172,7 +172,46 @@ func TestContainerRuntimeType(t *testing.T) {
 	assert.NoError(t, err)
 
 	runtimeType := m.Type()
-	assert.Equal(t, "fakeRuntime", runtimeType)
+	assert.Equal(t, apitest.FakeRuntimeName, runtimeType)
+}
+
+func TestGetPodStatus(t *testing.T) {
+	fakeRuntime, _, m, err := createTestRuntimeManager()
+	assert.NoError(t, err)
+
+	containers := []api.Container{
+		{
+			Name:            "foo1",
+			Image:           "busybox",
+			ImagePullPolicy: api.PullIfNotPresent,
+		},
+		{
+			Name:            "foo2",
+			Image:           "busybox",
+			ImagePullPolicy: api.PullIfNotPresent,
+		},
+	}
+	pod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			UID:       "12345678",
+			Name:      "foo",
+			Namespace: "new",
+		},
+		Spec: api.PodSpec{
+			Containers: containers,
+		},
+	}
+
+	// Set fake sandbox and faked containers to fakeRuntime.
+	_, _, err = makeAndSetFakePod(m, fakeRuntime, pod)
+	assert.NoError(t, err)
+
+	podStatus, err := m.GetPodStatus(pod.UID, pod.Name, pod.Namespace)
+	assert.NoError(t, err)
+	assert.Equal(t, pod.UID, podStatus.ID)
+	assert.Equal(t, pod.Name, podStatus.Name)
+	assert.Equal(t, pod.Namespace, podStatus.Namespace)
+	assert.Equal(t, apitest.FakePodSandboxIP, podStatus.IP)
 }
 
 func TestGetPods(t *testing.T) {

pkg/kubelet/kuberuntime/kuberuntime_sandbox.go

@@ -17,10 +17,14 @@ limitations under the License.
 package kuberuntime
 
 import (
+	"sort"
+
 	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/api"
 	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	"k8s.io/kubernetes/pkg/kubelet/network"
+	"k8s.io/kubernetes/pkg/kubelet/types"
 )
 
 // generatePodSandboxConfig generates pod sandbox config from api.Pod.
@@ -141,3 +145,55 @@ func (m *kubeGenericRuntimeManager) getKubeletSandboxes(all bool) ([]*runtimeApi
 
 	return result, nil
 }
+
+// determinePodSandboxIP determines the IP address of the given pod sandbox.
+// TODO: remove determinePodSandboxIP after networking is delegated to the container runtime.
+func (m *kubeGenericRuntimeManager) determinePodSandboxIP(podNamespace, podName string, podSandbox *runtimeApi.PodSandboxStatus) string {
+	ip := ""
+
+	if podSandbox.Network != nil {
+		ip = podSandbox.Network.GetIp()
+	}
+
+	if m.networkPlugin.Name() != network.DefaultPluginName {
+		// TODO: podInfraContainerID in GetPodNetworkStatus() interface should be renamed to sandboxID
+		netStatus, err := m.networkPlugin.GetPodNetworkStatus(podNamespace, podName, kubecontainer.ContainerID{
+			Type: m.runtimeName,
+			ID:   podSandbox.GetId(),
+		})
+		if err != nil {
+			glog.Errorf("NetworkPlugin %s failed on the status hook for pod '%s' - %v", m.networkPlugin.Name(), kubecontainer.BuildPodFullName(podName, podNamespace), err)
+		} else if netStatus != nil {
+			ip = netStatus.IP.String()
+		}
+	}
+
+	return ip
+}
+
+// getPodSandboxID gets the sandbox id by podUID and returns ([]sandboxID, error).
+// Param state could be nil in order to get all sandboxes belonging to same pod.
+func (m *kubeGenericRuntimeManager) getSandboxIDByPodUID(podUID string, state *runtimeApi.PodSandBoxState) ([]string, error) {
+	filter := &runtimeApi.PodSandboxFilter{
+		State:         state,
+		LabelSelector: map[string]string{types.KubernetesPodUIDLabel: podUID},
+	}
+	sandboxes, err := m.runtimeService.ListPodSandbox(filter)
+	if err != nil {
+		glog.Errorf("ListPodSandbox with pod UID %q failed: %v", podUID, err)
+		return nil, err
+	}
+
+	if len(sandboxes) == 0 {
+		return nil, nil
+	}
+
+	// Sort with newest first.
+	sandboxIDs := make([]string, len(sandboxes))
+	sort.Sort(podSandboxByCreated(sandboxes))
+	for i, s := range sandboxes {
+		sandboxIDs[i] = s.GetId()
+	}
+
+	return sandboxIDs, nil
+}

pkg/kubelet/pleg/generic.go

@@ -79,6 +79,9 @@ const (
 
 func convertState(state kubecontainer.ContainerState) plegContainerState {
 	switch state {
+	case kubecontainer.ContainerStateCreated:
+		// kubelet doesn't use the "created" state yet, hence convert it to "unknown".
+		return plegContainerUnknown
 	case kubecontainer.ContainerStateRunning:
 		return plegContainerRunning
 	case kubecontainer.ContainerStateExited: