Kubelet Metrics Summary Api Implementation

pkg/kubelet/cadvisor/cadvisor_fake.go

@@ -36,6 +36,10 @@ func (c *Fake) ContainerInfo(name string, req *cadvisorapi.ContainerInfoRequest)
 	return new(cadvisorapi.ContainerInfo), nil
 }
 
+func (c *Fake) ContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error) {
+	return map[string]cadvisorapiv2.ContainerInfo{}, nil
+}
+
 func (c *Fake) SubcontainerInfo(name string, req *cadvisorapi.ContainerInfoRequest) (map[string]*cadvisorapi.ContainerInfo, error) {
 	return map[string]*cadvisorapi.ContainerInfo{}, nil
 }

pkg/kubelet/cadvisor/cadvisor_linux.go

@@ -137,6 +137,10 @@ func (cc *cadvisorClient) ContainerInfo(name string, req *cadvisorapi.ContainerI
 	return cc.GetContainerInfo(name, req)
 }
 
+func (cc *cadvisorClient) ContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error) {
+	return cc.GetContainerInfoV2(name, options)
+}
+
 func (cc *cadvisorClient) VersionInfo() (*cadvisorapi.VersionInfo, error) {
 	return cc.GetVersionInfo()
 }

pkg/kubelet/cadvisor/cadvisor_mock.go

@@ -40,6 +40,12 @@ func (c *Mock) ContainerInfo(name string, req *cadvisorapi.ContainerInfoRequest)
 	return args.Get(0).(*cadvisorapi.ContainerInfo), args.Error(1)
 }
 
+// ContainerInfoV2 is a mock implementation of Interface.ContainerInfoV2.
+func (c *Mock) ContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error) {
+	args := c.Called(name, options)
+	return args.Get(0).(map[string]cadvisorapiv2.ContainerInfo), args.Error(1)
+}
+
 func (c *Mock) SubcontainerInfo(name string, req *cadvisorapi.ContainerInfoRequest) (map[string]*cadvisorapi.ContainerInfo, error) {
 	args := c.Called(name, req)
 	return args.Get(0).(map[string]*cadvisorapi.ContainerInfo), args.Error(1)

pkg/kubelet/cadvisor/cadvisor_unsupported.go

@@ -49,6 +49,10 @@ func (cu *cadvisorUnsupported) ContainerInfo(name string, req *cadvisorapi.Conta
 	return nil, unsupportedErr
 }
 
+func (cu *cadvisorUnsupported) ContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error) {
+	return nil, unsupportedErr
+}
+
 func (cu *cadvisorUnsupported) SubcontainerInfo(name string, req *cadvisorapi.ContainerInfoRequest) (map[string]*cadvisorapi.ContainerInfo, error) {
 	return nil, unsupportedErr
 }

pkg/kubelet/cadvisor/types.go

@@ -27,6 +27,7 @@ type Interface interface {
 	Start() error
 	DockerContainer(name string, req *cadvisorapi.ContainerInfoRequest) (cadvisorapi.ContainerInfo, error)
 	ContainerInfo(name string, req *cadvisorapi.ContainerInfoRequest) (*cadvisorapi.ContainerInfo, error)
+	ContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error)
 	SubcontainerInfo(name string, req *cadvisorapi.ContainerInfoRequest) (map[string]*cadvisorapi.ContainerInfo, error)
 	MachineInfo() (*cadvisorapi.MachineInfo, error)
 

pkg/kubelet/dockertools/labels.go

@@ -64,6 +64,22 @@ type labelledContainerInfo struct {
 	PreStopHandler            *api.Handler
 }
 
+func GetContainerName(labels map[string]string) string {
+	return labels[kubernetesContainerNameLabel]
+}
+
+func GetPodName(labels map[string]string) string {
+	return labels[kubernetesPodNameLabel]
+}
+
+func GetPodUID(labels map[string]string) string {
+	return labels[kubernetesPodUIDLabel]
+}
+
+func GetPodNamespace(labels map[string]string) string {
+	return labels[kubernetesPodNamespaceLabel]
+}
+
 func newLabels(container *api.Container, pod *api.Pod, restartCount int) map[string]string {
 	labels := map[string]string{}
 	labels[kubernetesPodNameLabel] = pod.Name

pkg/kubelet/kubelet.go

@@ -34,6 +34,7 @@ import (
 
 	"github.com/golang/glog"
 	cadvisorapi "github.com/google/cadvisor/info/v1"
+	cadvisorapiv2 "github.com/google/cadvisor/info/v2"
 	"k8s.io/kubernetes/pkg/api"
 	apierrors "k8s.io/kubernetes/pkg/api/errors"
 	"k8s.io/kubernetes/pkg/api/resource"
@@ -3413,6 +3414,19 @@ func (kl *Kubelet) GetContainerInfo(podFullName string, podUID types.UID, contai
 	return &ci, nil
 }
 
+// GetContainerInfoV2 returns stats (from Cadvisor) for containers.
+func (kl *Kubelet) GetContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error) {
+	return kl.cadvisor.ContainerInfoV2(name, options)
+}
+
+func (kl *Kubelet) DockerImagesFsInfo() (cadvisorapiv2.FsInfo, error) {
+	return kl.cadvisor.DockerImagesFsInfo()
+}
+
+func (kl *Kubelet) RootFsInfo() (cadvisorapiv2.FsInfo, error) {
+	return kl.cadvisor.RootFsInfo()
+}
+
 // Returns stats (from Cadvisor) for a non-Kubernetes container.
 func (kl *Kubelet) GetRawContainerInfo(containerName string, req *cadvisorapi.ContainerInfoRequest, subcontainers bool) (map[string]*cadvisorapi.ContainerInfo, error) {
 	if subcontainers {
@@ -3468,6 +3482,9 @@ func (kl *Kubelet) updatePodCIDR(cidr string) {
 		kl.networkPlugin.Event(network.NET_PLUGIN_EVENT_POD_CIDR_CHANGE, details)
 	}
 }
+func (kl *Kubelet) GetNodeConfig() cm.NodeConfig {
+	return kl.nodeConfig
+}
 
 var minRsrc = resource.MustParse("1k")
 var maxRsrc = resource.MustParse("1P")

pkg/kubelet/server/server.go

@@ -33,6 +33,7 @@ import (
 	restful "github.com/emicklei/go-restful"
 	"github.com/golang/glog"
 	cadvisorapi "github.com/google/cadvisor/info/v1"
+	cadvisorapiv2 "github.com/google/cadvisor/info/v2"
 	"github.com/prometheus/client_golang/prometheus"
 
 	"k8s.io/kubernetes/pkg/api"
@@ -45,6 +46,7 @@ import (
 	"k8s.io/kubernetes/pkg/client/unversioned/remotecommand"
 	"k8s.io/kubernetes/pkg/healthz"
 	"k8s.io/kubernetes/pkg/httplog"
+	"k8s.io/kubernetes/pkg/kubelet/cm"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/server/portforward"
 	"k8s.io/kubernetes/pkg/kubelet/server/stats"
@@ -140,6 +142,7 @@ type AuthInterface interface {
 // For testablitiy.
 type HostInterface interface {
 	GetContainerInfo(podFullName string, uid types.UID, containerName string, req *cadvisorapi.ContainerInfoRequest) (*cadvisorapi.ContainerInfo, error)
+	GetContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error)
 	GetRawContainerInfo(containerName string, req *cadvisorapi.ContainerInfoRequest, subcontainers bool) (map[string]*cadvisorapi.ContainerInfo, error)
 	GetCachedMachineInfo() (*cadvisorapi.MachineInfo, error)
 	GetPods() []*api.Pod
@@ -154,7 +157,11 @@ type HostInterface interface {
 	StreamingConnectionIdleTimeout() time.Duration
 	ResyncInterval() time.Duration
 	GetHostname() string
+	GetNode() (*api.Node, error)
+	GetNodeConfig() cm.NodeConfig
 	LatestLoopEntryTime() time.Time
+	DockerImagesFsInfo() (cadvisorapiv2.FsInfo, error)
+	RootFsInfo() (cadvisorapiv2.FsInfo, error)
 }
 
 // NewServer initializes and configures a kubelet.Server object to handle HTTP requests.

pkg/kubelet/server/server_test.go

@@ -34,10 +34,12 @@ import (
 	"time"
 
 	cadvisorapi "github.com/google/cadvisor/info/v1"
+	cadvisorapiv2 "github.com/google/cadvisor/info/v2"
 	"k8s.io/kubernetes/pkg/api"
 	apierrs "k8s.io/kubernetes/pkg/api/errors"
 	"k8s.io/kubernetes/pkg/auth/authorizer"
 	"k8s.io/kubernetes/pkg/auth/user"
+	"k8s.io/kubernetes/pkg/kubelet/cm"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/types"
@@ -129,6 +131,22 @@ func (fk *fakeKubelet) StreamingConnectionIdleTimeout() time.Duration {
 	return fk.streamingConnectionIdleTimeoutFunc()
 }
 
+// Unused functions
+func (_ *fakeKubelet) GetContainerInfoV2(_ string, _ cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error) {
+	return nil, nil
+}
+
+func (_ *fakeKubelet) DockerImagesFsInfo() (cadvisorapiv2.FsInfo, error) {
+	return cadvisorapiv2.FsInfo{}, fmt.Errorf("Unsupported Operation DockerImagesFsInfo")
+}
+
+func (_ *fakeKubelet) RootFsInfo() (cadvisorapiv2.FsInfo, error) {
+	return cadvisorapiv2.FsInfo{}, fmt.Errorf("Unsupport Operation RootFsInfo")
+}
+
+func (_ *fakeKubelet) GetNode() (*api.Node, error)  { return nil, nil }
+func (_ *fakeKubelet) GetNodeConfig() cm.NodeConfig { return cm.NodeConfig{} }
+
 type fakeAuth struct {
 	authenticateFunc func(*http.Request) (user.Info, bool, error)
 	attributesFunc   func(user.Info, *http.Request) authorizer.Attributes

pkg/kubelet/server/stats/handler.go

@@ -27,8 +27,10 @@ import (
 	"github.com/emicklei/go-restful"
 	"github.com/golang/glog"
 	cadvisorapi "github.com/google/cadvisor/info/v1"
+	cadvisorapiv2 "github.com/google/cadvisor/info/v2"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/kubelet/cm"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/types"
 )
@@ -36,16 +38,22 @@ import (
 // Host methods required by stats handlers.
 type StatsProvider interface {
 	GetContainerInfo(podFullName string, uid types.UID, containerName string, req *cadvisorapi.ContainerInfoRequest) (*cadvisorapi.ContainerInfo, error)
+	GetContainerInfoV2(name string, options cadvisorapiv2.RequestOptions) (map[string]cadvisorapiv2.ContainerInfo, error)
 	GetRawContainerInfo(containerName string, req *cadvisorapi.ContainerInfoRequest, subcontainers bool) (map[string]*cadvisorapi.ContainerInfo, error)
 	GetPodByName(namespace, name string) (*api.Pod, bool)
+	GetNode() (*api.Node, error)
+	GetNodeConfig() cm.NodeConfig
+	DockerImagesFsInfo() (cadvisorapiv2.FsInfo, error)
+	RootFsInfo() (cadvisorapiv2.FsInfo, error)
 }
 
 type handler struct {
-	provider StatsProvider
+	provider        StatsProvider
+	summaryProvider SummaryProvider
 }
 
 func CreateHandlers(provider StatsProvider) *restful.WebService {
-	h := &handler{provider}
+	h := &handler{provider, NewSummaryProvider(provider)}
 
 	ws := &restful.WebService{}
 	ws.Path("/stats/").
@@ -137,11 +145,12 @@ func (h *handler) handleStats(request *restful.Request, response *restful.Respon
 
 // Handles stats summary requests to /stats/summary
 func (h *handler) handleSummary(request *restful.Request, response *restful.Response) {
-	summary := Summary{}
-
-	// TODO(timstclair): Fill in summary from cAdvisor v2 endpoint.
-
-	writeResponse(response, summary)
+	summary, err := h.summaryProvider.Get()
+	if err != nil {
+		handleError(response, err)
+	} else {
+		writeResponse(response, summary)
+	}
 }
 
 // Handles non-kubernetes container stats requests to /stats/container/

pkg/kubelet/server/stats/types.go

@@ -91,7 +91,7 @@ type ContainerStats struct {
 	// Stats pertaining to container logs usage of filesystem resources.
 	// Logs.UsedBytes is the number of bytes used for the container logs.
 	Logs *FsStats `json:"logs,omitempty"`
-	// User defined metrics are arbitrary metrics exposed by containers in pods.
+	// User defined metrics that are exposed by containers in the pod. Typically, we expect only one container in the pod to be exposing user defined metrics. In the event of multiple containers exposing metrics, they will be combined here.
 	UserDefinedMetrics []UserDefinedMetric `json:"userDefinedMetrics,omitmepty" patchStrategy:"merge" patchMergeKey:"name"`
 }
 

test/e2e_node/e2e_node_suite_test.go

@@ -15,7 +15,8 @@ limitations under the License.
 */
 
 // To run tests in this suite
-// `$ ginkgo -- --node-name node-e2e-test-1  --api-server-address <serveraddress> --logtostderr`
+// Local: `$ ginkgo -- --logtostderr -v 2`
+// Remote: `$ ginkgo -- --node-name <hostname> --api-server-address=<hostname:api_port> --kubelet-address=<hostname=kubelet_port> --logtostderr -v 2`
 package e2e_node
 
 import (
@@ -26,9 +27,9 @@ import (
 	"testing"
 )
 
-var kubeletAddress = flag.String("kubelet-address", "localhost:10250", "Host and port of the kubelet")
-var apiServerAddress = flag.String("api-server-address", "localhost:8080", "Host and port of the api server")
-var nodeName = flag.String("node-name", "", "Name of the node")
+var kubeletAddress = flag.String("kubelet-address", "http://127.0.0.1:10255", "Host and port of the kubelet")
+var apiServerAddress = flag.String("api-server-address", "http://127.0.0.1:8080", "Host and port of the api server")
+var nodeName = flag.String("node-name", "127.0.0.1", "Name of the node")
 
 func TestE2eNode(t *testing.T) {
 	flag.Parse()

test/e2e_node/kubelet_test.go

@@ -18,13 +18,18 @@ package e2e_node
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
+	"net/http"
+	"strings"
 	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"io/ioutil"
 	"k8s.io/kubernetes/pkg/api"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
+	"k8s.io/kubernetes/pkg/kubelet/server/stats"
 )
 
 var _ = Describe("Kubelet", func() {
@@ -49,10 +54,9 @@ var _ = Describe("Kubelet", func() {
 						RestartPolicy: api.RestartPolicyNever,
 						Containers: []api.Container{
 							{
-								Image:           "gcr.io/google_containers/busybox",
-								Name:            "busybox",
-								Command:         []string{"echo", "'Hello World'"},
-								ImagePullPolicy: api.PullIfNotPresent,
+								Image:   "gcr.io/google_containers/busybox",
+								Name:    "busybox",
+								Command: []string{"echo", "'Hello World'"},
 							},
 						},
 					},
@@ -84,4 +88,125 @@ var _ = Describe("Kubelet", func() {
 			})
 		})
 	})
+
+	Describe("metrics api", func() {
+		statsPrefix := "stats-busybox-"
+		podNames := []string{}
+		podCount := 2
+		for i := 0; i < podCount; i++ {
+			podNames = append(podNames, fmt.Sprintf("%s%v", statsPrefix, i))
+		}
+		BeforeEach(func() {
+			for _, podName := range podNames {
+				createPod(cl, podName, []api.Container{
+					{
+						Image:   "gcr.io/google_containers/busybox",
+						Command: []string{"sh", "-c", "echo 'Hello World' | tee ~/file | tee -a ~/file | tee /test-empty-dir | sleep 60"},
+						Name:    podName + containerSuffix,
+					},
+				})
+			}
+
+			// Sleep long enough for cadvisor to see the pod and calculate all of its metrics
+			time.Sleep(60 * time.Second)
+		})
+
+		Context("when querying /stats/summary", func() {
+			It("it should report resource usage through the stats api", func() {
+				resp, err := http.Get(*kubeletAddress + "/stats/summary")
+				now := time.Now()
+				Expect(err).To(BeNil(), fmt.Sprintf("Failed to get /stats/summary"))
+				summary := stats.Summary{}
+				contentsBytes, err := ioutil.ReadAll(resp.Body)
+				Expect(err).To(BeNil(), fmt.Sprintf("Failed to read /stats/summary: %+v", resp))
+				contents := string(contentsBytes)
+				decoder := json.NewDecoder(strings.NewReader(contents))
+				err = decoder.Decode(&summary)
+				Expect(err).To(BeNil(), fmt.Sprintf("Failed to parse /stats/summary to go struct: %+v", resp))
+
+				// Verify Misc Stats
+				Expect(summary.Time.Time).To(BeTemporally("~", now, 20*time.Second))
+
+				// Verify Node Stats are present
+				Expect(summary.Node.NodeName).To(Equal(*nodeName))
+				Expect(summary.Node.CPU.UsageCoreNanoSeconds).NotTo(BeZero())
+				Expect(summary.Node.Memory.UsageBytes).NotTo(BeZero())
+				Expect(summary.Node.Memory.WorkingSetBytes).NotTo(BeZero())
+				Expect(summary.Node.Fs.UsedBytes).NotTo(BeZero())
+				Expect(summary.Node.Fs.CapacityBytes).NotTo(BeZero())
+				Expect(summary.Node.Fs.AvailableBytes).NotTo(BeZero())
+
+				sysContainers := map[string]stats.ContainerStats{}
+				sysContainersList := []string{}
+				for _, container := range summary.Node.SystemContainers {
+					sysContainers[container.Name] = container
+					sysContainersList = append(sysContainersList, container.Name)
+					Expect(container.CPU.UsageCoreNanoSeconds).NotTo(BeZero())
+					// TODO: Test Network
+					Expect(container.Memory.UsageBytes).NotTo(BeZero())
+					Expect(container.Memory.WorkingSetBytes).NotTo(BeZero())
+					Expect(container.Rootfs.CapacityBytes).NotTo(BeZero())
+					Expect(container.Rootfs.AvailableBytes).NotTo(BeZero())
+					Expect(container.Logs.CapacityBytes).NotTo(BeZero())
+					Expect(container.Logs.AvailableBytes).NotTo(BeZero())
+				}
+				Expect(sysContainersList).To(ConsistOf("kubelet", "runtime"))
+
+				// Verify Pods Stats are present
+				podsList := []string{}
+				for _, pod := range summary.Pods {
+					if !strings.HasPrefix(pod.PodRef.Name, statsPrefix) {
+						// Ignore pods created outside this test
+						continue
+
+					}
+					// TODO: Test network
+
+					podsList = append(podsList, pod.PodRef.Name)
+					Expect(pod.Containers).To(HaveLen(1))
+					container := pod.Containers[0]
+					Expect(container.Name).To(Equal(pod.PodRef.Name + containerSuffix))
+					Expect(container.CPU.UsageCoreNanoSeconds).NotTo(BeZero())
+					Expect(container.Memory.UsageBytes).NotTo(BeZero())
+					Expect(container.Memory.WorkingSetBytes).NotTo(BeZero())
+					Expect(container.Rootfs.CapacityBytes).NotTo(BeZero())
+					Expect(container.Rootfs.AvailableBytes).NotTo(BeZero())
+					Expect(*container.Rootfs.UsedBytes).NotTo(BeZero(), contents)
+					Expect(container.Logs.CapacityBytes).NotTo(BeZero())
+					Expect(container.Logs.AvailableBytes).NotTo(BeZero())
+					Expect(*container.Logs.UsedBytes).NotTo(BeZero(), contents)
+				}
+				Expect(podsList).To(ConsistOf(podNames))
+			})
+		})
+
+		AfterEach(func() {
+			for _, podName := range podNames {
+				err := cl.Pods(api.NamespaceDefault).Delete(podName, &api.DeleteOptions{})
+				Expect(err).To(BeNil(), fmt.Sprintf("Error deleting Pod %v", podName))
+			}
+		})
+	})
 })
+
+const (
+	containerSuffix = "-c"
+)
+
+func createPod(cl *client.Client, podName string, containers []api.Container) {
+	pod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			Name:      podName,
+			Namespace: api.NamespaceDefault,
+		},
+		Spec: api.PodSpec{
+			// Force the Pod to schedule to the node without a scheduler running
+			NodeName: *nodeName,
+			// Don't restart the Pod since it is expected to exit
+			RestartPolicy: api.RestartPolicyNever,
+			Containers:    containers,
+		},
+	}
+	_, err := cl.Pods(api.NamespaceDefault).Create(pod)
+	Expect(err).To(BeNil(), fmt.Sprintf("Error creating Pod %v", err))
+}