Filter out duplicated container stats

pkg/kubelet/stats/cadvisor_stats_provider.go

@@ -106,7 +106,7 @@ func (p *cadvisorStatsProvider) ListPodStats() ([]statsapi.PodStats, error) {
 		if !isPodManagedContainer(&cinfo) {
 			continue
 		}
-		ref := buildPodRef(&cinfo)
+		ref := buildPodRef(cinfo.Spec.Labels)
 
 		// Lookup the PodStats for the pod using the PodRef. If none exists,
 		// initialize a new entry.
@@ -172,10 +172,10 @@ func (p *cadvisorStatsProvider) ImageFsStats() (*statsapi.FsStats, error) {
 }
 
 // buildPodRef returns a PodReference that identifies the Pod managing cinfo
-func buildPodRef(cinfo *cadvisorapiv2.ContainerInfo) statsapi.PodReference {
-	podName := kubetypes.GetPodName(cinfo.Spec.Labels)
-	podNamespace := kubetypes.GetPodNamespace(cinfo.Spec.Labels)
-	podUID := kubetypes.GetPodUID(cinfo.Spec.Labels)
+func buildPodRef(containerLabels map[string]string) statsapi.PodReference {
+	podName := kubetypes.GetPodName(containerLabels)
+	podNamespace := kubetypes.GetPodNamespace(containerLabels)
+	podUID := kubetypes.GetPodUID(containerLabels)
 	return statsapi.PodReference{Name: podName, Namespace: podNamespace, UID: podUID}
 }
 
@@ -204,7 +204,7 @@ func removeTerminatedContainerInfo(containerInfo map[string]cadvisorapiv2.Contai
 			continue
 		}
 		cinfoID := containerID{
-			podRef:        buildPodRef(&cinfo),
+			podRef:        buildPodRef(cinfo.Spec.Labels),
 			containerName: kubetypes.GetContainerName(cinfo.Spec.Labels),
 		}
 		cinfoMap[cinfoID] = append(cinfoMap[cinfoID], containerInfoWithCgroup{

pkg/kubelet/stats/cri_stats_provider.go

@@ -18,6 +18,7 @@ package stats
 
 import (
 	"fmt"
+	"sort"
 	"time"
 
 	"github.com/golang/glog"
@@ -31,6 +32,7 @@ import (
 	statsapi "k8s.io/kubernetes/pkg/kubelet/apis/stats/v1alpha1"
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/server/stats"
+	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 )
 
 // criStatsProvider implements the containerStatsProvider interface by getting
@@ -74,15 +76,10 @@ func (p *criStatsProvider) ListPodStats() ([]statsapi.PodStats, error) {
 		return nil, fmt.Errorf("failed to get rootFs info: %v", err)
 	}
 
-	// Creates container map.
-	containerMap := make(map[string]*runtimeapi.Container)
 	containers, err := p.runtimeService.ListContainers(&runtimeapi.ContainerFilter{})
 	if err != nil {
 		return nil, fmt.Errorf("failed to list all containers: %v", err)
 	}
-	for _, c := range containers {
-		containerMap[c.Id] = c
-	}
 
 	// Creates pod sandbox map.
 	podSandboxMap := make(map[string]*runtimeapi.PodSandbox)
@@ -106,6 +103,14 @@ func (p *criStatsProvider) ListPodStats() ([]statsapi.PodStats, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to list all container stats: %v", err)
 	}
+
+	containers = removeTerminatedContainer(containers)
+	// Creates container map.
+	containerMap := make(map[string]*runtimeapi.Container)
+	for _, c := range containers {
+		containerMap[c.Id] = c
+	}
+
 	for _, stats := range resp {
 		containerID := stats.Attributes.Id
 		container, found := containerMap[containerID]
@@ -279,3 +284,39 @@ func (p *criStatsProvider) makeContainerStats(
 
 	return result
 }
+
+// removeTerminatedContainer returns the specified container but with
+// the stats of the terminated containers removed.
+func removeTerminatedContainer(containers []*runtimeapi.Container) []*runtimeapi.Container {
+	containerMap := make(map[containerID][]*runtimeapi.Container)
+	// Sort order by create time
+	sort.Slice(containers, func(i, j int) bool {
+		return containers[i].CreatedAt < containers[j].CreatedAt
+	})
+	for _, container := range containers {
+		refID := containerID{
+			podRef:        buildPodRef(container.Labels),
+			containerName: kubetypes.GetContainerName(container.Labels),
+		}
+		containerMap[refID] = append(containerMap[refID], container)
+	}
+
+	result := make([]*runtimeapi.Container, 0)
+	for _, refs := range containerMap {
+		if len(refs) == 1 {
+			result = append(result, refs[0])
+			continue
+		}
+		found := false
+		for i := 0; i < len(refs); i++ {
+			if refs[i].State == runtimeapi.ContainerState_CONTAINER_RUNNING {
+				found = true
+				result = append(result, refs[i])
+			}
+		}
+		if !found {
+			result = append(result, refs[len(refs)-1])
+		}
+	}
+	return result
+}