feat: support to isolate pods by cpu utilization

align cpuset with L3 cache

Signed-off-by: linzhecheng <linzhecheng@bytedance.com>

Author: cheney-lin

cmd/katalyst-agent/app/options/sysadvisor/qosaware/resource/cpu/cpu_isolation.go

@@ -19,6 +19,7 @@ package cpu
 import (
 	"fmt"
 	"strconv"
+	"time"
 
 	"github.com/spf13/pflag"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -59,6 +60,11 @@ type CPUIsolationOptions struct {
 	IsolationNonExclusivePools []string
 
 	IsolationIncludeSidecarRequirement bool
+	UtilWatermarkSupreme               float64
+	UtilWatermarkHigh                  float64
+	UtilWatermarkLow                   float64
+	MetricSyncPeriod                   time.Duration
+	MetricSlidingWindowTime            time.Duration
 }
 
 // NewCPUIsolationOptions creates a new Options with a default config
@@ -80,6 +86,12 @@ func NewCPUIsolationOptions() *CPUIsolationOptions {
 		IsolationDisabledPools:     []string{},
 		IsolationForceEnablePools:  []string{},
 		IsolationNonExclusivePools: []string{},
+
+		UtilWatermarkSupreme:    0.75,
+		UtilWatermarkHigh:       0.55,
+		UtilWatermarkLow:        0.40,
+		MetricSyncPeriod:        time.Second * 20,
+		MetricSlidingWindowTime: 2 * time.Minute,
 	}
 }
 
@@ -119,6 +131,11 @@ func (o *CPUIsolationOptions) AddFlags(fs *pflag.FlagSet) {
 		"isolation is non-exclusive for get given pool")
 	fs.BoolVar(&o.IsolationIncludeSidecarRequirement, "isolation-include-sidecar-requirement", o.IsolationIncludeSidecarRequirement,
 		"isolation include sidecar requirement")
+
+	fs.Float64Var(&o.UtilWatermarkLow, "isolation-util-watermark-low", o.UtilWatermarkLow, "cpu utilization watermark low for isolation")
+	fs.Float64Var(&o.UtilWatermarkHigh, "isolation-util-watermark-high", o.UtilWatermarkHigh, "cpu utilization watermark high for isolation")
+	fs.DurationVar(&o.MetricSlidingWindowTime, "isolation-metric-sliding-window-time", o.MetricSlidingWindowTime, "metric sliding window time for isolation")
+	fs.DurationVar(&o.MetricSyncPeriod, "isolation-metric-sync-period", o.MetricSyncPeriod, "metric sync period for isolation")
 }
 
 // ApplyTo fills up config with options
@@ -163,6 +180,11 @@ func (o *CPUIsolationOptions) ApplyTo(c *cpu.CPUIsolationConfiguration) error {
 	c.IsolationNonExclusivePools = sets.NewString(o.IsolationNonExclusivePools...)
 
 	c.IsolationIncludeSidecarRequirement = o.IsolationIncludeSidecarRequirement
+	c.UtilWatermarkSupreme = o.UtilWatermarkSupreme
+	c.UtilWatermarkHigh = o.UtilWatermarkHigh
+	c.UtilWatermarkLow = o.UtilWatermarkLow
+	c.MetricSlidingWindowTime = o.MetricSlidingWindowTime
+	c.MetricSyncPeriod = o.MetricSyncPeriod
 
 	return nil
 }

pkg/agent/qrm-plugins/cpu/dynamicpolicy/cpuadvisor/helper.go

@@ -95,6 +95,21 @@ type BlockEntry struct {
 	SubEntryName string
 }
 
+func GetBlocksByPoolName(blockInfos []*BlockInfo, poolNamePrefix string) []*BlockInfo {
+	result := make([]*BlockInfo, 0)
+	for _, blockInfo := range blockInfos {
+		if blockInfo == nil {
+			continue
+		}
+		for ownerPoolName := range blockInfo.OwnerPoolEntryMap {
+			if strings.HasPrefix(ownerPoolName, poolNamePrefix) {
+				result = append(result, blockInfo)
+			}
+		}
+	}
+	return result
+}
+
 // GetBlocks parses ListAndWatchResponse and returns map[int][]*BlockInfo,
 // the map is keyed as numa id -> blocks slice (which has been sorted and deduped)
 func (lwr *ListAndWatchResponse) GetBlocks() (map[int][]*BlockInfo, error) {

pkg/agent/qrm-plugins/cpu/dynamicpolicy/policy_advisor_handler.go

@@ -881,12 +881,8 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 
 		// First handle blocks for NUMA-bound dedicated_cores containers
 		// Reuse already allocated CPU sets when possible to minimize CPU migration
-		for _, block := range blocks {
-			if block == nil {
-				general.Warningf("got nil block")
-				continue
-			}
-
+		dedicatedBlocks := advisorapi.GetBlocksByPoolName(blocks, commonstate.PoolNameDedicated)
+		for _, block := range dedicatedBlocks {
 			entry, ok := block.OwnerPoolEntryMap[commonstate.PoolNameDedicated]
 			if !ok {
 				continue
@@ -917,6 +913,8 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 
 			var cpuset machine.CPUSet
 			alreadyAllocatedCPUs = alreadyAllocatedCPUs.Intersection(numaAvailableCPUs)
+
+			// shrink cpuSet
 			if alreadyAllocatedCPUs.Size() >= blockResult {
 				cpuset, err = calculator.TakeByTopology(machineInfo, alreadyAllocatedCPUs, blockResult, true)
 				if err != nil {
@@ -940,54 +938,65 @@ func (p *DynamicPolicy) generateBlockCPUSet(resp *advisorapi.ListAndWatchRespons
 
 		// Then handle blocks for NUMA-bound shared_cores containers and reclaimed_cores containers colocated with them
 		// These containers can share NUMA nodes with dedicated_cores containers
-		for _, block := range blocks {
-			if block == nil {
-				general.Warningf("got nil block")
-				continue
-			}
-
-			_, ok := block.OwnerPoolEntryMap[commonstate.PoolNameDedicated]
-			if ok {
-				continue
-			}
+		f := func(blocks []*advisorapi.BlockInfo, alignedByL3 bool) error {
+			for _, block := range blocks {
+				if block == nil {
+					general.Warningf("got nil block")
+					continue
+				}
 
-			blockID := block.BlockId
-			if _, found := blockCPUSet[blockID]; found {
-				general.Warningf("block: %v already allocated", blockID)
-				continue
-			}
+				blockID := block.BlockId
+				if _, found := blockCPUSet[blockID]; found {
+					general.Warningf("block: %v already allocated", blockID)
+					continue
+				}
 
-			blockResult, err := general.CovertUInt64ToInt(block.Result)
-			if err != nil {
-				return nil, fmt.Errorf("parse block: %s result failed with error: %v",
-					blockID, err)
-			}
+				blockResult, err := general.CovertUInt64ToInt(block.Result)
+				if err != nil {
+					return fmt.Errorf("parse block: %s result failed with error: %v",
+						blockID, err)
+				}
 
-			cpuset, err := calculator.TakeByTopology(machineInfo, numaAvailableCPUs, blockResult, false)
-			if err != nil {
-				return nil, fmt.Errorf("allocate cpuset for NUMA Aware block: %s in NUMA: %d failed with error: %v, numaAvailableCPUs: %d(%s), blockResult: %d",
-					blockID, numaID, err, numaAvailableCPUs.Size(), numaAvailableCPUs.String(), blockResult)
-			}
+				cpuset, err := calculator.TakeByTopology(machineInfo, numaAvailableCPUs, blockResult, alignedByL3)
+				if err != nil {
+					return fmt.Errorf("allocate cpuset for NUMA Aware block: %s in NUMA: %d failed with error: %v, numaAvailableCPUs: %d(%s), blockResult: %d",
+						blockID, numaID, err, numaAvailableCPUs.Size(), numaAvailableCPUs.String(), blockResult)
+				}
 
-			blockCPUSet[blockID] = cpuset
-			numaAvailableCPUs = numaAvailableCPUs.Difference(cpuset)
-			availableCPUs = availableCPUs.Difference(cpuset)
+				blockCPUSet[blockID] = cpuset
+				numaAvailableCPUs = numaAvailableCPUs.Difference(cpuset)
+				availableCPUs = availableCPUs.Difference(cpuset)
 
-			for poolName := range block.OwnerPoolEntryMap {
-				if commonstate.IsIsolationPool(poolName) || commonstate.IsShareNUMABindingPool(poolName) {
-					withNUMABindingShareOrDedicatedPod = true
-					break
+				for poolName := range block.OwnerPoolEntryMap {
+					if commonstate.IsIsolationPool(poolName) || commonstate.IsShareNUMABindingPool(poolName) {
+						withNUMABindingShareOrDedicatedPod = true
+						break
+					}
 				}
 			}
+
+			// Finally, if there are NUMA-bound containers on this NUMA node,
+			// deduct all numaAvailableCPUs from availableCPUs to ensure that
+			// NUMA-bound pods don't share the same NUMA node with non-NUMA-bound pods
+			if withNUMABindingShareOrDedicatedPod {
+				// Because numaAvailableCPUs is a subset of availableCPUs,
+				// we need to deduct all numaAvailableCPUs from availableCPUs
+				availableCPUs = availableCPUs.Difference(numaAvailableCPUs)
+			}
+			return nil
 		}
 
-		// Finally, if there are NUMA-bound containers on this NUMA node,
-		// deduct all numaAvailableCPUs from availableCPUs to ensure that
-		// NUMA-bound pods don't share the same NUMA node with non-NUMA-bound pods
-		if withNUMABindingShareOrDedicatedPod {
-			// Because numaAvailableCPUs is a subset of availableCPUs,
-			// we need to deduct all numaAvailableCPUs from availableCPUs
-			availableCPUs = availableCPUs.Difference(numaAvailableCPUs)
+		isolationBlocks := advisorapi.GetBlocksByPoolName(blocks, commonstate.PoolNamePrefixIsolation)
+		if err := f(isolationBlocks, true); err != nil {
+			return nil, err
+		}
+		shareBlocks := advisorapi.GetBlocksByPoolName(blocks, commonstate.PoolNameShare)
+		if err := f(shareBlocks, false); err != nil {
+			return nil, err
+		}
+		reclaimBlocks := advisorapi.GetBlocksByPoolName(blocks, commonstate.PoolNameReclaim)
+		if err := f(reclaimBlocks, false); err != nil {
+			return nil, err
 		}
 	}
 

pkg/agent/sysadvisor/plugin/qosaware/reporter/nodemetric_reporter.go

@@ -250,6 +250,7 @@ func (p *nodeMetricsReporterPlugin) ListAndWatchReportContent(_ *v1alpha1.Empty,
 
 func (p *nodeMetricsReporterPlugin) updateNodeMetrics() {
 	general.InfoS("try to update node metrics")
+	p.metricAggregators.gc()
 
 	var errList []error
 	nodeMetricInfo, err := p.getNodeMetricInfo()

pkg/agent/sysadvisor/plugin/qosaware/resource/cpu/advisor.go

@@ -105,7 +105,7 @@ type cpuResourceAdvisor struct {
 	provisionAssembler provisionassembler.ProvisionAssembler
 	headroomAssembler  headroomassembler.HeadroomAssembler
 
-	isolator        isolation.Isolator
+	isolators       []isolation.Isolator
 	isolationSafety bool
 
 	mutex      sync.RWMutex
@@ -131,7 +131,10 @@ func NewCPUResourceAdvisor(conf *config.Configuration, extraConf interface{}, me
 		numRegionsPerNuma:  make(map[int]int),
 		nonBindingNumas:    machine.NewCPUSet(),
 
-		isolator: isolation.NewLoadIsolator(conf, extraConf, emitter, metaCache, metaServer),
+		isolators: []isolation.Isolator{
+			isolation.NewLoadIsolator(conf, extraConf, emitter, metaCache, metaServer),
+			isolation.NewPerfIsolator(conf, extraConf, emitter, metaCache, metaServer),
+		},
 
 		metaCache:  metaCache,
 		metaServer: metaServer,
@@ -151,6 +154,11 @@ func NewCPUResourceAdvisor(conf *config.Configuration, extraConf interface{}, me
 }
 
 func (cra *cpuResourceAdvisor) Run(ctx context.Context) {
+	for _, isolator := range cra.isolators {
+		if err := isolator.Start(ctx); err != nil {
+			klog.Fatalf("[qosaware-cpu] start isolator failed: %v", err)
+		}
+	}
 	<-ctx.Done()
 }
 
@@ -234,7 +242,7 @@ func (cra *cpuResourceAdvisor) updateWithIsolationGuardian(tryIsolation bool) (
 	}
 
 	cra.updateNumasAvailableResource()
-	isolationExists := cra.setIsolatedContainers(tryIsolation)
+	cra.setIsolatedContainers(tryIsolation)
 
 	// assign containers to regions
 	if err := cra.assignContainersToRegions(); err != nil {
@@ -244,7 +252,7 @@ func (cra *cpuResourceAdvisor) updateWithIsolationGuardian(tryIsolation bool) (
 
 	cra.gcRegionMap()
 	cra.updateAdvisorEssentials()
-	if tryIsolation && isolationExists && !cra.checkIsolationSafety() {
+	if tryIsolation && cra.hasIsolatedPods() && !cra.checkIsolationSafety() {
 		klog.Errorf("[qosaware-cpu] failed to check isolation")
 		return nil, errIsolationSafetyCheckFailed
 	}
@@ -284,57 +292,81 @@ func (cra *cpuResourceAdvisor) updateWithIsolationGuardian(tryIsolation bool) (
 	return &calculationResult, nil
 }
 
-// setIsolatedContainers get isolation status from isolator and update into containers
-func (cra *cpuResourceAdvisor) setIsolatedContainers(enableIsolated bool) bool {
+func (cra *cpuResourceAdvisor) getTargetIsolatedPods() (sets.String, error) {
 	isolatedPods := sets.NewString()
-	if enableIsolated {
-		isolatedPods = sets.NewString(cra.isolator.GetIsolatedPods()...)
-	}
-	if len(isolatedPods) > 0 {
-		klog.Infof("[qosaware-cpu] current isolated pod: %v", isolatedPods.List())
+	general.InfoS("try to getTargetIsolatedPods", "isolators", general.ToString(cra.isolators))
+	for _, isolator := range cra.isolators {
+		pods, err := isolator.GetIsolatedPods()
+		if err != nil {
+			return nil, err
+		}
+		isolatedPods.Insert(pods...)
 	}
+	return isolatedPods, nil
+}
 
-	_ = cra.metaCache.RangeAndUpdateContainer(func(podUID string, _ string, ci *types.ContainerInfo) bool {
-		ci.Isolated = false
-		if isolatedPods.Has(podUID) {
-			ci.Isolated = true
+func (cra *cpuResourceAdvisor) hasIsolatedPods() bool {
+	ret := false
+	cra.metaCache.RangeContainer(func(podUID string, containerName string, ci *types.ContainerInfo) bool {
+		if ci.Isolated {
+			ret = true
+			return false
 		}
 		return true
 	})
-	return len(isolatedPods) > 0
+	return ret
 }
 
-// checkIsolationSafety returns true iff the isolated-limit-sum and share-pool-size exceed total capacity
-// todo: this logic contains a lot of assumptions and should be refined in the future
-func (cra *cpuResourceAdvisor) checkIsolationSafety() bool {
-	shareAndIsolationPoolSize := 0
-	nonBindingNumas := cra.metaServer.CPUDetails.NUMANodes()
-	for _, r := range cra.regionMap {
-		if r.Type() == configapi.QoSRegionTypeShare {
-			controlKnob, err := r.GetProvision()
-			if err != nil {
-				klog.Errorf("[qosaware-cpu] get controlKnob for %v err: %v", r.Name(), err)
-				return false
-			}
-			shareAndIsolationPoolSize += int(controlKnob[configapi.ControlKnobNonReclaimedCPURequirement].Value)
-		} else if r.Type() == configapi.QoSRegionTypeIsolation {
-			pods := r.GetPods()
-			cra.metaCache.RangeContainer(func(podUID string, _ string, containerInfo *types.ContainerInfo) bool {
-				if _, ok := pods[podUID]; ok {
-					shareAndIsolationPoolSize += int(containerInfo.CPULimit)
-				}
-				return true
-			})
-		} else if r.Type() == configapi.QoSRegionTypeDedicatedNumaExclusive {
-			nonBindingNumas = nonBindingNumas.Difference(r.GetBindingNumas())
+// setIsolatedContainers get isolation status from isolator and update into containers
+func (cra *cpuResourceAdvisor) setIsolatedContainers(enableIsolation bool) {
+	isolatedPods := sets.NewString()
+	if enableIsolation {
+		pods, err := cra.getTargetIsolatedPods()
+		if err != nil {
+			klog.Errorf("[qosaware-cpu] get isolated pods failed: %v", err)
+			return
 		}
+		isolatedPods = pods
 	}
 
-	nonBindingSize := cra.metaServer.NUMAToCPUs.CPUSizeInNUMAs(cra.nonBindingNumas.ToSliceNoSortInt()...)
-	klog.Infof("[qosaware-cpu] shareAndIsolationPoolSize %v, nonBindingSize %v", shareAndIsolationPoolSize, nonBindingSize)
-	if shareAndIsolationPoolSize > nonBindingSize {
-		return false
-	}
+	klog.InfoS("[qosaware-cpu]", "current isolated pod", isolatedPods.List(), "enableIsolation", enableIsolation)
+	_ = cra.metaCache.RangeAndUpdateContainer(func(podUID string, containerName string, ci *types.ContainerInfo) bool {
+		ci.Isolated = isolatedPods.Has(podUID)
+		return true
+	})
+}
+
+// checkIsolationSafety returns true iff the isolated-limit-sum and share-pool-size exceed total capacity
+// todo: this logic contains a lot of assumptions and should be refined in the future
+func (cra *cpuResourceAdvisor) checkIsolationSafety() bool {
+	//shareAndIsolationPoolSize := 0
+	//nonBindingNumas := cra.metaServer.CPUDetails.NUMANodes()
+	//for _, r := range cra.regionMap {
+	//	if r.Type() == configapi.QoSRegionTypeShare {
+	//		controlKnob, err := r.GetProvision()
+	//		if err != nil {
+	//			klog.Errorf("[qosaware-cpu] get controlKnob for %v err: %v", r.Name(), err)
+	//			return false
+	//		}
+	//		shareAndIsolationPoolSize += int(controlKnob[configapi.ControlKnobNonReclaimedCPURequirement].Value)
+	//	} else if r.Type() == configapi.QoSRegionTypeIsolation {
+	//		pods := r.GetPods()
+	//		cra.metaCache.RangeContainer(func(podUID string, _ string, containerInfo *types.ContainerInfo) bool {
+	//			if _, ok := pods[podUID]; ok {
+	//				shareAndIsolationPoolSize += int(containerInfo.CPULimit)
+	//			}
+	//			return true
+	//		})
+	//	} else if r.Type() == configapi.QoSRegionTypeDedicatedNumaExclusive {
+	//		nonBindingNumas = nonBindingNumas.Difference(r.GetBindingNumas())
+	//	}
+	//}
+	//
+	//nonBindingSize := cra.metaServer.NUMAToCPUs.CPUSizeInNUMAs(cra.nonBindingNumas.ToSliceNoSortInt()...)
+	//klog.Infof("[qosaware-cpu] shareAndIsolationPoolSize %v, nonBindingSize %v", shareAndIsolationPoolSize, nonBindingSize)
+	//if shareAndIsolationPoolSize > nonBindingSize {
+	//	return false
+	//}
 	return true
 }