Merge pull request kubernetes#89359 from gongguan/process

eviction by process number

Author: k8s-ci-robot

pkg/kubelet/apis/stats/v1alpha1/types.go

@@ -91,6 +91,13 @@ const (
 	SystemContainerPods = "pods"
 )
 
+// ProcessStats are stats pertaining to processes.
+type ProcessStats struct {
+	// Number of processes
+	// +optional
+	ProcessCount *uint64 `json:"process_count,omitempty"`
+}
+
 // PodStats holds pod-level unprocessed sample stats.
 type PodStats struct {
 	// Reference to the measured Pod.
@@ -119,6 +126,9 @@ type PodStats struct {
 	// EphemeralStorage reports the total filesystem usage for the containers and emptyDir-backed volumes in the measured Pod.
 	// +optional
 	EphemeralStorage *FsStats `json:"ephemeral-storage,omitempty"`
+	// ProcessStats pertaining to processes.
+	// +optional
+	ProcessStats *ProcessStats `json:"process_stats,omitempty"`
 }
 
 // ContainerStats holds container-level unprocessed sample stats.

pkg/kubelet/eviction/helpers.go

@@ -328,6 +328,15 @@ func memoryUsage(memStats *statsapi.MemoryStats) *resource.Quantity {
 	return resource.NewQuantity(usage, resource.BinarySI)
 }
 
+// processUsage converts working set into a process count.
+func processUsage(processStats *statsapi.ProcessStats) uint64 {
+	if processStats == nil || processStats.ProcessCount == nil {
+		return 0
+	}
+	usage := uint64(*processStats.ProcessCount)
+	return usage
+}
+
 // localVolumeNames returns the set of volumes for the pod that are local
 // TODO: summary API should report what volumes consume local storage rather than hard-code here.
 func localVolumeNames(pod *v1.Pod) []string {
@@ -566,6 +575,23 @@ func memory(stats statsFunc) cmpFunc {
 	}
 }
 
+// process compares pods by largest consumer of process number relative to request.
+func process(stats statsFunc) cmpFunc {
+	return func(p1, p2 *v1.Pod) int {
+		p1Stats, p1Found := stats(p1)
+		p2Stats, p2Found := stats(p2)
+		if !p1Found || !p2Found {
+			// prioritize evicting the pod for which no stats were found
+			return cmpBool(!p1Found, !p2Found)
+		}
+
+		p1Process := processUsage(p1Stats.ProcessStats)
+		p2Process := processUsage(p2Stats.ProcessStats)
+		// prioritize evicting the pod which has the larger consumption of process
+		return int(p2Process - p1Process)
+	}
+}
+
 // exceedDiskRequests compares whether or not pods' disk usage exceeds their requests
 func exceedDiskRequests(stats statsFunc, fsStatsToMeasure []fsStatsType, diskResource v1.ResourceName) cmpFunc {
 	return func(p1, p2 *v1.Pod) int {
@@ -640,7 +666,7 @@ func rankMemoryPressure(pods []*v1.Pod, stats statsFunc) {
 
 // rankPIDPressure orders the input pods by priority in response to PID pressure.
 func rankPIDPressure(pods []*v1.Pod, stats statsFunc) {
-	orderedBy(priority).Sort(pods)
+	orderedBy(priority, process(stats)).Sort(pods)
 }
 
 // rankDiskPressureFunc returns a rankFunc that measures the specified fs stats.

pkg/kubelet/eviction/helpers_test.go

@@ -958,6 +958,32 @@ func TestOrderedByPriorityMemory(t *testing.T) {
 	}
 }
 
+// TestOrderedByPriorityProcess ensures we order by priority and then process consumption relative to request.
+func TestOrderedByPriorityProcess(t *testing.T) {
+	pod1 := newPod("low-priority-high-usage", lowPriority, nil, nil)
+	pod2 := newPod("low-priority-low-usage", lowPriority, nil, nil)
+	pod3 := newPod("high-priority-high-usage", highPriority, nil, nil)
+	pod4 := newPod("high-priority-low-usage", highPriority, nil, nil)
+	stats := map[*v1.Pod]statsapi.PodStats{
+		pod1: newPodProcessStats(pod1, 20),
+		pod2: newPodProcessStats(pod2, 6),
+		pod3: newPodProcessStats(pod3, 20),
+		pod4: newPodProcessStats(pod4, 5),
+	}
+	statsFn := func(pod *v1.Pod) (statsapi.PodStats, bool) {
+		result, found := stats[pod]
+		return result, found
+	}
+	pods := []*v1.Pod{pod4, pod3, pod2, pod1}
+	expected := []*v1.Pod{pod1, pod2, pod3, pod4}
+	orderedBy(priority, process(statsFn)).Sort(pods)
+	for i := range expected {
+		if pods[i] != expected[i] {
+			t.Errorf("Expected pod[%d]: %s, but got: %s", i, expected[i].Name, pods[i].Name)
+		}
+	}
+}
+
 func TestSortByEvictionPriority(t *testing.T) {
 	for _, tc := range []struct {
 		name       string
@@ -1884,6 +1910,17 @@ func newPodMemoryStats(pod *v1.Pod, workingSet resource.Quantity) statsapi.PodSt
 	}
 }
 
+func newPodProcessStats(pod *v1.Pod, num uint64) statsapi.PodStats {
+	return statsapi.PodStats{
+		PodRef: statsapi.PodReference{
+			Name: pod.Name, Namespace: pod.Namespace, UID: string(pod.UID),
+		},
+		ProcessStats: &statsapi.ProcessStats{
+			ProcessCount: &num,
+		},
+	}
+}
+
 func newResourceList(cpu, memory, disk string) v1.ResourceList {
 	res := v1.ResourceList{}
 	if cpu != "" {

pkg/kubelet/stats/cadvisor_stats_provider.go

@@ -120,7 +120,7 @@ func (p *cadvisorStatsProvider) ListPodStats() ([]statsapi.PodStats, error) {
 		if containerName == leaky.PodInfraContainerName {
 			// Special case for infrastructure container which is hidden from
 			// the user and has network stats.
-			podStats.Network = cadvisorInfoToNetworkStats("pod:"+ref.Namespace+"_"+ref.Name, &cinfo)
+			podStats.Network = cadvisorInfoToNetworkStats(&cinfo)
 		} else {
 			podStats.Containers = append(podStats.Containers, *cadvisorInfoToContainerStats(containerName, &cinfo, &rootFsInfo, &imageFsInfo))
 		}
@@ -144,6 +144,7 @@ func (p *cadvisorStatsProvider) ListPodStats() ([]statsapi.PodStats, error) {
 			cpu, memory := cadvisorInfoToCPUandMemoryStats(podInfo)
 			podStats.CPU = cpu
 			podStats.Memory = memory
+			podStats.ProcessStats = cadvisorInfoToProcessStats(podInfo)
 		}
 
 		status, found := p.statusProvider.GetPodStatus(podUID)

pkg/kubelet/stats/cri_stats_provider.go

@@ -199,6 +199,7 @@ func (p *criStatsProvider) listPodStats(updateCPUNanoCoreUsage bool) ([]statsapi
 		cs := p.makeContainerStats(stats, container, &rootFsInfo, fsIDtoInfo, podSandbox.GetMetadata(), updateCPUNanoCoreUsage)
 		p.addPodNetworkStats(ps, podSandboxID, caInfos, cs, containerNetworkStats[podSandboxID])
 		p.addPodCPUMemoryStats(ps, types.UID(podSandbox.Metadata.Uid), allInfos, cs)
+		p.addProcessStats(ps, types.UID(podSandbox.Metadata.Uid), allInfos, cs)
 
 		// If cadvisor stats is available for the container, use it to populate
 		// container stats
@@ -425,7 +426,7 @@ func (p *criStatsProvider) addPodNetworkStats(
 	caPodSandbox, found := caInfos[podSandboxID]
 	// try get network stats from cadvisor first.
 	if found {
-		networkStats := cadvisorInfoToNetworkStats(ps.PodRef.Name, &caPodSandbox)
+		networkStats := cadvisorInfoToNetworkStats(&caPodSandbox)
 		if networkStats != nil {
 			ps.Network = networkStats
 			return
@@ -491,6 +492,20 @@ func (p *criStatsProvider) addPodCPUMemoryStats(
 	}
 }
 
+func (p *criStatsProvider) addProcessStats(
+	ps *statsapi.PodStats,
+	podUID types.UID,
+	allInfos map[string]cadvisorapiv2.ContainerInfo,
+	cs *statsapi.ContainerStats,
+) {
+	// try get process stats from cadvisor only.
+	info := getCadvisorPodInfoFromPodUID(podUID, allInfos)
+	if info != nil {
+		ps.ProcessStats = cadvisorInfoToProcessStats(info)
+		return
+	}
+}
+
 func (p *criStatsProvider) makeContainerStats(
 	stats *runtimeapi.ContainerStats,
 	container *runtimeapi.Container,

pkg/kubelet/stats/helper.go

@@ -153,9 +153,18 @@ func cadvisorInfoToContainerCPUAndMemoryStats(name string, info *cadvisorapiv2.C
 	return result
 }
 
+func cadvisorInfoToProcessStats(info *cadvisorapiv2.ContainerInfo) *statsapi.ProcessStats {
+	cstat, found := latestContainerStats(info)
+	if !found || cstat.Processes == nil {
+		return nil
+	}
+	num := cstat.Processes.ProcessCount
+	return &statsapi.ProcessStats{ProcessCount: uint64Ptr(num)}
+}
+
 // cadvisorInfoToNetworkStats returns the statsapi.NetworkStats converted from
 // the container info from cadvisor.
-func cadvisorInfoToNetworkStats(name string, info *cadvisorapiv2.ContainerInfo) *statsapi.NetworkStats {
+func cadvisorInfoToNetworkStats(info *cadvisorapiv2.ContainerInfo) *statsapi.NetworkStats {
 	if !info.Spec.HasNetwork {
 		return nil
 	}

pkg/kubelet/stats/stats_provider.go

@@ -114,7 +114,7 @@ func (p *StatsProvider) GetCgroupStats(cgroupName string, updateStats bool) (*st
 	}
 	// Rootfs and imagefs doesn't make sense for raw cgroup.
 	s := cadvisorInfoToContainerStats(cgroupName, info, nil, nil)
-	n := cadvisorInfoToNetworkStats(cgroupName, info)
+	n := cadvisorInfoToNetworkStats(info)
 	return s, n, nil
 }
 

test/e2e_node/eviction_test.go

@@ -422,15 +422,20 @@ var _ = framework.KubeDescribe("PriorityPidEvictionOrdering [Slow] [Serial] [Dis
 		})
 		specs := []podEvictSpec{
 			{
-				evictionPriority: 1,
-				pod:              pidConsumingPod("fork-bomb-container", 12000),
+				evictionPriority: 2,
+				pod:              pidConsumingPod("fork-bomb-container-with-low-priority", 12000),
 			},
 			{
 				evictionPriority: 0,
 				pod:              innocentPod(),
 			},
+			{
+				evictionPriority: 1,
+				pod:              pidConsumingPod("fork-bomb-container-with-high-priority", 12000),
+			},
 		}
 		specs[1].pod.Spec.PriorityClassName = highPriorityClassName
+		specs[2].pod.Spec.PriorityClassName = highPriorityClassName
 		runEvictionTest(f, pressureTimeout, expectedNodeCondition, expectedStarvedResource, logPidMetrics, specs)
 	})
 })

test/e2e_node/summary_test.go

@@ -260,6 +260,9 @@ var _ = framework.KubeDescribe("Summary API [NodeConformance]", func() {
 					"Inodes":         bounded(1e4, 1e8),
 					"InodesUsed":     bounded(0, 1e8),
 				}),
+				"ProcessStats": ptrMatchAllFields(gstruct.Fields{
+					"ProcessCount": bounded(0, 1e8),
+				}),
 			})
 
 			matchExpectations := ptrMatchAllFields(gstruct.Fields{