Merge pull request #776 from yehlemias/dev/optimize-qrm-remove-pod

optimize qrm remove pod

Author: xu282934741

pkg/agent/qrm-plugins/cpu/dynamicpolicy/calculator/cpu_assignment.go

@@ -42,12 +42,13 @@ type cpuAccumulator struct {
 }
 
 func newCPUAccumulator(machineInfo *machine.KatalystMachineInfo, availableCPUs machine.CPUSet, numCPUs int) *cpuAccumulator {
-	return &cpuAccumulator{
+	a := &cpuAccumulator{
 		numCPUsNeeded: numCPUs,
 		cpuTopology:   machineInfo.CPUTopology,
 		cpuDetails:    machineInfo.CPUDetails.KeepOnly(availableCPUs),
 		result:        machine.NewCPUSet(),
 	}
+	return a
 }
 
 func (a *cpuAccumulator) getDetails() machine.CPUDetails {
@@ -268,30 +269,22 @@ func TakeByNUMABalance(info *machine.KatalystMachineInfo, availableCPUs machine.
 ) (machine.CPUSet, machine.CPUSet, error) {
 	var err error
 	acc := newCPUAccumulator(info, availableCPUs, cpuRequirement)
+
 	if acc.isSatisfied() {
 		goto successful
 	}
 
+	if takeFreeCoresByNumaBalance(acc) {
+		goto successful
+	}
+
 	for {
 		if acc.isFailed() {
 			err = fmt.Errorf("not enough cpus available to satisfy request")
 			goto failed
 		}
 
-	numaLoop:
 		for _, s := range info.CPUDetails.NUMANodes().ToSliceInt() {
-			if acc.needs(acc.getTopology().CPUsPerCore()) && len(acc.freeCores()) > 0 {
-				for _, c := range acc.freeCoresInNUMANode(s) {
-					acc.take(acc.getDetails().CPUsInCores(c))
-					if acc.isSatisfied() {
-						goto successful
-					} else {
-						continue numaLoop
-					}
-				}
-				continue
-			}
-
 			for _, c := range acc.freeCPUsInNUMANode(s) {
 				if acc.needs(1) {
 					acc.take(machine.NewCPUSet(c))
@@ -359,30 +352,22 @@ func TakeByNUMABalanceReversely(info *machine.KatalystMachineInfo, availableCPUs
 ) (machine.CPUSet, machine.CPUSet, error) {
 	var err error
 	acc := newCPUAccumulator(info, availableCPUs, cpuRequirement)
+
 	if acc.isSatisfied() {
 		goto successful
 	}
 
+	if takeFreeCoresByNumaBalanceReversely(acc) {
+		goto successful
+	}
+
 	for {
 		if acc.isFailed() {
 			err = fmt.Errorf("not enough cpus available to satisfy request")
 			goto failed
 		}
 
-	numaLoop:
 		for _, s := range info.CPUDetails.NUMANodes().ToSliceInt() {
-			if acc.needs(acc.getTopology().CPUsPerCore()) && len(acc.freeCores()) > 0 {
-				for _, c := range acc.freeCoresInNUMANodeReversely(s) {
-					acc.take(acc.getDetails().CPUsInCores(c))
-					if acc.isSatisfied() {
-						goto successful
-					} else {
-						continue numaLoop
-					}
-				}
-				continue
-			}
-
 			for _, c := range acc.freeCPUsInNUMANodeReversely(s) {
 				if acc.needs(1) {
 					acc.take(machine.NewCPUSet(c))
@@ -403,3 +388,68 @@ failed:
 successful:
 	return acc.result.Clone(), availableCPUs.Difference(acc.result), nil
 }
+
+func takeFreeCoresByNumaBalance(acc *cpuAccumulator) bool {
+	info := acc.getTopology()
+	if !acc.needs(info.CPUsPerCore()) {
+		return false
+	}
+
+	numaSlice := info.CPUDetails.NUMANodes().ToSliceInt()
+	freeCoresMap := make(map[int][]int)
+	maxFreeCoresCountInNuma := 0
+	for _, s := range numaSlice {
+		freeCores := acc.freeCoresInNUMANode(s)
+		freeCoresMap[s] = freeCores
+		if len(freeCores) > maxFreeCoresCountInNuma {
+			maxFreeCoresCountInNuma = len(freeCores)
+		}
+	}
+
+	for i := 0; i < maxFreeCoresCountInNuma; i++ {
+		for _, s := range numaSlice {
+			if len(freeCoresMap[s]) > i {
+				c := freeCoresMap[s][i]
+				if acc.needs(info.CPUsPerCore()) {
+					acc.take(acc.getDetails().CPUsInCores(c))
+					if acc.isSatisfied() {
+						return true
+					}
+				}
+			}
+		}
+	}
+	return false
+}
+
+func takeFreeCoresByNumaBalanceReversely(acc *cpuAccumulator) bool {
+	info := acc.getTopology()
+	if !acc.needs(info.CPUsPerCore()) {
+		return false
+	}
+
+	numaSlice := info.CPUDetails.NUMANodes().ToSliceInt()
+	reverselyFreeCoresMap := make(map[int][]int)
+	maxFreeCoresCountInNuma := 0
+	for _, s := range numaSlice {
+		freeCores := acc.freeCoresInNUMANodeReversely(s)
+		reverselyFreeCoresMap[s] = freeCores
+		if len(freeCores) > maxFreeCoresCountInNuma {
+			maxFreeCoresCountInNuma = len(freeCores)
+		}
+	}
+	for i := 0; i < maxFreeCoresCountInNuma; i++ {
+		for _, s := range numaSlice {
+			if len(reverselyFreeCoresMap[s]) > i {
+				c := reverselyFreeCoresMap[s][i]
+				if acc.needs(info.CPUsPerCore()) {
+					acc.take(acc.getDetails().CPUsInCores(c))
+					if acc.isSatisfied() {
+						return true
+					}
+				}
+			}
+		}
+	}
+	return false
+}

pkg/agent/qrm-plugins/cpu/dynamicpolicy/cpueviction/strategy/pressure_load_test.go

@@ -386,7 +386,7 @@ func TestThresholdMet(t *testing.T) {
 
 		for entryName, entries := range tt.podEntries {
 			for subEntryName, entry := range entries {
-				stateImpl.SetAllocationInfo(entryName, subEntryName, entry)
+				stateImpl.SetAllocationInfo(entryName, subEntryName, entry, true)
 
 				if entries.IsPoolEntry() {
 					continue
@@ -760,7 +760,7 @@ func TestGetTopEvictionPods(t *testing.T) {
 
 		for entryName, entries := range tt.podEntries {
 			for subEntryName, entry := range entries {
-				stateImpl.SetAllocationInfo(entryName, subEntryName, entry)
+				stateImpl.SetAllocationInfo(entryName, subEntryName, entry, true)
 
 				if entries.IsPoolEntry() {
 					continue
@@ -1630,7 +1630,7 @@ func TestCPUPressureLoadEviction_collectMetrics(t *testing.T) {
 			now := time.Now()
 			for entryName, entries := range tt.podEntries {
 				for subEntryName, entry := range entries {
-					stateImpl.SetAllocationInfo(entryName, subEntryName, entry)
+					stateImpl.SetAllocationInfo(entryName, subEntryName, entry, true)
 
 					if entries.IsPoolEntry() {
 						continue

pkg/agent/qrm-plugins/cpu/dynamicpolicy/cpueviction/strategy/pressure_suppression_test.go

@@ -312,7 +312,7 @@ func TestCPUPressureSuppression_GetEvictPods(t *testing.T) {
 
 			for entryName, entries := range tt.podEntries {
 				for subEntryName, entry := range entries {
-					stateImpl.SetAllocationInfo(entryName, subEntryName, entry)
+					stateImpl.SetAllocationInfo(entryName, subEntryName, entry, true)
 
 					if entries.IsPoolEntry() {
 						continue

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

@@ -464,7 +464,7 @@ func (p *DynamicPolicy) GetResourcesAllocation(_ context.Context,
 				if p.applySidecarAllocationInfoFromMainContainer(allocationInfo, mainContainerAllocationInfo) {
 					general.Infof("pod: %s/%s, container: %s sync allocation info from main container",
 						allocationInfo.PodNamespace, allocationInfo.PodName, containerName)
-					p.state.SetAllocationInfo(podUID, containerName, allocationInfo)
+					p.state.SetAllocationInfo(podUID, containerName, allocationInfo, true)
 					needUpdateMachineState = true
 				}
 			}
@@ -488,11 +488,11 @@ func (p *DynamicPolicy) GetResourcesAllocation(_ context.Context,
 				}
 
 				allocationInfo.InitTimestamp = time.Now().Format(util.QRMTimeFormat)
-				p.state.SetAllocationInfo(podUID, containerName, allocationInfo)
+				p.state.SetAllocationInfo(podUID, containerName, allocationInfo, true)
 			} else if allocationInfo.RampUp && time.Now().After(initTs.Add(p.transitionPeriod)) {
 				general.Infof("pod: %s/%s, container: %s ramp up finished", allocationInfo.PodNamespace, allocationInfo.PodName, allocationInfo.ContainerName)
 				allocationInfo.RampUp = false
-				p.state.SetAllocationInfo(podUID, containerName, allocationInfo)
+				p.state.SetAllocationInfo(podUID, containerName, allocationInfo, true)
 
 				if allocationInfo.CheckShared() {
 					allocationInfosJustFinishRampUp = append(allocationInfosJustFinishRampUp, allocationInfo)
@@ -503,7 +503,7 @@ func (p *DynamicPolicy) GetResourcesAllocation(_ context.Context,
 	}
 
 	if len(allocationInfosJustFinishRampUp) > 0 {
-		if err = p.putAllocationsAndAdjustAllocationEntries(allocationInfosJustFinishRampUp, true); err != nil {
+		if err = p.putAllocationsAndAdjustAllocationEntries(allocationInfosJustFinishRampUp, true, true); err != nil {
 			// not influencing return response to kubelet when putAllocationsAndAdjustAllocationEntries failed
 			general.Errorf("putAllocationsAndAdjustAllocationEntries failed with error: %v", err)
 		}
@@ -517,7 +517,7 @@ func (p *DynamicPolicy) GetResourcesAllocation(_ context.Context,
 			general.Errorf("GetResourcesAllocation GenerateMachineStateFromPodEntries failed with error: %v", err)
 			return nil, fmt.Errorf("GenerateMachineStateFromPodEntries failed with error: %v", err)
 		}
-		p.state.SetMachineState(updatedMachineState)
+		p.state.SetMachineState(updatedMachineState, true)
 	}
 
 	podEntries = p.state.GetPodEntries()
@@ -843,14 +843,17 @@ func (p *DynamicPolicy) Allocate(ctx context.Context,
 			if err != nil {
 				resp = nil
 				respErr = fmt.Errorf("add container to qos aware server failed with error: %v", err)
-				_ = p.removeContainer(req.PodUid, req.ContainerName)
+				_ = p.removeContainer(req.PodUid, req.ContainerName, false)
 			}
 		} else if respErr != nil {
-			_ = p.removeContainer(req.PodUid, req.ContainerName)
+			_ = p.removeContainer(req.PodUid, req.ContainerName, false)
 			_ = p.emitter.StoreInt64(util.MetricNameAllocateFailed, 1, metrics.MetricTypeNameRaw,
 				metrics.MetricTag{Key: "error_message", Val: metric.MetricTagValueFormat(respErr)},
 				metrics.MetricTag{Key: util.MetricTagNameInplaceUpdateResizing, Val: strconv.FormatBool(util.PodInplaceUpdateResizing(req))})
 		}
+		if err := p.state.StoreState(); err != nil {
+			general.ErrorS(err, "store state failed", "podName", req.PodName, "containerName", req.ContainerName)
+		}
 
 		p.Unlock()
 		if respErr != nil {
@@ -908,7 +911,7 @@ func (p *DynamicPolicy) Allocate(ctx context.Context,
 	if p.allocationHandlers[qosLevel] == nil {
 		return nil, fmt.Errorf("katalyst QoS level: %s is not supported yet", qosLevel)
 	}
-	return p.allocationHandlers[qosLevel](ctx, req)
+	return p.allocationHandlers[qosLevel](ctx, req, false)
 }
 
 // AllocateForPod is called during pod admit so that the resource
@@ -961,34 +964,40 @@ func (p *DynamicPolicy) RemovePod(ctx context.Context,
 		}
 	}
 
-	err = p.removePod(req.PodUid, podEntries)
+	err = p.removePod(req.PodUid, podEntries, false)
 	if err != nil {
 		general.ErrorS(err, "remove pod failed with error", "podUID", req.PodUid)
 		return nil, err
 	}
 
-	aErr := p.adjustAllocationEntries()
+	aErr := p.adjustAllocationEntries(false)
 	if aErr != nil {
 		general.ErrorS(aErr, "adjustAllocationEntries failed", "podUID", req.PodUid)
 	}
+	if err := p.state.StoreState(); err != nil {
+		general.ErrorS(err, "store state failed", "podUID", req.PodUid)
+	}
 
 	return &pluginapi.RemovePodResponse{}, nil
 }
 
-func (p *DynamicPolicy) removePod(podUID string, podEntries state.PodEntries) error {
+func (p *DynamicPolicy) removePod(podUID string, podEntries state.PodEntries, persistCheckpoint bool) error {
 	delete(podEntries, podUID)
 
 	updatedMachineState, err := generateMachineStateFromPodEntries(p.machineInfo.CPUTopology, podEntries)
 	if err != nil {
 		return fmt.Errorf("GenerateMachineStateFromPodEntries failed with error: %v", err)
 	}
 
-	p.state.SetPodEntries(podEntries)
-	p.state.SetMachineState(updatedMachineState)
+	p.state.SetPodEntries(podEntries, false)
+	p.state.SetMachineState(updatedMachineState, false)
+	if persistCheckpoint {
+		return p.state.StoreState()
+	}
 	return nil
 }
 
-func (p *DynamicPolicy) removeContainer(podUID, containerName string) error {
+func (p *DynamicPolicy) removeContainer(podUID, containerName string, persistCheckpoint bool) error {
 	podEntries := p.state.GetPodEntries()
 
 	found := false
@@ -1007,8 +1016,11 @@ func (p *DynamicPolicy) removeContainer(podUID, containerName string) error {
 		return fmt.Errorf("GenerateMachineStateFromPodEntries failed with error: %v", err)
 	}
 
-	p.state.SetPodEntries(podEntries)
-	p.state.SetMachineState(updatedMachineState)
+	p.state.SetPodEntries(podEntries, false)
+	p.state.SetMachineState(updatedMachineState, false)
+	if persistCheckpoint {
+		return p.state.StoreState()
+	}
 	return nil
 }
 
@@ -1074,8 +1086,11 @@ func (p *DynamicPolicy) cleanPools() error {
 			return fmt.Errorf("calculate machineState by podEntries failed with error: %v", err)
 		}
 
-		p.state.SetPodEntries(podEntries)
-		p.state.SetMachineState(machineState)
+		p.state.SetPodEntries(podEntries, false)
+		p.state.SetMachineState(machineState, false)
+		if err := p.state.StoreState(); err != nil {
+			general.ErrorS(err, "store state failed")
+		}
 	} else {
 		general.Infof("there is no pool to delete")
 	}
@@ -1105,7 +1120,7 @@ func (p *DynamicPolicy) initReservePool() error {
 		TopologyAwareAssignments:         topologyAwareAssignments,
 		OriginalTopologyAwareAssignments: machine.DeepcopyCPUAssignment(topologyAwareAssignments),
 	}
-	p.state.SetAllocationInfo(commonstate.PoolNameReserve, commonstate.FakedContainerName, curReserveAllocationInfo)
+	p.state.SetAllocationInfo(commonstate.PoolNameReserve, commonstate.FakedContainerName, curReserveAllocationInfo, true)
 
 	return nil
 }
@@ -1173,7 +1188,7 @@ func (p *DynamicPolicy) initReclaimPool() error {
 			TopologyAwareAssignments:         topologyAwareAssignments,
 			OriginalTopologyAwareAssignments: machine.DeepcopyCPUAssignment(topologyAwareAssignments),
 		}
-		p.state.SetAllocationInfo(commonstate.PoolNameReclaim, commonstate.FakedContainerName, curPoolAllocationInfo)
+		p.state.SetAllocationInfo(commonstate.PoolNameReclaim, commonstate.FakedContainerName, curPoolAllocationInfo, true)
 	} else {
 		general.Infof("exist initial %s: %s", commonstate.PoolNameReclaim, reclaimedAllocationInfo.AllocationResult.String())
 	}

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

@@ -458,7 +458,7 @@ func (p *DynamicPolicy) allocateByCPUAdvisor(
 	if curAllowSharedCoresOverlapReclaimedCores != resp.AllowSharedCoresOverlapReclaimedCores {
 		general.Infof("set allowSharedCoresOverlapReclaimedCores from %v to %v",
 			curAllowSharedCoresOverlapReclaimedCores, resp.AllowSharedCoresOverlapReclaimedCores)
-		p.state.SetAllowSharedCoresOverlapReclaimedCores(resp.AllowSharedCoresOverlapReclaimedCores)
+		p.state.SetAllowSharedCoresOverlapReclaimedCores(resp.AllowSharedCoresOverlapReclaimedCores, true)
 	}
 
 	return nil
@@ -814,9 +814,11 @@ func (p *DynamicPolicy) applyBlocks(blockCPUSet advisorapi.BlockCPUSet, resp *ad
 	if err != nil {
 		return fmt.Errorf("calculate machineState by newPodEntries failed with error: %v", err)
 	}
-	p.state.SetPodEntries(newEntries)
-	p.state.SetMachineState(newMachineState)
-
+	p.state.SetPodEntries(newEntries, false)
+	p.state.SetMachineState(newMachineState, false)
+	if err := p.state.StoreState(); err != nil {
+		general.ErrorS(err, "store state failed")
+	}
 	return nil
 }
 
@@ -861,7 +863,7 @@ func (p *DynamicPolicy) applyNUMAHeadroom(resp *advisorapi.ListAndWatchResponse)
 				advisorapi.ControlKnobKeyCPUNUMAHeadroom, cpuNUMAHeadroomValue, err)
 		}
 
-		p.state.SetNUMAHeadroom(*cpuNUMAHeadroom)
+		p.state.SetNUMAHeadroom(*cpuNUMAHeadroom, true)
 		general.Infof("cpuNUMAHeadroom: %v", cpuNUMAHeadroom)
 	}