Get machine stats from cAdvisor

pkg/kubelet/kubelet.go

@@ -18,6 +18,7 @@ package kubelet
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"math/rand"
@@ -815,17 +816,27 @@ func (kl *Kubelet) GetPodInfo(podID string) (api.PodInfo, error) {
 	return info, nil
 }
 
-//Returns stats (from Cadvisor) for a container
+// Returns the docker id corresponding to pod-id-container-name pair.
-func (kl *Kubelet) GetContainerStats(name string) (*api.ContainerStats, error) {
+func (kl *Kubelet) getDockerIDFromPodIDAndContainerName(podID, containerName string) (DockerID, error) {
-	if kl.CadvisorClient == nil {
+	containerList, err := kl.DockerClient.ListContainers(docker.ListContainersOptions{})
-		return nil, nil
+	if err != nil {
+		return "", err
 	}
-	dockerID, found, err := kl.getContainerIdFromName(name)
+	for _, value := range containerList {
-	if err != nil || !found {
+		manifestID, cName := parseDockerName(value.Names[0])
-		return nil, err
+		if manifestID == podID && cName == containerName {
+			return DockerID(value.ID), nil
+		}
 	}
+	return "", errors.New("couldn't find container")
+}
 
-	info, err := kl.CadvisorClient.ContainerInfo(fmt.Sprintf("/docker/%s", string(dockerID)))
+// This method takes a container's absolute path and returns the stats for the
+// container.  The container's absolute path refers to its hierarchy in the
+// cgroup file system. e.g. The root container, which represents the whole
+// machine, has path "/"; all docker containers have path "/docker/<docker id>"
+func (kl *Kubelet) statsFromContainerPath(containerPath string) (*api.ContainerStats, error) {
+	info, err := kl.CadvisorClient.ContainerInfo(containerPath)
 
 	if err != nil {
 		return nil, err
@@ -855,3 +866,20 @@ func (kl *Kubelet) GetContainerStats(name string) (*api.ContainerStats, error) {
 	}
 	return ret, nil
 }
+
+// Returns stats (from Cadvisor) for a container.
+func (kl *Kubelet) GetContainerStats(podID, containerName string) (*api.ContainerStats, error) {
+	if kl.CadvisorClient == nil {
+		return nil, nil
+	}
+	dockerID, err := kl.getDockerIDFromPodIDAndContainerName(podID, containerName)
+	if err != nil || len(dockerID) == 0 {
+		return nil, err
+	}
+	return kl.statsFromContainerPath(fmt.Sprintf("/docker/%s", string(dockerID)))
+}
+
+// Returns stats (from Cadvisor) of current machine.
+func (kl *Kubelet) GetMachineStats() (*api.ContainerStats, error) {
+	return kl.statsFromContainerPath("/")
+}

pkg/kubelet/kubelet_server.go

@@ -18,10 +18,13 @@ package kubelet
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	"path"
+	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
@@ -36,7 +39,8 @@ type KubeletServer struct {
 // kubeletInterface contains all the kubelet methods required by the server.
 // For testablitiy.
 type kubeletInterface interface {
-	GetContainerStats(name string) (*api.ContainerStats, error)
+	GetContainerStats(podID, containerName string) (*api.ContainerStats, error)
+	GetMachineStats() (*api.ContainerStats, error)
 	GetPodInfo(name string) (api.PodInfo, error)
 }
 
@@ -81,34 +85,6 @@ func (s *KubeletServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 			}
 			s.UpdateChannel <- manifestUpdate{httpServerSource, manifests}
 		}
-	case u.Path == "/containerStats":
-		// NOTE: The master appears to pass a Pod.ID
-		container := u.Query().Get("container")
-		if len(container) == 0 {
-			w.WriteHeader(http.StatusBadRequest)
-			fmt.Fprint(w, "Missing container query arg.")
-			return
-		}
-		stats, err := s.Kubelet.GetContainerStats(container)
-		if err != nil {
-			w.WriteHeader(http.StatusInternalServerError)
-			fmt.Fprintf(w, "Internal Error: %v", err)
-			return
-		}
-		if stats == nil {
-			w.WriteHeader(http.StatusOK)
-			fmt.Fprint(w, "{}")
-			return
-		}
-		data, err := json.Marshal(stats)
-		if err != nil {
-			w.WriteHeader(http.StatusInternalServerError)
-			fmt.Fprintf(w, "Internal Error: %v", err)
-			return
-		}
-		w.WriteHeader(http.StatusOK)
-		w.Header().Add("Content-type", "application/json")
-		w.Write(data)
 	case u.Path == "/podInfo":
 		podID := u.Query().Get("podID")
 		if len(podID) == 0 {
@@ -131,8 +107,52 @@ func (s *KubeletServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		w.WriteHeader(http.StatusOK)
 		w.Header().Add("Content-type", "application/json")
 		w.Write(data)
+	case strings.HasPrefix(u.Path, "/stats"):
+		s.serveStats(w, req)
 	default:
 		w.WriteHeader(http.StatusNotFound)
 		fmt.Fprint(w, "Not found.")
 	}
 }
+
+func (s *KubeletServer) serveStats(w http.ResponseWriter, req *http.Request) {
+	// /stats/<podid>/<containerName>
+	components := strings.Split(strings.TrimPrefix(path.Clean(req.URL.Path), "/"), "/")
+	var stats *api.ContainerStats
+	var err error
+	switch len(components) {
+	case 1:
+		// Machine stats
+		stats, err = s.Kubelet.GetMachineStats()
+	case 2:
+		// pod stats
+		// TODO(monnand) Implement this
+		errors.New("pod level status currently unimplemented")
+	case 3:
+		stats, err = s.Kubelet.GetContainerStats(components[1], components[2])
+	default:
+		w.WriteHeader(http.StatusNotFound)
+		fmt.Fprint(w, "unknown resource.")
+		return
+	}
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(w, "Internal Error: %v", err)
+		return
+	}
+	if stats == nil {
+		w.WriteHeader(http.StatusOK)
+		fmt.Fprint(w, "{}")
+		return
+	}
+	data, err := json.Marshal(stats)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(w, "Internal Error: %v", err)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Header().Add("Content-type", "application/json")
+	w.Write(data)
+	return
+}

pkg/kubelet/kubelet_server_test.go

@@ -32,16 +32,21 @@ import (
 )
 
 type fakeKubelet struct {
-	infoFunc  func(name string) (api.PodInfo, error)
+	infoFunc           func(name string) (api.PodInfo, error)
-	statsFunc func(name string) (*api.ContainerStats, error)
+	containerStatsFunc func(podID, containerName string) (*api.ContainerStats, error)
+	machineStatsFunc   func() (*api.ContainerStats, error)
 }
 
 func (fk *fakeKubelet) GetPodInfo(name string) (api.PodInfo, error) {
 	return fk.infoFunc(name)
 }
 
-func (fk *fakeKubelet) GetContainerStats(name string) (*api.ContainerStats, error) {
+func (fk *fakeKubelet) GetContainerStats(podID, containerName string) (*api.ContainerStats, error) {
-	return fk.statsFunc(name)
+	return fk.containerStatsFunc(podID, containerName)
+}
+
+func (fk *fakeKubelet) GetMachineStats() (*api.ContainerStats, error) {
+	return fk.machineStatsFunc()
 }
 
 type serverTestFramework struct {
@@ -156,15 +161,51 @@ func TestContainerStats(t *testing.T) {
 			{90, 190},
 		},
 	}
+	expectedPodID := "somepod"
 	expectedContainerName := "goodcontainer"
-	fw.fakeKubelet.statsFunc = func(name string) (*api.ContainerStats, error) {
+	fw.fakeKubelet.containerStatsFunc = func(podID, containerName string) (*api.ContainerStats, error) {
-		if name != expectedContainerName {
+		if podID != expectedPodID || containerName != expectedContainerName {
-			return nil, fmt.Errorf("bad container name: %v", name)
+			return nil, fmt.Errorf("bad podID or containerName: podID=%v; containerName=%v", podID, containerName)
 		}
 		return expectedStats, nil
 	}
 
-	resp, err := http.Get(fw.testHttpServer.URL + fmt.Sprintf("/containerStats?container=%v", expectedContainerName))
+	resp, err := http.Get(fw.testHttpServer.URL + fmt.Sprintf("/stats/%v/%v", expectedPodID, expectedContainerName))
+	if err != nil {
+		t.Fatalf("Got error GETing: %v", err)
+	}
+	defer resp.Body.Close()
+	var receivedStats api.ContainerStats
+	decoder := json.NewDecoder(resp.Body)
+	err = decoder.Decode(&receivedStats)
+	if err != nil {
+		t.Fatalf("received invalid json data: %v", err)
+	}
+	if !reflect.DeepEqual(&receivedStats, expectedStats) {
+		t.Errorf("received wrong data: %#v", receivedStats)
+	}
+}
+
+func TestMachineStats(t *testing.T) {
+	fw := makeServerTest()
+	expectedStats := &api.ContainerStats{
+		MaxMemoryUsage: 1024001,
+		CpuUsagePercentiles: []api.Percentile{
+			{50, 150},
+			{80, 180},
+			{90, 190},
+		},
+		MemoryUsagePercentiles: []api.Percentile{
+			{50, 150},
+			{80, 180},
+			{90, 190},
+		},
+	}
+	fw.fakeKubelet.machineStatsFunc = func() (*api.ContainerStats, error) {
+		return expectedStats, nil
+	}
+
+	resp, err := http.Get(fw.testHttpServer.URL + "/stats")
 	if err != nil {
 		t.Fatalf("Got error GETing: %v", err)
 	}

pkg/kubelet/kubelet_test.go

@@ -892,12 +892,55 @@ func TestGetContainerStats(t *testing.T) {
 	kubelet.CadvisorClient = mockCadvisor
 	fakeDocker.containerList = []docker.APIContainers{
 		{
-			Names: []string{"foo"},
+			ID: containerID,
-			ID:    containerID,
+			// pod id: qux
+			// container id: foo
+			Names: []string{"/k8s--foo--qux--1234"},
 		},
 	}
 
-	stats, err := kubelet.GetContainerStats("foo")
+	stats, err := kubelet.GetContainerStats("qux", "foo")
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	if stats.MaxMemoryUsage != containerInfo.StatsPercentiles.MaxMemoryUsage {
+		t.Errorf("wrong max memory usage")
+	}
+	areSamePercentiles(containerInfo.StatsPercentiles.CpuUsagePercentiles, stats.CpuUsagePercentiles, t)
+	areSamePercentiles(containerInfo.StatsPercentiles.MemoryUsagePercentiles, stats.MemoryUsagePercentiles, t)
+	mockCadvisor.AssertExpectations(t)
+}
+
+func TestGetMachineStats(t *testing.T) {
+	containerPath := "/"
+	containerInfo := &info.ContainerInfo{
+		ContainerReference: info.ContainerReference{
+			Name: containerPath,
+		}, StatsPercentiles: &info.ContainerStatsPercentiles{MaxMemoryUsage: 1024000, MemoryUsagePercentiles: []info.Percentile{{50, 100}, {80, 180},
+			{90, 190},
+		},
+			CpuUsagePercentiles: []info.Percentile{
+				{51, 101},
+				{81, 181},
+				{91, 191},
+			},
+		},
+	}
+	fakeDocker := FakeDockerClient{
+		err: nil,
+	}
+
+	mockCadvisor := &mockCadvisorClient{}
+	mockCadvisor.On("ContainerInfo", containerPath).Return(containerInfo, nil)
+
+	kubelet := Kubelet{
+		DockerClient:   &fakeDocker,
+		DockerPuller:   &FakeDockerPuller{},
+		CadvisorClient: mockCadvisor,
+	}
+
+	// If the container name is an empty string, then it means the root container.
+	stats, err := kubelet.GetMachineStats()
 	if err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
@@ -913,11 +956,14 @@ func TestGetContainerStatsWithoutCadvisor(t *testing.T) {
 	kubelet, _, fakeDocker := makeTestKubelet(t)
 	fakeDocker.containerList = []docker.APIContainers{
 		{
-			Names: []string{"foo"},
+			ID: "foobar",
+			// pod id: qux
+			// container id: foo
+			Names: []string{"/k8s--foo--qux--1234"},
 		},
 	}
 
-	stats, _ := kubelet.GetContainerStats("foo")
+	stats, _ := kubelet.GetContainerStats("qux", "foo")
 	// When there's no cAdvisor, the stats should be either nil or empty
 	if stats == nil {
 		return
@@ -946,12 +992,14 @@ func TestGetContainerStatsWhenCadvisorFailed(t *testing.T) {
 	kubelet.CadvisorClient = mockCadvisor
 	fakeDocker.containerList = []docker.APIContainers{
 		{
-			Names: []string{"foo"},
+			ID: containerID,
-			ID:    containerID,
+			// pod id: qux
+			// container id: foo
+			Names: []string{"/k8s--foo--qux--1234"},
 		},
 	}
 
-	stats, err := kubelet.GetContainerStats("foo")
+	stats, err := kubelet.GetContainerStats("qux", "foo")
 	if stats != nil {
 		t.Errorf("non-nil stats on error")
 	}
@@ -972,7 +1020,7 @@ func TestGetContainerStatsOnNonExistContainer(t *testing.T) {
 	kubelet.CadvisorClient = mockCadvisor
 	fakeDocker.containerList = []docker.APIContainers{}
 
-	stats, _ := kubelet.GetContainerStats("foo")
+	stats, _ := kubelet.GetContainerStats("qux", "foo")
 	if stats != nil {
 		t.Errorf("non-nil stats on non exist container")
 	}