func filterPools(pools []*Pool, test func(*Pool) bool) (ret []*Pool) {
	for _, pool := range pools {
		if test(pool) {
			ret = append(ret, pool)
		}
	}
	return
}
filterPools returns pools for which the test function returns true
package podpools
import (
	"fmt"
	"sort"
	"strconv"
	"strings"
	corev1 "k8s.io/api/core/v1"
	resapi "k8s.io/apimachinery/pkg/api/resource"
	"k8s.io/kubernetes/pkg/kubelet/cm/cpuset"
	pkgcfg "github.com/intel/cri-resource-manager/pkg/config"
	"github.com/intel/cri-resource-manager/pkg/cpuallocator"
	"github.com/intel/cri-resource-manager/pkg/cri/resource-manager/cache"
	"github.com/intel/cri-resource-manager/pkg/cri/resource-manager/events"
	"github.com/intel/cri-resource-manager/pkg/cri/resource-manager/introspect"
	"github.com/intel/cri-resource-manager/pkg/cri/resource-manager/kubernetes"
	"github.com/intel/cri-resource-manager/pkg/cri/resource-manager/policy"
	policyapi "github.com/intel/cri-resource-manager/pkg/cri/resource-manager/policy"
	logger "github.com/intel/cri-resource-manager/pkg/log"
	"github.com/intel/cri-resource-manager/pkg/utils"
	idset "github.com/intel/goresctrl/pkg/utils"
)
const (
	PolicyName = "podpools"
	PolicyDescription = "Pod-granularity workload placement"
	PolicyPath = "policy." + PolicyName
	podpoolKey = "pool." + PolicyName + "." + kubernetes.ResmgrKeyNamespace
	reservedPoolDefName = "reserved"
	defaultPoolDefName = "default"
	podMilliCPUErrorMargin = int64(10)
)
type podpools struct {
	options         *policyapi.BackendOptions  
	ppoptions       PodpoolsOptions            
	cch             cache.Cache                
	allowed         cpuset.CPUSet              
	reserved        cpuset.CPUSet              
	reservedPoolDef *PoolDef                   
	defaultPoolDef  *PoolDef                   
	pools           []*Pool                    
	podMaxMilliCPU  map[string]int64           
	cpuAllocator    cpuallocator.CPUAllocator  
}
type Pool struct {
	Def *PoolDef
	Instance int
	CPUs cpuset.CPUSet
	Mems idset.IDSet
	PodIDs map[string][]string
}
var log logger.Logger = logger.NewLogger("policy")
func (pool Pool) String() string {
	podCount := len(pool.PodIDs)
	contCount := 0
	for _, contIDs := range pool.PodIDs {
		contCount += len(contIDs)
	}
	s := fmt.Sprintf("%s{cpus:%s, mems:%s, pods:%d/%d, containers:%d}",
		pool.PrettyName(), pool.CPUs, pool.Mems, podCount, pool.Def.MaxPods, contCount)
	return s
}
func (pool Pool) PrettyName() string {
	return fmt.Sprintf("%s[%d]", pool.Def.Name, pool.Instance)
}
func CreatePodpoolsPolicy(policyOptions *policy.BackendOptions) policy.Backend {
	p := &podpools{
		options: policyOptions,
		cch:     policyOptions.Cache,
		reservedPoolDef: &PoolDef{
			Name:    reservedPoolDefName,
			MaxPods: 0,
		},
		defaultPoolDef: &PoolDef{
			Name:    defaultPoolDefName,
			MaxPods: 0,
		},
		podMaxMilliCPU: make(map[string]int64),
		cpuAllocator:   cpuallocator.NewCPUAllocator(policyOptions.System),
	}
	log.Info("creating %s policy...", PolicyName)
	if allowed, ok := policyOptions.Available[policyapi.DomainCPU]; ok {
		p.allowed = allowed.(cpuset.CPUSet)
	} else {
		p.allowed = policyOptions.System.CPUSet().Difference(policyOptions.System.Offlined())
	}
	p.reserved = cpuset.NewCPUSet()
	if reserved, ok := p.options.Reserved[policyapi.DomainCPU]; ok {
		switch v := reserved.(type) {
		case cpuset.CPUSet:
			p.reserved = p.allowed.Intersection(v)
		case resapi.Quantity:
			reserveCnt := (int(v.MilliValue()) + 999) / 1000
			cpus, err := p.cpuAllocator.AllocateCpus(&p.allowed, reserveCnt, cpuallocator.PriorityNone)
			if err != nil {
				log.Fatal("failed to allocate reserved CPUs: %s", err)
			}
			p.reserved = cpus
			p.allowed = p.allowed.Union(cpus)
		}
	}
	if p.reserved.IsEmpty() {
		log.Fatal("%s cannot run without reserved CPUs that are also AvailableResources", PolicyName)
	}
	log.Debug("creating %s configuration", PolicyName)
	if err := p.setConfig(podpoolsOptions); err != nil {
		log.Fatal("failed to create %s policy: %v", PolicyName, err)
	}
	pkgcfg.GetModule(PolicyPath).AddNotify(p.configNotify)
	return p
}
func (p *podpools) Name() string {
	return PolicyName
}
func (p *podpools) Description() string {
	return PolicyDescription
}
func (p *podpools) Start(add []cache.Container, del []cache.Container) error {
	log.Info("%s policy started", PolicyName)
	return p.Sync(add, del)
}
func (p *podpools) Sync(add []cache.Container, del []cache.Container) error {
	log.Debug("synchronizing state...")
	for _, c := range del {
		p.ReleaseResources(c)
	}
	for _, c := range add {
		p.AllocateResources(c)
	}
	return nil
}
func (p *podpools) AllocateResources(c cache.Container) error {
	log.Debug("allocating container %s...", c.PrettyName())
	pod, ok := c.GetPod()
	if !ok {
		return podpoolsError("cannot find pod of container %s from the cache", c.PrettyName())
	}
	if pool := p.allocatePool(pod); pool != nil {
		p.assignContainer(c, pool)
		p.trackPodCPU(pod, pool)
		if log.DebugEnabled() {
			log.Debug(p.dumpPool(pool))
		}
	} else {
		return podpoolsError("cannot find CPUs to run container %s - no default or reserved CPUs available", c.PrettyName())
	}
	return nil
}
func (p *podpools) ReleaseResources(c cache.Container) error {
	log.Debug("releasing container %s...", c.PrettyName())
	pod, ok := c.GetPod()
	if !ok {
		return podpoolsError("cannot find pod of container %s from the cache", c.PrettyName())
	}
	if pool := p.allocatedPool(pod); pool != nil {
		p.dismissContainer(c, pool)
		if log.DebugEnabled() {
			log.Debug(p.dumpPool(pool))
		}
		if p.containersInPool(pod, pool) == 0 {
			log.Debug("all containers removed, free pool allocation %s for pod %q", pool.PrettyName(), pod.GetName())
			p.validatePodCPU(pod, pool)
			p.freePool(pod, pool)
		}
	} else {
		log.Debug("ReleaseResources: pool-less container %s, nothing to release", c.PrettyName())
	}
	return nil
}
func (p *podpools) UpdateResources(c cache.Container) error {
	log.Debug("(not) updating container %s...", c.PrettyName())
	return nil
}
func (p *podpools) Rebalance() (bool, error) {
	log.Debug("(not) rebalancing containers...")
	return false, nil
}
func (p *podpools) HandleEvent(*events.Policy) (bool, error) {
	log.Debug("(not) handling event...")
	return false, nil
}
func (p *podpools) ExportResourceData(c cache.Container) map[string]string {
	return nil
}
func (p *podpools) Introspect(*introspect.State) {
	return
}
func (p *podpools) allocatedPool(pod cache.Pod) *Pool {
	podID := pod.GetID()
	pools := filterPools(p.pools,
		func(pl *Pool) bool { _, ok := pl.PodIDs[podID]; return ok })
	if len(pools) == 0 {
		return nil
	}
	return pools[0]
}
func (p *podpools) allocatePool(pod cache.Pod) *Pool {
	if pool := p.allocatedPool(pod); pool != nil {
		return pool
	}
	poolDef := p.getPoolDef(pod)
	if poolDef == nil {
		return nil
	}
	pools := filterPools(p.pools,
		func(pl *Pool) bool {
			return poolDef.Name == pl.Def.Name && (pl.Def.MaxPods > len(pl.PodIDs) || pl.Def.MaxPods == 0)
		})
	switch poolDef.FillOrder {
	case FillBalanced:
		sort.Slice(pools, func(i, j int) bool {
			return len(pools[i].PodIDs) < len(pools[j].PodIDs)
		})
	case FillPacked:
		sort.Slice(pools, func(i, j int) bool {
			return len(pools[i].PodIDs) > len(pools[j].PodIDs)
		})
	case FillFirstFree:
	}
	if len(pools) == 0 {
		log.Error("cannot find free %q pool for pod %q, falling back to %q", poolDef.Name, pod.GetName(), defaultPoolDefName)
		pools = []*Pool{p.pools[1]}
	}
	podID := pod.GetID()
	pool := pools[0]
	pool.PodIDs[podID] = []string{}
	log.Debug("allocated pool %s[%d] for pod %q", pool.Def.Name, pool.Instance, pod.GetName())
	return pool
}
func (p *podpools) containersInPool(pod cache.Pod, pool *Pool) int {
	if cnts, ok := pool.PodIDs[pod.GetID()]; ok {
		return len(cnts)
	}
	return 0
}
func (p *podpools) dumpPool(pool *Pool) string {
	conts := []string{}
	pods := []string{}
	for podID, contIDs := range pool.PodIDs {
		podName := podID
		if pod, ok := p.cch.LookupPod(podID); ok {
			podName = pod.GetName()
		}
		pods = append(pods, fmt.Sprintf("%s (mCPU: %d, max=%d)", podName, p.getPodMilliCPU(podID), p.podMaxMilliCPU[podID]))
		for _, contID := range contIDs {
			if cont, ok := p.cch.LookupContainer(contID); ok {
				conts = append(conts, cont.PrettyName())
			} else {
				conts = append(conts, podName+":"+contID)
			}
		}
	}
	s := fmt.Sprintf("Pool{Def.Name: %q, Instance: %d, CPUs: %s, Mems: %s, Def.MaxPods: %d, pods: %v, containers:%v}",
		pool.Def.Name, pool.Instance, pool.CPUs, pool.Mems, pool.Def.MaxPods, pods, conts)
	return s
}
func (p *podpools) freePool(pod cache.Pod, pool *Pool) {
	podID := pod.GetID()
	delete(pool.PodIDs, podID)
	delete(p.podMaxMilliCPU, podID)
}
func (p *podpools) trackPodCPU(pod cache.Pod, pool *Pool) {
	podID := pod.GetID()
	current := p.getPodMilliCPU(podID)
	if max, ok := p.podMaxMilliCPU[podID]; ok {
		if max < current {
			p.podMaxMilliCPU[podID] = current
		}
	} else {
		p.podMaxMilliCPU[podID] = current
	}
	if cpuAvail := p.availableMilliCPUs(pool); cpuAvail < 0 {
		log.Error("overbooked pool %q, cpuset:%s: %dm / %dm CPUs used, %d mCPU available", pool.PrettyName(), pool.CPUs, pool.CPUs.Size()*1000-int(cpuAvail), pool.CPUs.Size()*1000, cpuAvail)
	}
}
func (p *podpools) validatePodCPU(pod cache.Pod, pool *Pool) {
	podID := pod.GetID()
	if podmCPU, ok := p.podMaxMilliCPU[podID]; ok {
		if pool.Def.MaxPods > 0 {
			poolmCPUperPod := int64(pool.CPUs.Size() * 1000 / pool.Def.MaxPods)
			mCPUerr := podmCPU - poolmCPUperPod
			if mCPUerr < -podMilliCPUErrorMargin || mCPUerr > podMilliCPUErrorMargin {
				podName := ""
				if pod, ok := p.cch.LookupPod(podID); ok {
					podName = pod.GetName()
				}
				log.Error("bad CPU requests: pod %q requested %d mCPUs, but in pool %q pods must request %d mCPUs.", podName, podmCPU, pool.Def.Name, poolmCPUperPod)
			}
		}
	}
}
func (p *podpools) getPodMilliCPU(podID string) int64 {
	cpuRequested := int64(0)
	for _, c := range p.cch.GetContainers() {
		if c.GetPodID() == podID {
			if reqCpu, ok := c.GetResourceRequirements().Requests[corev1.ResourceCPU]; ok {
				cpuRequested += reqCpu.MilliValue()
			}
		}
	}
	return cpuRequested
}
func (p *podpools) configNotify(event pkgcfg.Event, source pkgcfg.Source) error {
	log.Info("configuration %s", event)
	if err := p.setConfig(podpoolsOptions); err != nil {
		log.Error("config update failed: %v", err)
		return err
	}
	log.Info("config updated successfully")
	p.Sync(p.cch.GetContainers(), nil)
	return nil
}
func (p *podpools) getPoolDefName(pod cache.Pod) string {
	if poolDefName, ok := pod.GetEffectiveAnnotation(podpoolKey, ""); ok {
		return poolDefName
	}
	if pod.GetNamespace() == "kube-system" {
		return reservedPoolDefName
	}
	return defaultPoolDefName
}
func (p *podpools) getPoolDef(pod cache.Pod) *PoolDef {
	poolDefName := p.getPoolDefName(pod)
	if poolDefName == reservedPoolDefName {
		return p.reservedPoolDef
	}
	if poolDefName == defaultPoolDefName {
		return p.defaultPoolDef
	}
	for _, poolDef := range p.ppoptions.PoolDefs {
		if poolDef.Name == poolDefName {
			return poolDef
		}
	}
	log.Error("pod %q pool %q does not match any pool definition, falling back to %q", pod.GetName(), poolDefName, p.defaultPoolDef.Name)
	return p.defaultPoolDef
}
func (p *podpools) applyPoolDef(pools *[]*Pool, poolDef *PoolDef, freeCpus *cpuset.CPUSet, nonReservedCpuCount int) error {
	if len(*pools) < 2 {
		return podpoolsError("internal error: reserved and default pools missing, cannot apply pool definitions")
	}
	reservedPool := (*pools)[0]
	defaultPool := (*pools)[1]
	switch poolDef.Name {
	case "":
		return podpoolsError("undefined or empty pool name")
	case reservedPool.Def.Name:
		if poolDef.CPU != "" || poolDef.Instances != "" {
			poolCount, cpusPerPool, err := parseInstancesCPUs(poolDef.Instances, poolDef.CPU, nonReservedCpuCount)
			if err != nil {
				return podpoolsError("pool %q: %w", poolDef.Name, err)
			}
			if poolCount != 1 {
				return podpoolsError("pool %q: cannot change the number of instances", poolDef.Name)
			}
			if cpusPerPool != reservedPool.CPUs.Size() {
				return podpoolsError("pool %q: number of CPUs is conflicting ReservedResources CPUs", poolDef.Name)
			}
		}
		reservedPool.Def.MaxPods = poolDef.MaxPods
	case defaultPool.Def.Name:
		if poolDef.CPU != "" || poolDef.Instances != "" {
			poolCount, cpusPerPool, err := parseInstancesCPUs(poolDef.Instances, poolDef.CPU, nonReservedCpuCount)
			if err != nil {
				return podpoolsError("pool %q: %w", poolDef.Name, err)
			}
			if poolCount != 1 {
				return podpoolsError("pool %q: cannot change the number of instances", poolDef.Name)
			}
			cpus, err := p.cpuAllocator.AllocateCpus(freeCpus, cpusPerPool, cpuallocator.PriorityNormal)
			if err != nil {
				return podpoolsError("could not allocate %d CPUs for pool %q: %w", cpusPerPool, poolDef.Name, err)
			}
			defaultPool.CPUs = cpus
		}
		defaultPool.Def.MaxPods = poolDef.MaxPods
	default:
		poolCount, cpusPerPool, err := parseInstancesCPUs(poolDef.Instances, poolDef.CPU, nonReservedCpuCount)
		if err != nil {
			return podpoolsError("pool %q: %w", poolDef.Name, err)
		}
		if poolCount == 0 {
			return podpoolsError("pool %q: insufficient CPUs to create any instances", poolDef.Name)
		}
		if poolCount > 1 && poolDef.FillOrder == FillPacked && poolDef.MaxPods == 0 {
			return podpoolsError("pool %q: %d pool(s) unreachable due to unlimited pod capacity and FillOrder: %s", poolDef.Name, poolCount-1, poolDef.FillOrder)
		}
		log.Debug("allocating %d out of %d non-reserved CPUs for %d %q pools", poolCount*cpusPerPool, nonReservedCpuCount, poolCount, poolDef.Name)
		for poolIndex := 0; poolIndex < poolCount; poolIndex++ {
			if cpusPerPool > freeCpus.Size() {
				return podpoolsError("insufficient CPUs when trying to allocate %d CPUs for pool %s[%d]", cpusPerPool, poolDef.Name, poolIndex)
			}
			cpus, err := p.cpuAllocator.AllocateCpus(freeCpus, cpusPerPool, cpuallocator.PriorityNormal)
			if err != nil {
				return podpoolsError("could not allocate %d CPUs for instance %d of pool %q: %w", cpusPerPool, poolIndex, poolDef.Name, err)
			}
			pool := Pool{
				Def:      poolDef,
				Instance: poolIndex,
				CPUs:     cpus,
			}
			*pools = append(*pools, &pool)
		}
	}
	return nil
}
func (p *podpools) setConfig(ppoptions *PodpoolsOptions) error {
	pools := []*Pool{}
	reservedPool := Pool{
		Def:  p.reservedPoolDef,
		CPUs: p.reserved,
	}
	pools = append(pools, &reservedPool)
	defaultPool := Pool{
		Def:  p.defaultPoolDef,
		CPUs: reservedPool.CPUs,
	}
	pools = append(pools, &defaultPool)
	freeCpus := p.allowed.Clone()
	freeCpus = freeCpus.Difference(p.reserved)
	nonReservedCpuCount := freeCpus.Size()
	userPoolDefs := 0
	for _, poolDef := range ppoptions.PoolDefs {
		if err := p.applyPoolDef(&pools, poolDef, &freeCpus, nonReservedCpuCount); err != nil {
			return err
		}
		if poolDef.Name != reservedPoolDefName && poolDef.Name != defaultPoolDefName {
			userPoolDefs += 1
		}
	}
	if freeCpus.Size() > 0 {
		if defaultPool.CPUs.Intersection(reservedPool.CPUs).IsEmpty() {
			log.Debug("%d unused CPUs are added to the default pool.", freeCpus.Size())
			defaultPool.CPUs = defaultPool.CPUs.Union(freeCpus)
		} else {
			log.Debug("%d unused CPUs are used as the default pool.", freeCpus.Size())
			defaultPool.CPUs = freeCpus
		}
	}
	log.Info("%s policy pools:", PolicyName)
	for index, pool := range pools {
		pool.Mems = p.closestMems(pool.CPUs)
		pool.PodIDs = make(map[string][]string)
		log.Info("- pool %d: %s", index, pool)
	}
	log.Debug("new %s configuration:\n%s", PolicyName, utils.DumpJSON(ppoptions))
	p.pools = pools
	p.ppoptions = *ppoptions
	if userPoolDefs > 1 {
		log.Warn("Multiple (%d) user-defined pool definitions on the node. kube-scheduler does not know which of the pools has CPUs left for new workloads, and may overbook pools on the node.", userPoolDefs)
	}
	return nil
}
func (p *podpools) closestMems(cpus cpuset.CPUSet) idset.IDSet {
	mems := idset.NewIDSet()
	sys := p.options.System
	for _, nodeID := range sys.NodeIDs() {
		if !cpus.Intersection(sys.Node(nodeID).CPUSet()).IsEmpty() {
			mems.Add(nodeID)
		}
	}
	return mems
}