Merge pull request #174 from monnand/kubelet-cadvisor

Letting kubelet retrieve container stats from cAdvisor

pkg/api/types.go

@@ -68,6 +68,20 @@ type Container struct {
 	VolumeMounts []VolumeMount `yaml:"volumeMounts,omitempty" json:"volumeMounts,omitempty"`
 }
 
+// Percentile represents a pair which contains a percentage from 0 to 100 and
+// its corresponding value.
+type Percentile struct {
+	Percentage int    `json:"percentage,omitempty"`
+	Value      uint64 `json:"value,omitempty"`
+}
+
+// ContainerStats represents statistical information of a container
+type ContainerStats struct {
+	CpuUsagePercentiles    []Percentile `json:"cpu_usage_percentiles,omitempty"`
+	MemoryUsagePercentiles []Percentile `json:"memory_usage_percentiles,omitempty"`
+	MaxMemoryUsage         uint64       `json:"max_memory_usage,omitempty"`
+}
+
 // Event is the representation of an event logged to etcd backends
 type Event struct {
 	Event     string             `json:"event,omitempty"`

pkg/kubelet/kubelet.go

@@ -36,6 +36,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/coreos/go-etcd/etcd"
 	"github.com/fsouza/go-dockerclient"
+	"github.com/google/cadvisor/info"
 	"gopkg.in/v1/yaml"
 )
 
@@ -58,11 +59,17 @@ type DockerInterface interface {
 	StopContainer(id string, timeout uint) error
 }
 
+type CadvisorInterface interface {
+	ContainerInfo(name string) (*info.ContainerInfo, error)
+	MachineInfo() (*info.MachineInfo, error)
+}
+
 // The main kubelet implementation
 type Kubelet struct {
 	Hostname           string
 	Client             util.EtcdClient
 	DockerClient       DockerInterface
+	CadvisorClient     CadvisorInterface
 	FileCheckFrequency time.Duration
 	SyncFrequency      time.Duration
 	HTTPCheckFrequency time.Duration
@@ -641,3 +648,43 @@ func (kl *Kubelet) GetContainerInfo(name string) (string, error) {
 	data, err := json.Marshal(info)
 	return string(data), err
 }
+
+func (kl *Kubelet) GetContainerStats(name string) (*api.ContainerStats, error) {
+	if kl.CadvisorClient == nil {
+		return nil, nil
+	}
+	id, found, err := kl.GetContainerID(name)
+	if err != nil || !found {
+		return nil, err
+	}
+
+	info, err := kl.CadvisorClient.ContainerInfo(fmt.Sprintf("/docker/%v", id))
+
+	if err != nil {
+		return nil, err
+	}
+	// When the stats data for the container is not available yet.
+	if info.StatsPercentiles == nil {
+		return nil, nil
+	}
+
+	ret := new(api.ContainerStats)
+	ret.MaxMemoryUsage = info.StatsPercentiles.MaxMemoryUsage
+	if len(info.StatsPercentiles.CpuUsagePercentiles) > 0 {
+		percentiles := make([]api.Percentile, len(info.StatsPercentiles.CpuUsagePercentiles))
+		for i, p := range info.StatsPercentiles.CpuUsagePercentiles {
+			percentiles[i].Percentage = p.Percentage
+			percentiles[i].Value = p.Value
+		}
+		ret.CpuUsagePercentiles = percentiles
+	}
+	if len(info.StatsPercentiles.MemoryUsagePercentiles) > 0 {
+		percentiles := make([]api.Percentile, len(info.StatsPercentiles.MemoryUsagePercentiles))
+		for i, p := range info.StatsPercentiles.MemoryUsagePercentiles {
+			percentiles[i].Percentage = p.Percentage
+			percentiles[i].Value = p.Value
+		}
+		ret.MemoryUsagePercentiles = percentiles
+	}
+	return ret, nil
+}

pkg/kubelet/kubelet_server.go

@@ -17,6 +17,7 @@ limitations under the License.
 package kubelet
 
 import (
+	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -35,6 +36,7 @@ type KubeletServer struct {
 // For testablitiy.
 type kubeletInterface interface {
 	GetContainerID(name string) (string, bool, error)
+	GetContainerStats(name string) (*api.ContainerStats, error)
 	GetContainerInfo(name string) (string, error)
 }
 
@@ -64,6 +66,33 @@ func (s *KubeletServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 			return
 		}
 		s.UpdateChannel <- manifest
+	case u.Path == "/containerStats":
+		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 == "/containerInfo":
 		container := u.Query().Get("container")
 		if len(container) == 0 {

pkg/kubelet/kubelet_server_test.go

@@ -2,6 +2,7 @@ package kubelet
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -15,8 +16,9 @@ import (
 )
 
 type fakeKubelet struct {
-	infoFunc func(name string) (string, error)
-	idFunc   func(name string) (string, bool, error)
+	infoFunc  func(name string) (string, error)
+	idFunc    func(name string) (string, bool, error)
+	statsFunc func(name string) (*api.ContainerStats, error)
 }
 
 func (fk *fakeKubelet) GetContainerInfo(name string) (string, error) {
@@ -27,6 +29,10 @@ func (fk *fakeKubelet) GetContainerID(name string) (string, bool, error) {
 	return fk.idFunc(name)
 }
 
+func (fk *fakeKubelet) GetContainerStats(name string) (*api.ContainerStats, error) {
+	return fk.statsFunc(name)
+}
+
 // If we made everything distribute a list of ContainerManifests, we could just use
 // channelReader.
 type channelReaderSingle struct {
@@ -129,3 +135,42 @@ func TestContainerInfo(t *testing.T) {
 		t.Errorf("Expected: '%v', got: '%v'", expected, got)
 	}
 }
+
+func TestContainerStats(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},
+		},
+	}
+	expectedContainerName := "goodcontainer"
+	fw.fakeKubelet.statsFunc = func(name string) (*api.ContainerStats, error) {
+		if name != expectedContainerName {
+			return nil, fmt.Errorf("bad container name: %v", name)
+		}
+		return expectedStats, nil
+	}
+
+	resp, err := http.Get(fw.testHttpServer.URL + fmt.Sprintf("/containerStats?container=%v", 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)
+	}
+}

pkg/kubelet/kubelet_test.go

@@ -29,6 +29,8 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/coreos/go-etcd/etcd"
 	"github.com/fsouza/go-dockerclient"
+	"github.com/google/cadvisor/info"
+	"github.com/stretchr/testify/mock"
 )
 
 // TODO: This doesn't reduce typing enough to make it worth the less readable errors. Remove.
@@ -905,3 +907,183 @@ func TestWatchEtcd(t *testing.T) {
 		t.Errorf("Unexpected manifest(s) %#v %#v", read[0], manifest)
 	}
 }
+
+type mockCadvisorClient struct {
+	mock.Mock
+}
+
+func (self *mockCadvisorClient) ContainerInfo(name string) (*info.ContainerInfo, error) {
+	args := self.Called(name)
+	return args.Get(0).(*info.ContainerInfo), args.Error(1)
+}
+
+func (self *mockCadvisorClient) MachineInfo() (*info.MachineInfo, error) {
+	args := self.Called()
+	return args.Get(0).(*info.MachineInfo), args.Error(1)
+}
+
+func areSamePercentiles(
+	cadvisorPercentiles []info.Percentile,
+	kubePercentiles []api.Percentile,
+	t *testing.T,
+) {
+	if len(cadvisorPercentiles) != len(kubePercentiles) {
+		t.Errorf("cadvisor gives %v percentiles; kubelet got %v", len(cadvisorPercentiles), len(kubePercentiles))
+		return
+	}
+	for _, ap := range cadvisorPercentiles {
+		found := false
+		for _, kp := range kubePercentiles {
+			if ap.Percentage == kp.Percentage {
+				found = true
+				if ap.Value != kp.Value {
+					t.Errorf("%v percentile from cadvisor is %v; kubelet got %v",
+						ap.Percentage,
+						ap.Value,
+						kp.Value)
+				}
+			}
+		}
+		if !found {
+			t.Errorf("Unable to find %v percentile in kubelet's data", ap.Percentage)
+		}
+	}
+}
+
+func TestGetContainerStats(t *testing.T) {
+	containerId := "ab2cdf"
+	containerPath := fmt.Sprintf("/docker/%v", containerId)
+	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,
+		CadvisorClient: mockCadvisor,
+	}
+	fakeDocker.containerList = []docker.APIContainers{
+		{
+			Names: []string{"foo"},
+			ID:    containerId,
+		},
+	}
+
+	stats, err := kubelet.GetContainerStats("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 TestGetContainerStatsWithoutCadvisor(t *testing.T) {
+	fakeDocker := FakeDockerClient{
+		err: nil,
+	}
+
+	kubelet := Kubelet{
+		DockerClient: &fakeDocker,
+	}
+	fakeDocker.containerList = []docker.APIContainers{
+		{
+			Names: []string{"foo"},
+		},
+	}
+
+	stats, _ := kubelet.GetContainerStats("foo")
+	// When there's no cAdvisor, the stats should be either nil or empty
+	if stats == nil {
+		return
+	}
+	if stats.MaxMemoryUsage != 0 {
+		t.Errorf("MaxMemoryUsage is %v even if there's no cadvisor", stats.MaxMemoryUsage)
+	}
+	if len(stats.CpuUsagePercentiles) > 0 {
+		t.Errorf("Cpu usage percentiles is not empty (%+v) even if there's no cadvisor", stats.CpuUsagePercentiles)
+	}
+	if len(stats.MemoryUsagePercentiles) > 0 {
+		t.Errorf("Memory usage percentiles is not empty (%+v) even if there's no cadvisor", stats.MemoryUsagePercentiles)
+	}
+}
+
+func TestGetContainerStatsWhenCadvisorFailed(t *testing.T) {
+	containerId := "ab2cdf"
+	containerPath := fmt.Sprintf("/docker/%v", containerId)
+	fakeDocker := FakeDockerClient{
+		err: nil,
+	}
+
+	containerInfo := &info.ContainerInfo{}
+	mockCadvisor := &mockCadvisorClient{}
+	expectedErr := fmt.Errorf("some error")
+	mockCadvisor.On("ContainerInfo", containerPath).Return(containerInfo, expectedErr)
+
+	kubelet := Kubelet{
+		DockerClient:   &fakeDocker,
+		CadvisorClient: mockCadvisor,
+	}
+	fakeDocker.containerList = []docker.APIContainers{
+		{
+			Names: []string{"foo"},
+			ID:    containerId,
+		},
+	}
+
+	stats, err := kubelet.GetContainerStats("foo")
+	if stats != nil {
+		t.Errorf("non-nil stats on error")
+	}
+	if err == nil {
+		t.Errorf("expect error but received nil error")
+		return
+	}
+	if err.Error() != expectedErr.Error() {
+		t.Errorf("wrong error message. expect %v, got %v", err, expectedErr)
+	}
+	mockCadvisor.AssertExpectations(t)
+}
+
+func TestGetContainerStatsOnNonExistContainer(t *testing.T) {
+	fakeDocker := FakeDockerClient{
+		err: nil,
+	}
+
+	mockCadvisor := &mockCadvisorClient{}
+
+	kubelet := Kubelet{
+		DockerClient:   &fakeDocker,
+		CadvisorClient: mockCadvisor,
+	}
+	fakeDocker.containerList = []docker.APIContainers{}
+
+	stats, _ := kubelet.GetContainerStats("foo")
+	if stats != nil {
+		t.Errorf("non-nil stats on non exist container")
+	}
+	mockCadvisor.AssertExpectations(t)
+}