Fix and cleanup container metrics

Signed-off-by: Lantao Liu <lantaol@google.com>

pkg/server/container_stats.go

@@ -27,24 +27,22 @@ import (
 // ContainerStats returns stats of the container. If the container does not
 // exist, the call returns an error.
 func (c *criContainerdService) ContainerStats(ctx context.Context, in *runtime.ContainerStatsRequest) (*runtime.ContainerStatsResponse, error) {
-	// Validate the stats request
+	cntr, err := c.containerStore.Get(in.GetContainerId())
-	if in.GetContainerId() == "" {
-		return nil, fmt.Errorf("invalid container stats request")
-	}
-	containerID := in.GetContainerId()
-	_, err := c.containerStore.Get(containerID)
 	if err != nil {
-		return nil, fmt.Errorf("failed to find container %q: %v", containerID, err)
+		return nil, fmt.Errorf("failed to find container: %v", err)
 	}
-	request := &tasks.MetricsRequest{Filters: []string{"id==" + containerID}}
+	request := &tasks.MetricsRequest{Filters: []string{"id==" + cntr.ID}}
 	resp, err := c.taskService.Metrics(ctx, request)
 	if err != nil {
-		return nil, fmt.Errorf("failed to fetch metrics for tasks: %v", err)
+		return nil, fmt.Errorf("failed to fetch metrics for task: %v", err)
+	}
+	if len(resp.Metrics) != 1 {
+		return nil, fmt.Errorf("unexpected metrics response: %+v", resp.Metrics)
 	}
 
-	var cs runtime.ContainerStats
+	cs, err := c.getContainerMetrics(cntr.Metadata, resp.Metrics[0])
-	if err := c.getContainerMetrics(containerID, resp.Metrics[0], &cs); err != nil {
+	if err != nil {
 		return nil, fmt.Errorf("failed to decode container metrics: %v", err)
 	}
-	return &runtime.ContainerStatsResponse{Stats: &cs}, nil
+	return &runtime.ContainerStatsResponse{Stats: cs}, nil
 }

pkg/server/container_stats_list.go

@@ -23,9 +23,10 @@ import (
 	tasks "github.com/containerd/containerd/api/services/tasks/v1"
 	"github.com/containerd/containerd/api/types"
 	"github.com/containerd/typeurl"
-	"github.com/golang/glog"
 	"golang.org/x/net/context"
 	"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
+
+	containerstore "github.com/kubernetes-incubator/cri-containerd/pkg/store/container"
 )
 
 // ListContainerStats returns stats of all running containers.
@@ -33,7 +34,7 @@ func (c *criContainerdService) ListContainerStats(
 	ctx context.Context,
 	in *runtime.ListContainerStatsRequest,
 ) (*runtime.ListContainerStatsResponse, error) {
-	request, candidateContainers, err := c.buildTaskMetricsRequest(in)
+	request, containers, err := c.buildTaskMetricsRequest(in)
 	if err != nil {
 		return nil, fmt.Errorf("failed to build metrics request: %v", err)
 	}
@@ -41,7 +42,7 @@ func (c *criContainerdService) ListContainerStats(
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch metrics for tasks: %v", err)
 	}
-	criStats, err := c.toCRIContainerStats(resp.Metrics, candidateContainers)
+	criStats, err := c.toCRIContainerStats(resp.Metrics, containers)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert to cri containerd stats format: %v", err)
 	}
@@ -50,39 +51,30 @@ func (c *criContainerdService) ListContainerStats(
 
 func (c *criContainerdService) toCRIContainerStats(
 	stats []*types.Metric,
-	candidateContainers map[string]bool,
+	containers []containerstore.Container,
 ) (*runtime.ListContainerStatsResponse, error) {
-	containerStats := new(runtime.ListContainerStatsResponse)
+	statsMap := make(map[string]*types.Metric)
 	for _, stat := range stats {
-		var cs runtime.ContainerStats
+		statsMap[stat.ID] = stat
-		if err := c.getContainerMetrics(stat.ID, stat, &cs); err != nil {
-			glog.Errorf("failed to decode container metrics: %v", err)
-			continue
 	}
-		delete(candidateContainers, stat.ID)
+	containerStats := new(runtime.ListContainerStatsResponse)
-		containerStats.Stats = append(containerStats.Stats, &cs)
+	for _, cntr := range containers {
+		cs, err := c.getContainerMetrics(cntr.Metadata, statsMap[cntr.ID])
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode container metrics for %q: %v", cntr.ID, err)
 		}
-	// If there is a state where containers are dead at the time of query
+		containerStats.Stats = append(containerStats.Stats, cs)
-	// but not removed, then check if the writeableLayer information
-	// is present and attach the same.
-	for id := range candidateContainers {
-		var cs runtime.ContainerStats
-		if err := c.getContainerMetrics(id, nil, &cs); err != nil {
-			glog.Errorf("failed to decode container metrics: %v", err)
-			continue
-		}
-		containerStats.Stats = append(containerStats.Stats, &cs)
 	}
 	return containerStats, nil
 }
 
 func (c *criContainerdService) getContainerMetrics(
-	containerID string,
+	meta containerstore.Metadata,
 	stats *types.Metric,
-	cs *runtime.ContainerStats,
+) (*runtime.ContainerStats, error) {
-) error {
+	var cs runtime.ContainerStats
 	var usedBytes, inodesUsed uint64
-	sn, err := c.snapshotStore.Get(containerID)
+	sn, err := c.snapshotStore.Get(meta.ID)
 	// If snapshotstore doesn't have cached snapshot information
 	// set WritableLayer usage to zero
 	if err == nil {
@@ -97,23 +89,17 @@ func (c *criContainerdService) getContainerMetrics(
 		UsedBytes:  &runtime.UInt64Value{usedBytes},
 		InodesUsed: &runtime.UInt64Value{inodesUsed},
 	}
-
-	// Get the container from store and extract the attributes
-	cnt, err := c.containerStore.Get(containerID)
-	if err != nil {
-		return fmt.Errorf("failed to find container %q in container store: %v", containerID, err)
-	}
 	cs.Attributes = &runtime.ContainerAttributes{
-		Id:          containerID,
+		Id:          meta.ID,
-		Metadata:    cnt.Config.GetMetadata(),
+		Metadata:    meta.Config.GetMetadata(),
-		Labels:      cnt.Config.GetLabels(),
+		Labels:      meta.Config.GetLabels(),
-		Annotations: cnt.Config.GetAnnotations(),
+		Annotations: meta.Config.GetAnnotations(),
 	}
 
 	if stats != nil {
 		s, err := typeurl.UnmarshalAny(stats.Data)
 		if err != nil {
-			return fmt.Errorf("failed to extract container metrics: %v", err)
+			return nil, fmt.Errorf("failed to extract container metrics: %v", err)
 		}
 		metrics := s.(*cgroups.Metrics)
 		cs.Cpu = &runtime.CpuUsage{
@@ -126,36 +112,34 @@ func (c *criContainerdService) getContainerMetrics(
 		}
 	}
 
-	return nil
+	return &cs, nil
 }
 
 // buildTaskMetricsRequest constructs a tasks.MetricsRequest based on
 // the information in the stats request and the containerStore
 func (c *criContainerdService) buildTaskMetricsRequest(
 	r *runtime.ListContainerStatsRequest,
-) (tasks.MetricsRequest, map[string]bool, error) {
+) (tasks.MetricsRequest, []containerstore.Container, error) {
 	var req tasks.MetricsRequest
-	if r.GetFilter == nil {
+	if r.GetFilter() == nil {
 		return req, nil, nil
 	}
-
+	var containers []containerstore.Container
-	candidateContainers := make(map[string]bool)
+	for _, cntr := range c.containerStore.List() {
-	for _, c := range c.containerStore.List() {
+		if r.GetFilter().GetId() != "" && cntr.ID != r.GetFilter().GetId() {
-		if r.Filter.GetId() != "" && c.ID != r.Filter.GetId() {
 			continue
 		}
-		if r.Filter.GetPodSandboxId() != "" && c.SandboxID != r.Filter.GetPodSandboxId() {
+		if r.GetFilter().GetPodSandboxId() != "" && cntr.SandboxID != r.GetFilter().GetPodSandboxId() {
 			continue
 		}
-		if r.Filter.GetLabelSelector() != nil && !matchLabelSelector(r.Filter.GetLabelSelector(), c.Config.GetLabels()) {
+		if r.GetFilter().GetLabelSelector() != nil &&
+			!matchLabelSelector(r.GetFilter().GetLabelSelector(), cntr.Config.GetLabels()) {
 			continue
 		}
-		candidateContainers[c.ID] = true
+		containers = append(containers, cntr)
+		req.Filters = append(req.Filters, "id=="+cntr.ID)
 	}
-	for id := range candidateContainers {
+	return req, containers, nil
-		req.Filters = append(req.Filters, "id=="+id)
-	}
-	return req, candidateContainers, nil
 }
 
 func matchLabelSelector(selector, labels map[string]string) bool {