feat(gpu): implement topology hints for associated devices

cmd/katalyst-agent/app/options/dynamic/adminqos/reclaimedresource/memoryheadroom/util_based.go

@@ -27,6 +27,7 @@ const (
 	defaultFreeBasedRatio      = 0.6
 	defaultStaticBasedCapacity = 20 << 30 // 20GB
 	defaultCacheBasedRatio     = 0
+	defaultRequestBasedRatio   = 0.1
 	defaultMaxOversoldRate     = 2.0
 )
 
@@ -35,6 +36,7 @@ type UtilBasedOptions struct {
 	FreeBasedRatio      float64
 	StaticBasedCapacity float64
 	CacheBasedRatio     float64
+	RequestBasedRatio   float64
 	MaxOversoldRate     float64
 }
 
@@ -44,6 +46,7 @@ func NewUtilBasedOptions() *UtilBasedOptions {
 		FreeBasedRatio:      defaultFreeBasedRatio,
 		StaticBasedCapacity: defaultStaticBasedCapacity,
 		CacheBasedRatio:     defaultCacheBasedRatio,
+		RequestBasedRatio:   defaultRequestBasedRatio,
 		MaxOversoldRate:     defaultMaxOversoldRate,
 	}
 }
@@ -57,6 +60,8 @@ func (o *UtilBasedOptions) AddFlags(fs *pflag.FlagSet) {
 		"the static oversold memory size by bytes")
 	fs.Float64Var(&o.CacheBasedRatio, "memory-headroom-cache-based-ratio", o.CacheBasedRatio,
 		"the rate of cache oversold, 0 means disable cache oversold")
+	fs.Float64Var(&o.RequestBasedRatio, "memory-headroom-request-based-ratio", o.RequestBasedRatio,
+		"the rate of reserved memory for request, 0 means disable reserved memory")
 	fs.Float64Var(&o.MaxOversoldRate, "memory-headroom-max-oversold-rate", o.MaxOversoldRate,
 		"the max oversold rate of memory headroom to the memory limit of reclaimed_cores cgroup")
 }
@@ -66,6 +71,7 @@ func (o *UtilBasedOptions) ApplyTo(c *memoryheadroom.MemoryUtilBasedConfiguratio
 	c.FreeBasedRatio = o.FreeBasedRatio
 	c.StaticBasedCapacity = o.StaticBasedCapacity
 	c.CacheBasedRatio = o.CacheBasedRatio
+	c.RequestBasedRatio = o.RequestBasedRatio
 	c.MaxOversoldRate = o.MaxOversoldRate
 	return nil
 }

cmd/katalyst-agent/app/options/qrm/cpu_plugin.go

@@ -54,6 +54,7 @@ type CPUDynamicPolicyOptions struct {
 	EnableCPUBurst                      bool
 	EnableDefaultDedicatedCoresCPUBurst bool
 	EnableDefaultSharedCoresCPUBurst    bool
+	EnableCPUBurstForMainContainerOnly  bool
 	*irqtuner.IRQTunerOptions
 	*hintoptimizer.HintOptimizerOptions
 }
@@ -140,6 +141,8 @@ func (o *CPUOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		o.EnableDefaultSharedCoresCPUBurst, "if set true, it will enable cpu burst for shared cores by default")
 	fs.BoolVar(&o.EnableDefaultDedicatedCoresCPUBurst, "enable-default-dedicated-cores-cpu-burst",
 		o.EnableDefaultDedicatedCoresCPUBurst, "if set true, it will enable cpu burst for dedicated cores by default")
+	fs.BoolVar(&o.EnableCPUBurstForMainContainerOnly, "enable-cpu-burst-for-main-container-only",
+		o.EnableCPUBurstForMainContainerOnly, "if set true, it will enable cpu burst for main container only")
 	o.HintOptimizerOptions.AddFlags(fss)
 	o.IRQTunerOptions.AddFlags(fss)
 }
@@ -164,6 +167,7 @@ func (o *CPUOptions) ApplyTo(conf *qrmconfig.CPUQRMPluginConfig) error {
 	conf.EnableCPUBurst = o.EnableCPUBurst
 	conf.EnableDefaultDedicatedCoresCPUBurst = o.EnableDefaultDedicatedCoresCPUBurst
 	conf.EnableDefaultSharedCoresCPUBurst = o.EnableDefaultSharedCoresCPUBurst
+	conf.EnableCPUBurstForMainContainerOnly = o.EnableCPUBurstForMainContainerOnly
 	if err := o.HintOptimizerOptions.ApplyTo(conf.HintOptimizerConfiguration); err != nil {
 		return err
 	}

cmd/katalyst-agent/app/options/qrm/mb_plugin.go

@@ -52,6 +52,7 @@ type MBOptions struct {
 	CrossDomainGroups              []string
 	ResetResctrlOnly               bool
 	LocalIsVictimAndTotalIsAllRead bool
+	ExtraGroupPriorities           map[string]int
 }
 
 func NewMBOptions() *MBOptions {
@@ -89,6 +90,8 @@ func (o *MBOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		o.ResetResctrlOnly, "not to run mb plugin really, and only reset to ensure resctrl FS in default status")
 	fs.BoolVar(&o.LocalIsVictimAndTotalIsAllRead, "mb-local-is-victim",
 		o.LocalIsVictimAndTotalIsAllRead, "turn resctrl local as victim")
+	fs.StringToIntVar(&o.ExtraGroupPriorities, "mb-extra-group-priorities",
+		o.ExtraGroupPriorities, "extra resctrl groups with priorities")
 }
 
 func (o *MBOptions) ApplyTo(conf *qrm.MBQRMPluginConfig) error {
@@ -103,5 +106,6 @@ func (o *MBOptions) ApplyTo(conf *qrm.MBQRMPluginConfig) error {
 	conf.DomainGroupAwareCapacityPCT = o.DomainGroupAwareCapacityPCT
 	conf.ResetResctrlOnly = o.ResetResctrlOnly
 	conf.LocalIsVictimAndTotalIsAllRead = o.LocalIsVictimAndTotalIsAllRead
+	conf.ExtraGroupPriorities = o.ExtraGroupPriorities
 	return nil
 }

cmd/katalyst-agent/app/options/qrm/sriov_plugin.go

@@ -46,6 +46,7 @@ type SriovStaticPolicyOptions struct {
 }
 
 type SriovDynamicPolicyOptions struct {
+	PodRequiredAnnotations      map[string]string
 	LargeSizeVFQueueCount       int
 	LargeSizeVFCPUThreshold     int
 	LargeSizeVFFailOnExhaustion bool
@@ -69,6 +70,7 @@ func NewSriovOptions() *SriovOptions {
 			MaxBondingVFQueueCount: math.MaxInt32,
 		},
 		SriovDynamicPolicyOptions: SriovDynamicPolicyOptions{
+			PodRequiredAnnotations:      map[string]string{},
 			LargeSizeVFQueueCount:       32,
 			LargeSizeVFCPUThreshold:     24,
 			LargeSizeVFFailOnExhaustion: true,
@@ -90,6 +92,7 @@ func (o *SriovOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 	fs.StringToStringVar(&o.ExtraAnnotations, "sriov-vf-extra-annotations", o.ExtraAnnotations, "Extra annotations for sriov vf")
 	fs.IntVar(&o.MinBondingVFQueueCount, "static-min-bonding-vf-queue-count", o.MinBondingVFQueueCount, "Min queue count of bonding VF can be allocated in static policy")
 	fs.IntVar(&o.MaxBondingVFQueueCount, "static-max-bonding-vf-queue-count", o.MaxBondingVFQueueCount, "Max queue count of bonding VF can be allocated in static policy")
+	fs.StringToStringVar(&o.PodRequiredAnnotations, "dynamic-pod-vf-required-annotations", o.PodRequiredAnnotations, "Required annotations for pod to dynamic allocate VF")
 	fs.IntVar(&o.LargeSizeVFQueueCount, "dynamic-large-size-vf-queue-count", o.LargeSizeVFQueueCount, "Queue count for VF to be identified as large size VF in dynamic policy")
 	fs.IntVar(&o.LargeSizeVFCPUThreshold, "dynamic-large-size-vf-cpu-threshold", o.LargeSizeVFCPUThreshold, "Threshold of cpu quantity to allocate large size VF in dynamic policy")
 	fs.BoolVar(&o.LargeSizeVFFailOnExhaustion, "dynamic-large-size-vf-fail-on-exhaustion", o.LargeSizeVFFailOnExhaustion, "Should fail or not when large size VF is exhausted in dynamic policy")
@@ -109,6 +112,7 @@ func (s *SriovOptions) ApplyTo(config *qrmconfig.SriovQRMPluginConfig) error {
 	}
 	config.MinBondingVFQueueCount = s.MinBondingVFQueueCount
 	config.MaxBondingVFQueueCount = s.MaxBondingVFQueueCount
+	config.PodRequiredAnnotations = s.PodRequiredAnnotations
 	config.LargeSizeVFQueueCount = s.LargeSizeVFQueueCount
 	config.LargeSizeVFCPUThreshold = s.LargeSizeVFCPUThreshold
 	config.LargeSizeVFFailOnExhaustion = s.LargeSizeVFFailOnExhaustion

cmd/katalyst-agent/app/options/reporter/reporter_base.go

@@ -35,6 +35,7 @@ const (
 type GenericReporterOptions struct {
 	CollectInterval        time.Duration
 	InnerPlugins           []string
+	AgentReporters         []string
 	RefreshLatestCNRPeriod time.Duration
 	DefaultCNRLabels       map[string]string
 }
@@ -43,6 +44,7 @@ type GenericReporterOptions struct {
 func NewGenericReporterOptions() *GenericReporterOptions {
 	return &GenericReporterOptions{
 		InnerPlugins:           []string{"*"},
+		AgentReporters:         []string{"*"},
 		CollectInterval:        defaultCollectInterval,
 		RefreshLatestCNRPeriod: defaultRefreshLatestCNRPeriod,
 		DefaultCNRLabels:       make(map[string]string),
@@ -61,6 +63,9 @@ func (o *GenericReporterOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 	fs.StringSliceVar(&o.InnerPlugins, "reporter-plugins", o.InnerPlugins, fmt.Sprintf(""+
 		"A list of reporter plugins to enable. '*' enables all on-by-default reporter plugins, 'foo' enables the reporter plugin "+
 		"named 'foo', '-foo' disables the reporter plugin named 'foo'"))
+	fs.StringSliceVar(&o.AgentReporters, "agent-reporters", o.AgentReporters, fmt.Sprintf(""+
+		"A list of agent reporters to enable. '*' enables all on-by-default reporters, 'foo' enables the reporter "+
+		"named 'foo', '-foo' disables the reporter named 'foo'"))
 	fs.StringToStringVar(&o.DefaultCNRLabels, "default-cnr-labels", o.DefaultCNRLabels,
 		"the default labels of cnr created by agent, this config must be consistent with the label-selector in katalyst-controller.")
 }
@@ -69,6 +74,7 @@ func (o *GenericReporterOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 func (o *GenericReporterOptions) ApplyTo(c *reporterconfig.GenericReporterConfiguration) error {
 	c.CollectInterval = o.CollectInterval
 	c.InnerPlugins = o.InnerPlugins
+	c.AgentReporters = o.AgentReporters
 	c.RefreshLatestCNRPeriod = o.RefreshLatestCNRPeriod
 	c.DefaultCNRLabels = o.DefaultCNRLabels
 	return nil

cmd/katalyst-controller/app/controller/lifecycle.go

@@ -41,18 +41,20 @@ func StartLifeCycleController(ctx context.Context, controlCtx *katalystbase.Gene
 		err          error
 	)
 
-	cnrLifecycle, err = lifecycle.NewCNRLifecycle(ctx,
-		conf.GenericConfiguration,
-		conf.GenericControllerConfiguration,
-		conf.ControllersConfiguration.CNRLifecycleConfig,
-		controlCtx.Client,
-		controlCtx.KubeInformerFactory.Core().V1().Nodes(),
-		controlCtx.InternalInformerFactory.Node().V1alpha1().CustomNodeResources(),
-		controlCtx.EmitterPool.GetDefaultMetricsEmitter(),
-	)
-	if err != nil {
-		klog.Errorf("failed to new CNR lifecycle controller")
-		return false, err
+	if conf.LifeCycleConfig.EnableCNRLifecycle {
+		cnrLifecycle, err = lifecycle.NewCNRLifecycle(ctx,
+			conf.GenericConfiguration,
+			conf.GenericControllerConfiguration,
+			conf.ControllersConfiguration.CNRLifecycleConfig,
+			controlCtx.Client,
+			controlCtx.KubeInformerFactory.Core().V1().Nodes(),
+			controlCtx.InternalInformerFactory.Node().V1alpha1().CustomNodeResources(),
+			controlCtx.EmitterPool.GetDefaultMetricsEmitter(),
+		)
+		if err != nil {
+			klog.Errorf("failed to new CNR lifecycle controller")
+			return false, err
+		}
 	}
 
 	if conf.LifeCycleConfig.EnableCNCLifecycle {

cmd/katalyst-controller/app/options/lifecycle.go

@@ -47,6 +47,7 @@ type HealthzOptions struct {
 type LifeCycleOptions struct {
 	EnableHealthz      bool
 	EnableCNCLifecycle bool
+	EnableCNRLifecycle bool
 
 	*HealthzOptions
 }
@@ -56,6 +57,7 @@ func NewLifeCycleOptions() *LifeCycleOptions {
 	return &LifeCycleOptions{
 		EnableHealthz:      false,
 		EnableCNCLifecycle: true,
+		EnableCNRLifecycle: true,
 		HealthzOptions: &HealthzOptions{
 			DryRun:        false,
 			NodeSelector:  "",
@@ -83,6 +85,8 @@ func (o *LifeCycleOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 		"whether to enable the healthz controller")
 	fs.BoolVar(&o.EnableCNCLifecycle, "cnc-lifecycle-enabled", o.EnableCNCLifecycle,
 		"whether to enable the cnc lifecycle controller")
+	fs.BoolVar(&o.EnableCNRLifecycle, "cnr-lifecycle-enabled", o.EnableCNRLifecycle,
+		"whether to enable the cnr lifecycle controller")
 
 	fs.BoolVar(&o.DryRun, "healthz-dry-run", o.DryRun,
 		"a bool to enable and disable dry-run logic of healthz controller.")
@@ -117,6 +121,7 @@ func (o *LifeCycleOptions) AddFlags(fss *cliflag.NamedFlagSets) {
 func (o *LifeCycleOptions) ApplyTo(c *controller.LifeCycleConfig) error {
 	c.EnableHealthz = o.EnableHealthz
 	c.EnableCNCLifecycle = o.EnableCNCLifecycle
+	c.EnableCNRLifecycle = o.EnableCNRLifecycle
 
 	c.DryRun = o.DryRun
 

go.mod

@@ -20,7 +20,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/h2non/gock v1.2.0
 	github.com/klauspost/cpuid/v2 v2.2.6
-	github.com/kubewharf/katalyst-api v0.5.11-0.20260324091059-cae1d07d9882
+	github.com/kubewharf/katalyst-api v0.5.11-0.20260423040236-f1a2330d266e
 	github.com/moby/sys/mountinfo v0.6.2
 	github.com/montanaflynn/stats v0.7.1
 	github.com/opencontainers/runc v1.1.6

go.sum

@@ -574,8 +574,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/kubewharf/katalyst-api v0.5.11-0.20260324091059-cae1d07d9882 h1:4KYYk/mAJAOIYDW5V+43wnjnP8p3bwHXAkAcw/AbzuQ=
-github.com/kubewharf/katalyst-api v0.5.11-0.20260324091059-cae1d07d9882/go.mod h1:BZMVGVl3EP0eCn5xsDgV41/gjYkoh43abIYxrB10e3k=
+github.com/kubewharf/katalyst-api v0.5.11-0.20260423040236-f1a2330d266e h1:/0LP1rzsXEI09g3eqUZCJcyv/XGU8MofdFyi7NBtZA0=
+github.com/kubewharf/katalyst-api v0.5.11-0.20260423040236-f1a2330d266e/go.mod h1:BZMVGVl3EP0eCn5xsDgV41/gjYkoh43abIYxrB10e3k=
 github.com/kubewharf/kubelet v1.24.6-kubewharf-pre.2 h1:2KLMzgntDypiFJRX4fSQJCD+a6zIgHuhcAzd/7nAGmU=
 github.com/kubewharf/kubelet v1.24.6-kubewharf-pre.2/go.mod h1:MxbSZUx3wXztFneeelwWWlX7NAAStJ6expqq7gY2J3c=
 github.com/kyoh86/exportloopref v0.1.7/go.mod h1:h1rDl2Kdj97+Kwh4gdz3ujE7XHmH51Q0lUiZ1z4NLj8=

pkg/agent/qrm-plugins/cpu/dynamicpolicy/cpuburst/manager.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/kubewharf/katalyst-api/pkg/consts"
 	"github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/cpu/util"
+	qrmutil "github.com/kubewharf/katalyst-core/pkg/agent/qrm-plugins/util"
 	"github.com/kubewharf/katalyst-core/pkg/config"
 	"github.com/kubewharf/katalyst-core/pkg/config/agent/dynamic"
 	"github.com/kubewharf/katalyst-core/pkg/metaserver"
@@ -90,10 +91,15 @@ func (m *managerImpl) UpdateCPUBurst(conf *config.Configuration, dynamicConfig *
 			continue
 		}
 
+		var mainContainerName string
+		if conf.EnableCPUBurstForMainContainerOnly {
+			mainContainerName = qrmutil.GetMainContainer(pod, conf.MainContainerAnnotationKey)
+		}
+
 		switch cpuBurstPolicy {
 		case consts.PodAnnotationCPUEnhancementCPUBurstPolicyClosed:
 			// For closed policy, we just set the cpu burst value to be 0.
-			if err = m.updateCPUBurstByPercent(0, pod); err != nil {
+			if err = m.updateCPUBurstByPercent(0, pod, mainContainerName); err != nil {
 				errList = append(errList, fmt.Errorf("error setting cpu burst for policy %s for pod %s: %v",
 					consts.PodAnnotationCPUEnhancementCPUBurstPolicyClosed, pod.Name, err))
 			}
@@ -107,7 +113,7 @@ func (m *managerImpl) UpdateCPUBurst(conf *config.Configuration, dynamicConfig *
 				continue
 			}
 
-			if err = m.updateCPUBurstByPercent(cpuBurstPercent, pod); err != nil {
+			if err = m.updateCPUBurstByPercent(cpuBurstPercent, pod, mainContainerName); err != nil {
 				errList = append(errList, fmt.Errorf("error setting cpu burst for policy %s for pod %s: %v",
 					consts.PodAnnotationCPUEnhancementCPUBurstPolicyStatic, pod.Name, err))
 			}
@@ -123,13 +129,20 @@ func (m *managerImpl) UpdateCPUBurst(conf *config.Configuration, dynamicConfig *
 
 // updateCPUBurstByPercent updates the value of cpu burst for static policy by taking the
 // cpu quota from cgroup and calculating the cpu burst value by taking cpu quota * percent / 100.
-func (m *managerImpl) updateCPUBurstByPercent(percent float64, pod *v1.Pod) error {
+// If the main container name is not empty, we update the cpu burst only for that container,
+// otherwise we update the cpu burst for all containers.
+func (m *managerImpl) updateCPUBurstByPercent(percent float64, pod *v1.Pod, mainContainerName string) error {
 	var errList []error
 	podUID := string(pod.GetUID())
 	podName := pod.Name
 
 	for _, container := range pod.Spec.Containers {
 		containerName := container.Name
+		// skip updating the container if the container is not the main container
+		if mainContainerName != "" && containerName != mainContainerName {
+			continue
+		}
+
 		containerID, err := m.metaServer.GetContainerID(podUID, containerName)
 		if err != nil {
 			general.Errorf("get container id failed, pod: %s, podName: %s, container: %s(%s), err: %v", podUID, podName, containerName, containerID, err)