Kubelet: implement GetPods for kuberuntime

pkg/kubelet/api/v1alpha1/runtime/api.proto

@@ -495,6 +495,9 @@ message Container {
     optional ContainerState state = 5;
     // Labels are key value pairs that may be used to scope and select individual resources.
     map<string, string> labels = 6;
+    // Annotations is an unstructured key value map that may be set by external
+    // tools to store and retrieve arbitrary metadata.
+    map<string, string> annotations = 7;
 }
 
 message ListContainersResponse {

pkg/kubelet/kuberuntime/helpers.go

@@ -40,10 +40,22 @@ const (
 	milliCPUToCPU = 1000
 
 	// 100000 is equivalent to 100ms
-	quotaPeriod    = 100000
+	quotaPeriod    = 100 * minQuotaPeriod
 	minQuotaPeriod = 1000
 )
 
+type podsByID []*kubecontainer.Pod
+
+func (b podsByID) Len() int           { return len(b) }
+func (b podsByID) Swap(i, j int)      { b[i], b[j] = b[j], b[i] }
+func (b podsByID) Less(i, j int) bool { return b[i].ID < b[j].ID }
+
+type containersByID []*kubecontainer.Container
+
+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 }
+
 // buildSandboxName creates a name which can be reversed to identify sandbox full name
 func buildSandboxName(pod *api.Pod) string {
 	_, sandboxName, _ := buildKubeGenericName(pod, kubeSandboxNamePrefix)
@@ -107,7 +119,23 @@ func parseContainerName(name string) (podName, podNamespace, podUID, containerNa
 	return parts[2], parts[3], parts[4], containerName, hash, nil
 }
 
-// toRuntimeProtocol converts api.Protocol to runtimeApi.Protocol
+// toKubeContainerState converts runtimeApi.ContainerState to kubecontainer.ContainerState.
+func toKubeContainerState(state runtimeApi.ContainerState) kubecontainer.ContainerState {
+	switch state {
+	case runtimeApi.ContainerState_CREATED:
+		return kubecontainer.ContainerStateCreated
+	case runtimeApi.ContainerState_RUNNING:
+		return kubecontainer.ContainerStateRunning
+	case runtimeApi.ContainerState_EXITED:
+		return kubecontainer.ContainerStateExited
+	case runtimeApi.ContainerState_UNKNOWN:
+		return kubecontainer.ContainerStateUnknown
+	}
+
+	return kubecontainer.ContainerStateUnknown
+}
+
+// toRuntimeProtocol converts api.Protocol to runtimeApi.Protocol.
 func toRuntimeProtocol(protocol api.Protocol) runtimeApi.Protocol {
 	switch protocol {
 	case api.ProtocolTCP:
@@ -120,6 +148,23 @@ func toRuntimeProtocol(protocol api.Protocol) runtimeApi.Protocol {
 	return runtimeApi.Protocol_TCP
 }
 
+// toKubeContainer converts runtimeApi.Container to kubecontainer.Container.
+func (m *kubeGenericRuntimeManager) toKubeContainer(c *runtimeApi.Container) (*kubecontainer.Container, error) {
+	if c == nil || c.Id == nil || c.Image == nil || c.State == nil {
+		return nil, fmt.Errorf("unable to convert a nil pointer to a runtime container")
+	}
+
+	labeledInfo := getContainerInfoFromLabels(c.Labels)
+	annotatedInfo := getContainerInfoFromAnnotations(c.Annotations)
+	return &kubecontainer.Container{
+		ID:    kubecontainer.ContainerID{Type: m.runtimeName, ID: c.GetId()},
+		Name:  labeledInfo.ContainerName,
+		Image: c.Image.GetImage(),
+		Hash:  annotatedInfo.Hash,
+		State: toKubeContainerState(c.GetState()),
+	}, nil
+}
+
 // milliCPUToShares converts milliCPU to CPU shares
 func milliCPUToShares(milliCPU int64) int64 {
 	if milliCPU == 0 {

pkg/kubelet/kuberuntime/kuberuntime_container.go

@@ -191,6 +191,37 @@ func makeMounts(cid string, opts *kubecontainer.RunContainerOptions, container *
 	return volumeMounts
 }
 
+// getKubeletContainers lists containers managed by kubelet.
+// The boolean parameter specifies whether returns all containers including
+// those already exited and dead containers (used for garbage collection).
+func (m *kubeGenericRuntimeManager) getKubeletContainers(allContainers bool) ([]*runtimeApi.Container, error) {
+	filter := &runtimeApi.ContainerFilter{
+		LabelSelector: map[string]string{kubernetesManagedLabel: "true"},
+	}
+	if !allContainers {
+		runningState := runtimeApi.ContainerState_RUNNING
+		filter.State = &runningState
+	}
+
+	containers, err := m.getContainersHelper(filter)
+	if err != nil {
+		glog.Errorf("getKubeletContainers failed: %v", err)
+		return nil, err
+	}
+
+	return containers, nil
+}
+
+// getContainers lists containers by filter.
+func (m *kubeGenericRuntimeManager) getContainersHelper(filter *runtimeApi.ContainerFilter) ([]*runtimeApi.Container, error) {
+	resp, err := m.runtimeService.ListContainers(filter)
+	if err != nil {
+		return nil, err
+	}
+
+	return resp, err
+}
+
 // 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_image_test.go

@@ -25,7 +25,7 @@ import (
 )
 
 func TestPullImage(t *testing.T) {
-	_, _, fakeManager, err := createTestFakeRuntimeManager()
+	_, _, fakeManager, err := createTestRuntimeManager()
 	assert.NoError(t, err)
 
 	err = fakeManager.PullImage(kubecontainer.ImageSpec{Image: "busybox"}, nil)
@@ -38,7 +38,7 @@ func TestPullImage(t *testing.T) {
 }
 
 func TestListImages(t *testing.T) {
-	_, fakeImageService, fakeManager, err := createTestFakeRuntimeManager()
+	_, fakeImageService, fakeManager, err := createTestRuntimeManager()
 	assert.NoError(t, err)
 
 	images := []string{"1111", "2222", "3333"}
@@ -56,7 +56,7 @@ func TestListImages(t *testing.T) {
 }
 
 func TestIsImagePresent(t *testing.T) {
-	_, fakeImageService, fakeManager, err := createTestFakeRuntimeManager()
+	_, fakeImageService, fakeManager, err := createTestRuntimeManager()
 	assert.NoError(t, err)
 
 	image := "busybox"
@@ -67,7 +67,7 @@ func TestIsImagePresent(t *testing.T) {
 }
 
 func TestRemoveImage(t *testing.T) {
-	_, fakeImageService, fakeManager, err := createTestFakeRuntimeManager()
+	_, fakeImageService, fakeManager, err := createTestRuntimeManager()
 	assert.NoError(t, err)
 
 	err = fakeManager.PullImage(kubecontainer.ImageSpec{Image: "busybox"}, nil)

pkg/kubelet/kuberuntime/kuberuntime_manager.go

@@ -223,7 +223,53 @@ func (m *kubeGenericRuntimeManager) Status() error {
 // specifies whether the runtime returns all containers including those already
 // exited and dead containers (used for garbage collection).
 func (m *kubeGenericRuntimeManager) GetPods(all bool) ([]*kubecontainer.Pod, error) {
-	return nil, fmt.Errorf("not implemented")
+	pods := make(map[kubetypes.UID]*kubecontainer.Pod)
+	sandboxes, err := m.getKubeletSandboxes(all)
+	if err != nil {
+		return nil, err
+	}
+	for _, s := range sandboxes {
+		name, namespace, uid, _ := parseSandboxName(s.GetName())
+		podUID := kubetypes.UID(uid)
+		pods[podUID] = &kubecontainer.Pod{
+			ID:        podUID,
+			Name:      name,
+			Namespace: namespace,
+		}
+	}
+
+	containers, err := m.getKubeletContainers(all)
+	if err != nil {
+		return nil, err
+	}
+	for _, c := range containers {
+		labelledInfo := getContainerInfoFromLabels(c.Labels)
+		pod, found := pods[labelledInfo.PodUID]
+		if !found {
+			pod = &kubecontainer.Pod{
+				ID:        labelledInfo.PodUID,
+				Name:      labelledInfo.PodName,
+				Namespace: labelledInfo.PodNamespace,
+			}
+			pods[labelledInfo.PodUID] = pod
+		}
+
+		converted, err := m.toKubeContainer(c)
+		if err != nil {
+			glog.Warningf("Convert %s container %v of pod %q failed: %v", m.runtimeName, c, labelledInfo.PodUID, err)
+			continue
+		}
+
+		pod.Containers = append(pod.Containers, converted)
+	}
+
+	// Convert map to list.
+	var result []*kubecontainer.Pod
+	for _, pod := range pods {
+		result = append(result, pod)
+	}
+
+	return result, nil
 }
 
 // SyncPod syncs the running pod into the desired pod.

pkg/kubelet/kuberuntime/kuberuntime_manager_test.go

@@ -17,17 +17,27 @@ limitations under the License.
 package kuberuntime
 
 import (
+	"reflect"
+	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/apis/componentconfig"
 	apitest "k8s.io/kubernetes/pkg/kubelet/api/testing"
+	runtimeApi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	containertest "k8s.io/kubernetes/pkg/kubelet/container/testing"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	nettest "k8s.io/kubernetes/pkg/kubelet/network/testing"
+	kubetypes "k8s.io/kubernetes/pkg/types"
 )
 
-func createTestFakeRuntimeManager() (*apitest.FakeRuntimeService, *apitest.FakeImageService, *kubeGenericRuntimeManager, error) {
+var (
+	fakeCreatedAt int64 = 1
+)
+
+func createTestRuntimeManager() (*apitest.FakeRuntimeService, *apitest.FakeImageService, *kubeGenericRuntimeManager, error) {
 	fakeRuntimeService := apitest.NewFakeRuntimeService()
 	fakeImageService := apitest.NewFakeImageService()
 	networkPlugin, _ := network.InitNetworkPlugin(
@@ -42,13 +52,113 @@ func createTestFakeRuntimeManager() (*apitest.FakeRuntimeService, *apitest.FakeI
 	return fakeRuntimeService, fakeImageService, manager, err
 }
 
+func makeAndSetFakePod(m *kubeGenericRuntimeManager, fakeRuntime *apitest.FakeRuntimeService, pod *api.Pod) (*apitest.FakePodSandbox, []*apitest.FakeContainer, error) {
+	fakePodSandbox, err := makeFakePodSandbox(m, pod)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	fakeContainers, err := makeFakeContainers(m, pod, pod.Spec.Containers)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	fakeRuntime.SetFakeSandboxes([]*apitest.FakePodSandbox{fakePodSandbox})
+	fakeRuntime.SetFakeContainers(fakeContainers)
+	return fakePodSandbox, fakeContainers, nil
+}
+
+func makeFakePodSandbox(m *kubeGenericRuntimeManager, pod *api.Pod) (*apitest.FakePodSandbox, error) {
+	config, err := m.generatePodSandboxConfig(pod, "")
+	if err != nil {
+		return nil, err
+	}
+
+	podSandboxID := config.GetName()
+	readyState := runtimeApi.PodSandBoxState_READY
+	return &apitest.FakePodSandbox{
+		PodSandbox: runtimeApi.PodSandbox{
+			Id:        &podSandboxID,
+			Name:      config.Name,
+			State:     &readyState,
+			CreatedAt: &fakeCreatedAt,
+			Labels:    config.Labels,
+		},
+		Annotations: config.Annotations,
+	}, nil
+}
+
+func makeFakeContainer(m *kubeGenericRuntimeManager, pod *api.Pod, container api.Container, sandboxConfig *runtimeApi.PodSandboxConfig) (*apitest.FakeContainer, error) {
+	containerConfig, err := m.generateContainerConfig(&container, pod, 0, "")
+	if err != nil {
+		return nil, err
+	}
+
+	containerID := containerConfig.GetName()
+	runningState := runtimeApi.ContainerState_RUNNING
+	imageRef := containerConfig.Image.GetImage()
+	return &apitest.FakeContainer{
+		ContainerStatus: runtimeApi.ContainerStatus{
+			Id:          &containerID,
+			Name:        containerConfig.Name,
+			Image:       containerConfig.Image,
+			ImageRef:    &imageRef,
+			CreatedAt:   &fakeCreatedAt,
+			State:       &runningState,
+			Labels:      containerConfig.Labels,
+			Annotations: containerConfig.Annotations,
+		},
+		SandboxID: sandboxConfig.GetName(),
+	}, nil
+}
+
+func makeFakeContainers(m *kubeGenericRuntimeManager, pod *api.Pod, containers []api.Container) ([]*apitest.FakeContainer, error) {
+	sandboxConfig, err := m.generatePodSandboxConfig(pod, "")
+	if err != nil {
+		return nil, err
+	}
+
+	result := make([]*apitest.FakeContainer, len(containers))
+	for idx, c := range containers {
+		containerWithState, err := makeFakeContainer(m, pod, c, sandboxConfig)
+		if err != nil {
+			return nil, err
+		}
+
+		result[idx] = containerWithState
+	}
+
+	return result, nil
+}
+
+// verifyPods returns true if the two pod slices are equal.
+func verifyPods(a, b []*kubecontainer.Pod) bool {
+	if len(a) != len(b) {
+		return false
+	}
+
+	// Sort the containers within a pod.
+	for i := range a {
+		sort.Sort(containersByID(a[i].Containers))
+	}
+	for i := range b {
+		sort.Sort(containersByID(b[i].Containers))
+	}
+
+	// Sort the pods by UID.
+	sort.Sort(podsByID(a))
+	sort.Sort(podsByID(b))
+
+	return reflect.DeepEqual(a, b)
+}
+
 func TestNewKubeRuntimeManager(t *testing.T) {
-	_, _, _, err := createTestFakeRuntimeManager()
+	_, _, _, err := createTestRuntimeManager()
 	assert.NoError(t, err)
 }
 
 func TestVersion(t *testing.T) {
-	_, _, m, err := createTestFakeRuntimeManager()
+	_, _, m, err := createTestRuntimeManager()
 	assert.NoError(t, err)
 
 	version, err := m.Version()
@@ -57,9 +167,71 @@ func TestVersion(t *testing.T) {
 }
 
 func TestContainerRuntimeType(t *testing.T) {
-	_, _, m, err := createTestFakeRuntimeManager()
+	_, _, m, err := createTestRuntimeManager()
 	assert.NoError(t, err)
 
 	runtimeType := m.Type()
 	assert.Equal(t, "fakeRuntime", runtimeType)
 }
+
+func TestGetPods(t *testing.T) {
+	fakeRuntime, _, m, err := createTestRuntimeManager()
+	assert.NoError(t, err)
+
+	pod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			UID:       "12345678",
+			Name:      "foo",
+			Namespace: "new",
+		},
+		Spec: api.PodSpec{
+			Containers: []api.Container{
+				{
+					Name:  "foo1",
+					Image: "busybox",
+				},
+				{
+					Name:  "foo2",
+					Image: "busybox",
+				},
+			},
+		},
+	}
+
+	// Set fake sandbox and fake containers to fakeRuntime.
+	_, fakeContainers, err := makeAndSetFakePod(m, fakeRuntime, pod)
+	assert.NoError(t, err)
+
+	// Convert the fakeContainers to kubecontainer.Container
+	containers := make([]*kubecontainer.Container, len(fakeContainers))
+	for i := range containers {
+		fakeContainer := fakeContainers[i]
+		c, err := m.toKubeContainer(&runtimeApi.Container{
+			Id:       fakeContainer.Id,
+			Name:     fakeContainer.Name,
+			State:    fakeContainer.State,
+			Image:    fakeContainer.Image,
+			ImageRef: fakeContainer.ImageRef,
+			Labels:   fakeContainer.Labels,
+		})
+		if err != nil {
+			t.Fatalf("unexpected error %v", err)
+		}
+		containers[i] = c
+	}
+	expected := []*kubecontainer.Pod{
+		{
+			ID:         kubetypes.UID("12345678"),
+			Name:       "foo",
+			Namespace:  "new",
+			Containers: []*kubecontainer.Container{containers[0], containers[1]},
+		},
+	}
+
+	actual, err := m.GetPods(false)
+	assert.NoError(t, err)
+
+	if !verifyPods(expected, actual) {
+		t.Errorf("expected %#v, got %#v", expected, actual)
+	}
+}

pkg/kubelet/kuberuntime/kuberuntime_sandbox.go

@@ -17,6 +17,7 @@ limitations under the License.
 package kuberuntime
 
 import (
+	"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"
@@ -105,3 +106,32 @@ func generatePodSandboxLinuxConfig(pod *api.Pod, cgroupParent string) *runtimeAp
 
 	return linuxPodSandboxConfig
 }
+
+// getKubeletSandboxes lists all (or just the running) sandboxes managed by kubelet.
+func (m *kubeGenericRuntimeManager) getKubeletSandboxes(all bool) ([]*runtimeApi.PodSandbox, error) {
+	var filter *runtimeApi.PodSandboxFilter
+	if !all {
+		readyState := runtimeApi.PodSandBoxState_READY
+		filter = &runtimeApi.PodSandboxFilter{
+			State: &readyState,
+		}
+	}
+
+	resp, err := m.runtimeService.ListPodSandbox(filter)
+	if err != nil {
+		glog.Errorf("ListPodSandbox failed: %v", err)
+		return nil, err
+	}
+
+	result := []*runtimeApi.PodSandbox{}
+	for _, s := range resp {
+		if !isManagedByKubelet(s.Labels) {
+			glog.V(5).Infof("Sandbox %s is not managed by kubelet", s.GetName())
+			continue
+		}
+
+		result = append(result, s)
+	}
+
+	return result, nil
+}

pkg/kubelet/kuberuntime/labels.go

@@ -39,6 +39,9 @@ const (
 	containerTerminationMessagePathLabel = "io.kubernetes.container.terminationMessagePath"
 	containerPreStopHandlerLabel         = "io.kubernetes.container.preStopHandler"
 	containerPortsLabel                  = "io.kubernetes.container.ports"
+
+	// kubernetesManagedLabel is used to distinguish whether a container/sandbox is managed by kubelet or not
+	kubernetesManagedLabel = "io.kubernetes.managed"
 )
 
 type labeledPodSandboxInfo struct {
@@ -83,6 +86,7 @@ func newPodLabels(pod *api.Pod) map[string]string {
 	labels[types.KubernetesPodNameLabel] = pod.Name
 	labels[types.KubernetesPodNamespaceLabel] = pod.Namespace
 	labels[types.KubernetesPodUIDLabel] = string(pod.UID)
+	labels[kubernetesManagedLabel] = "true"
 
 	return labels
 }
@@ -99,6 +103,7 @@ func newContainerLabels(container *api.Container, pod *api.Pod) map[string]strin
 	labels[types.KubernetesPodNamespaceLabel] = pod.Namespace
 	labels[types.KubernetesPodUIDLabel] = string(pod.UID)
 	labels[types.KubernetesContainerNameLabel] = container.Name
+	labels[kubernetesManagedLabel] = "true"
 
 	return labels
 }
@@ -150,7 +155,7 @@ func getPodSandboxInfoFromLabels(labels map[string]string) *labeledPodSandboxInf
 
 	// Remain only labels from api.Pod
 	for k, v := range labels {
-		if k != types.KubernetesPodNameLabel && k != types.KubernetesPodNamespaceLabel && k != types.KubernetesPodUIDLabel {
+		if k != types.KubernetesPodNameLabel && k != types.KubernetesPodNamespaceLabel && k != types.KubernetesPodUIDLabel && k != kubernetesManagedLabel {
 			podSandboxInfo.Labels[k] = v
 		}
 	}
@@ -175,6 +180,15 @@ func getContainerInfoFromLabels(labels map[string]string) *labeledContainerInfo
 	}
 }
 
+// isManagedByKubelet returns true is the sandbox/container is managed by kubelet.
+func isManagedByKubelet(labels map[string]string) bool {
+	if _, ok := labels[kubernetesManagedLabel]; ok {
+		return true
+	}
+
+	return false
+}
+
 // getContainerInfoFromAnnotations gets annotatedContainerInfo from annotations.
 func getContainerInfoFromAnnotations(annotations map[string]string) *annotatedContainerInfo {
 	var err error