From 8c6dd3296a2abede9daf890e0722370fbb1a576f Mon Sep 17 00:00:00 2001 From: Antti Kervinen Date: Wed, 22 Jul 2026 15:05:04 +0300 Subject: [PATCH 1/4] balloons: preferCloseToDevices supports podresourceapi: Enable CPU selection based on locality of individual container's physical devices. podresourceapi: syntax allows specifying which physical device(s) is preferred to be local to container's CPUs. In order to allocate all container's CPUs near container's physical device, the container must be placed into a new balloon. Technically, this changes balloon creation internals: so far CPU selection to a new balloon has been affected by balloon types and instances only, but after this change the first container put into a new balloon may affect it, too. Signed-off-by: Antti Kervinen --- .../balloons/policy/balloons-policy.go | 220 ++++++++++++++++-- cmd/plugins/balloons/policy/cpuclass_test.go | 14 -- cmd/plugins/balloons/policy/cputree.go | 2 + 3 files changed, 208 insertions(+), 28 deletions(-) diff --git a/cmd/plugins/balloons/policy/balloons-policy.go b/cmd/plugins/balloons/policy/balloons-policy.go index 26eafc724..30ea8fe7c 100644 --- a/cmd/plugins/balloons/policy/balloons-policy.go +++ b/cmd/plugins/balloons/policy/balloons-policy.go @@ -16,13 +16,17 @@ package balloons import ( "fmt" + "maps" "math" + "path" "path/filepath" + "slices" "sort" "strconv" "strings" "sync" + "github.com/containers/nri-plugins/pkg/agent/podresapi" cfgapi "github.com/containers/nri-plugins/pkg/apis/config/v1alpha1/resmgr/policy/balloons" "github.com/containers/nri-plugins/pkg/cpuallocator" "github.com/containers/nri-plugins/pkg/kubernetes" @@ -992,7 +996,7 @@ func (p *balloons) virtDevsChangeDuringCpuAllocation(loadClassNames []string) bo return false } -func (p *balloons) newCompositeBalloon(blnDef *BalloonDef, confCpus bool, freeInstance int) (*Balloon, error) { +func (p *balloons) newCompositeBalloon(blnDef *BalloonDef, confCpus bool, freeInstance int, c cache.Container) (*Balloon, error) { componentBlns := make([]*Balloon, 0, len(blnDef.Components)) deleteComponentBlns := func() { for _, compBln := range componentBlns { @@ -1011,7 +1015,7 @@ func (p *balloons) newCompositeBalloon(blnDef *BalloonDef, confCpus bool, freeIn return nil, balloonsError("unknown balloon definition %q in composite balloon %q", comp.DefName, blnDef.Name) } - compBln, err := p.newBalloon(compDef, confCpus) + compBln, err := p.newBalloon(compDef, confCpus, c) if err != nil || compBln == nil { deleteComponentBlns() return nil, balloonsError("failed to create component balloon %q for composite balloon %q: %v", @@ -1049,7 +1053,7 @@ func (p *balloons) newCompositeBalloon(blnDef *BalloonDef, confCpus bool, freeIn return nil, balloonsError("unknown balloon definition %q in balanced composite balloon %q", comp.DefName, blnDef.Name) } - compBln, err := p.newBalloon(compDef, confCpus) + compBln, err := p.newBalloon(compDef, confCpus, c) if err != nil || compBln == nil { deleteComponentBlns() return nil, balloonsError("failed to create component balloon %q for composite balloon %q: %v", @@ -1093,7 +1097,7 @@ func (p *balloons) newCompositeBalloon(blnDef *BalloonDef, confCpus bool, freeIn return bln, nil } -func (p *balloons) newBalloon(blnDef *BalloonDef, confCpus bool) (*Balloon, error) { +func (p *balloons) newBalloon(blnDef *BalloonDef, confCpus bool, c cache.Container) (*Balloon, error) { var cpus cpuset.CPUSet var err error blnsOfDef := p.balloonsByDef(blnDef) @@ -1116,7 +1120,7 @@ func (p *balloons) newBalloon(blnDef *BalloonDef, confCpus bool) (*Balloon, erro } } if len(blnDef.Components) > 0 { - return p.newCompositeBalloon(blnDef, confCpus, freeInstance) + return p.newCompositeBalloon(blnDef, confCpus, freeInstance, c) } // Configure cpuTreeAllocator for this balloon. The reserved // balloon always prefers to be close to the virtual device @@ -1134,6 +1138,13 @@ func (p *balloons) newBalloon(blnDef *BalloonDef, confCpus bool) (*Balloon, erro virtDevPCores: {p.cpuAllocator.GetCPUPriorities()[cpuallocator.PriorityHigh]}, }, } + // Pod resource hints to container's physical devices (GPUs, + // NICs, etc.) do not have acceptable alternative CPU + // sets. Therefore, apply these hints before CPU class hints. + // The latter may include alternative sets, which enables the + // allocator to choose the best alternative CPU set for the + // pod resources. + p.applyPodResourcesHints(&allocatorOptions, c) p.applyCpuClassHints(&allocatorOptions, p.resolveCpuClassName(blnDef.CpuClass), cpuset.New(), 0) if blnDef.AllocatorTopologyBalancing != nil { allocatorOptions.topologyBalancing = *blnDef.AllocatorTopologyBalancing @@ -1225,7 +1236,7 @@ func (p *balloons) fillableBalloonInstances(blnDef *BalloonDef, fm FillMethod, c } return nil, nil } - newBln, err := p.newBalloon(blnDef, false) + newBln, err := p.newBalloon(blnDef, false, c) if err != nil { if fm == FillNewBalloonMust { return nil, err @@ -1590,7 +1601,7 @@ func (p *balloons) Reconfigure(newCfg interface{}) error { // balloons according to the blnDef. Does not initialize balloon CPUs. func (p *balloons) applyBalloonDef(balloons *[]*Balloon, blnDef *BalloonDef, freeCpus *cpuset.CPUSet) error { for blnIdx := 0; blnIdx < blnDef.MinBalloons; blnIdx++ { - newBln, err := p.newBalloon(blnDef, false) + newBln, err := p.newBalloon(blnDef, false, nil) if err != nil { return err } @@ -1609,6 +1620,7 @@ func (p *balloons) validateConfig(bpoptions *BalloonsOptions) error { undefinedSchedulingClasses := map[string]struct{}{} compositeBlnDefs := map[string]*BalloonDef{} for _, blnDef := range bpoptions.BalloonDefs { + podResourceDevices := map[string]struct{}{} if blnDef.Name == "" { return balloonsError("missing or empty name in a balloon type") } @@ -1679,6 +1691,18 @@ func (p *balloons) validateConfig(bpoptions *BalloonsOptions) error { for _, load := range blnDef.Loads { undefinedLoadClasses[load] = struct{}{} } + for _, dev := range blnDef.PreferCloseToDevices { + if resourceName, ok := podResourceDeviceName(dev); ok { + podResourceDevices[resourceName] = struct{}{} + } + } + if len(podResourceDevices) > 0 { + if !blnDef.PreferNewBalloons || !blnDef.PreferSpreadingPods { + devs := slices.Sorted(maps.Keys(podResourceDevices)) + log.Warnf("balloon type %q prefers affinity container's devices %v, but the same balloon may contain many containers because PreferNewBalloons=%v (suggest: true) and PreferSpreadingPods=%v (suggest: true) to prefer single container per balloon", blnDef.Name, devs, blnDef.PreferNewBalloons, blnDef.PreferSpreadingPods) + } + } + if blnDef.SchedulingClass != "" { undefinedSchedulingClasses[blnDef.SchedulingClass] = struct{}{} } @@ -2141,8 +2165,8 @@ func mergeCpuClassHints(opts *cpuTreeAllocatorOptions, provider cpuClassHints, i if opts.virtDevCpusets == nil { opts.virtDevCpusets = map[string][]cpuset.CPUSet{} } - opts.preferCloseToDevices = filterOutHintDevs(opts.preferCloseToDevices) - opts.preferFarFromDevices = filterOutHintDevs(opts.preferFarFromDevices) + opts.preferCloseToDevices = filterOutPrefixDevs(opts.preferCloseToDevices, cpuClassHintDevPrefix) + opts.preferFarFromDevices = filterOutPrefixDevs(opts.preferFarFromDevices, cpuClassHintDevPrefix) for name := range opts.virtDevCpusets { if strings.HasPrefix(name, cpuClassHintDevPrefix) { delete(opts.virtDevCpusets, name) @@ -2163,19 +2187,177 @@ func mergeCpuClassHints(opts *cpuTreeAllocatorOptions, provider cpuClassHints, i } } -// filterOutHintDevs returns devs with all cpuClass hint device names -// (those carrying cpuClassHintDevPrefix) removed. The returned slice -// reuses devs' backing array. -func filterOutHintDevs(devs []string) []string { +// podResourceHintDevPrefix is the prefix of synthetic virtual device +// names that carry per-container pod-resource device locality hints. +const podResourceHintDevPrefix = "__podres_" + +// podResourceDeviceName returns the device-plugin extended resource +// name and true if dev refers to a pod-resource device, that is if it +// is of the form "podresourceapi:". Otherwise it +// returns "", false. +func podResourceDeviceName(dev string) (string, bool) { + if podresapi.IsPodResourceHint(dev) { + return strings.TrimPrefix(dev, podresapi.HintProvider), true + } + return "", false +} + +// podResourceHintDevName returns the synthetic virtual device name +// for the pod-resource device locality hint of resourceName assigned +// to the container. If resourceName is empty, returned name is a +// common prefix for all synthetic device names associated with the +// given (and only the given) container. +func podResourceHintDevName(c cache.Container, resourceName string) string { + ctrID := "unknown" + if c != nil { + ctrID = c.GetID() + } + return podResourceHintDevPrefix + ctrID + "_" + resourceName +} + +// filterOutPrefixDevs returns devs with all entries carrying prefix +// removed. The returned slice reuses devs' backing array. +func filterOutPrefixDevs(devs []string, prefix string) []string { out := devs[:0] for _, d := range devs { - if !strings.HasPrefix(d, cpuClassHintDevPrefix) { + if !strings.HasPrefix(d, prefix) { out = append(out, d) } } return out } +// containerDeviceCpus returns the set of CPUs that are topologically +// close to the device instance(s) of resourceName that the kubelet +// device manager assigned to container c. The second return value is +// false if the pod resources are not available, if c has no such +// device assigned, or if the assigned devices carry no NUMA topology +// information. +func (p *balloons) containerDeviceCpus(c cache.Container, resourceName string) (cpuset.CPUSet, bool) { + if c == nil { + return emptyCpuSet, false + } + ctrRes := c.GetPodResources() + if ctrRes == nil { + return emptyCpuSet, false + } + numas := map[int64]struct{}{} + for _, dev := range ctrRes.GetDevices() { + // glob match resourceName (wildcards allowed) against + // the device name reported by the pod resources API + if matched, err := path.Match(resourceName, dev.GetResourceName()); err != nil || !matched { + if err != nil { + log.Errorf("invalid resource glob pattern %q: %v", resourceName, err) + } + continue + } + devNumas := make([]int64, 0) + for _, node := range dev.GetTopology().GetNodes() { + numas[node.GetID()] = struct{}{} + devNumas = append(devNumas, node.GetID()) + } + log.Debugf("pod-resource device %q of container %s matches pattern %q, NUMA nodes %v, device IDs %v", + dev.GetResourceName(), c.PrettyName(), resourceName, devNumas, dev.GetDeviceIds()) + } + if len(numas) == 0 { + return emptyCpuSet, false + } + numaIDs := make([]string, 0, len(numas)) + for id := range numas { + numaIDs = append(numaIDs, strconv.FormatInt(id, 10)) + } + cpuStr := p.options.System.NodeHintToCPUs(strings.Join(numaIDs, ",")) + if cpuStr == "" { + return emptyCpuSet, false + } + cpus, err := cpuset.Parse(cpuStr) + if err != nil { + log.Errorf("failed to parse CPUs %q of device %q for container %s: %v", + cpuStr, resourceName, c.PrettyName(), err) + return emptyCpuSet, false + } + return cpus.Intersection(p.allowed), true +} + +// applyPodResourcesHints resolves every "podresourceapi:" +// entry in opts to the CPUs close to the device(s) that the kubelet +// assigned to container c, and injects the result as synthetic virtual +// devices. It must be called before creating the CPU tree allocator so +// that the resolved CPUs are visible to it. Entries are resolved in +// place, preserving their original priority order among other device +// hints. Unresolvable entries (no container context, missing pod +// resources, no assigned device, or no topology info) are dropped. +func (p *balloons) applyPodResourcesHints(opts *cpuTreeAllocatorOptions, c cache.Container) { + if opts == nil { + return + } + if opts.virtDevCpusets == nil { + opts.virtDevCpusets = map[string][]cpuset.CPUSet{} + } + opts.preferCloseToDevices = p.resolvePodResourceDevs(opts.preferCloseToDevices, opts.virtDevCpusets, c) +} + +// resolvePodResourceDevs replaces every pod-resource device entry in +// devs with a synthetic per-container virtual device, registering the +// resolved CPUs in virtDevCpusets. Non-pod-resource entries are kept +// unchanged in their original position. Unresolvable pod-resource +// entries are dropped. +func (p *balloons) resolvePodResourceDevs(devs []string, virtDevCpusets map[string][]cpuset.CPUSet, c cache.Container) []string { + out := make([]string, 0, len(devs)) + for _, dev := range devs { + resourceName, isPodRes := podResourceDeviceName(dev) + if !isPodRes { + out = append(out, dev) + continue + } + if c == nil { + log.Debugf("dropping pod-resource device %q hint: no container context", dev) + continue + } + cpus, ok := p.containerDeviceCpus(c, resourceName) + if !ok { + log.Debugf("dropping pod-resource device %q hint: no device locality for container %s", + dev, c.PrettyName()) + continue + } + name := podResourceHintDevName(c, resourceName) + virtDevCpusets[name] = []cpuset.CPUSet{cpus} + out = append(out, name) + log.Debugf("pod-resource hint: device %q of container %s prefers CPUs %s", + resourceName, c.PrettyName(), cpus) + } + return out +} + +// removePodResourcesHints removes all pod-resource device locality +// hints that originate from container c from the allocator options of +// balloon bln and its component balloons. +func (p *balloons) removePodResourcesHints(bln *Balloon, c cache.Container) { + if bln == nil || c == nil { + return + } + prefix := podResourceHintDevName(c, "") + blns := append([]*Balloon{bln}, bln.components...) + for _, b := range blns { + if b.cpuTreeAlloc == nil { + continue + } + opts := &b.cpuTreeAlloc.options + removed := false + for name := range opts.virtDevCpusets { + if strings.HasPrefix(name, prefix) { + delete(opts.virtDevCpusets, name) + removed = true + } + } + if removed { + opts.preferCloseToDevices = filterOutPrefixDevs(opts.preferCloseToDevices, prefix) + log.Debugf("removed pod-resource device hints of container %s from balloon %s", + c.PrettyName(), b.PrettyName()) + } + } +} + // fillFarFromDevices adds BalloonDefs implicit device anti-affinities // towards devices that other BalloonDefs prefer to be close to. func (p *balloons) fillFarFromDevices(blnDefs []*BalloonDef) { @@ -2193,6 +2375,13 @@ func (p *balloons) fillFarFromDevices(blnDefs []*BalloonDef) { } for _, blnDef := range blnDefs { for _, closeDev := range blnDef.PreferCloseToDevices { + if podresapi.IsPodResourceHint(closeDev) { + // A logical pod resources can match multiple + // physical devices. Avoiding a logical resource + // would mean avoiding the physical device that + // the container actually got assigned. + continue + } if _, ok := devDefClose[closeDev]; !ok { avoidDevs = append(avoidDevs, closeDev) devDefClose[closeDev] = map[string]bool{} @@ -2549,6 +2738,9 @@ func (p *balloons) dismissContainer(c cache.Container, bln *Balloon) { delete(bln.PodIDs, podID) } bln.updateGroups(c, -1) + // Drop pod-resource device locality hints of this container + // so that they no longer bias resizing of the balloon. + p.removePodResourcesHints(bln, c) } // pinCpuMem pins container to CPUs and memory nodes if flagged diff --git a/cmd/plugins/balloons/policy/cpuclass_test.go b/cmd/plugins/balloons/policy/cpuclass_test.go index c2aa63345..06da2a988 100644 --- a/cmd/plugins/balloons/policy/cpuclass_test.go +++ b/cmd/plugins/balloons/policy/cpuclass_test.go @@ -142,20 +142,6 @@ func TestMergeCpuClassHintsNoAccumulation(t *testing.T) { } } -func TestFilterOutHintDevs(t *testing.T) { - in := []string{"a", "__cls_pref_0_x", "b", "__cls_avoid_0_y", "c"} - got := filterOutHintDevs(in) - want := []string{"a", "b", "c"} - if len(got) != len(want) { - t.Fatalf("filterOutHintDevs len=%d, want %d: got=%v", len(got), len(want), got) - } - for i := range want { - if got[i] != want[i] { - t.Errorf("filterOutHintDevs[%d] = %q, want %q", i, got[i], want[i]) - } - } -} - func userDevs(devs []string) []string { out := []string{} for _, d := range devs { diff --git a/cmd/plugins/balloons/policy/cputree.go b/cmd/plugins/balloons/policy/cputree.go index 266cec1ba..ad8d9722c 100644 --- a/cmd/plugins/balloons/policy/cputree.go +++ b/cmd/plugins/balloons/policy/cputree.go @@ -665,11 +665,13 @@ func (ta *cpuTreeAllocator) resizeCpusWithDevices(resizers []cpuResizerFunc, cur allCloseCpuSets := [][]cpuset.CPUSet{} for _, devPath := range ta.options.preferCloseToDevices { if closeCpuSets := ta.topologyHintCpus(devPath); len(closeCpuSets) > 0 { + log.Debugf(" - prepare: close to %q, prefer cpusets: %v", devPath, closeCpuSets) allCloseCpuSets = append(allCloseCpuSets, closeCpuSets) } } for _, devPath := range ta.options.preferFarFromDevices { for _, farCpuSet := range ta.topologyHintCpus(devPath) { + log.Debugf(" - prepare: far from %q, prefer cpusets: %s", devPath, freeCpus.Difference(farCpuSet)) allCloseCpuSets = append(allCloseCpuSets, []cpuset.CPUSet{freeCpus.Difference(farCpuSet)}) } } From 8e0e9e62255b50057d94c1050cc5cb702cf6a445 Mon Sep 17 00:00:00 2001 From: Antti Kervinen Date: Wed, 22 Jul 2026 15:30:01 +0300 Subject: [PATCH 2/4] scripts/testing: add a fake device plugin with device locality Signed-off-by: Antti Kervinen --- .../fake-device-plugin/fake-device-plugin.go | 220 ++++++++++++++++++ 1 file changed, 220 insertions(+) create mode 100644 scripts/testing/fake-device-plugin/fake-device-plugin.go diff --git a/scripts/testing/fake-device-plugin/fake-device-plugin.go b/scripts/testing/fake-device-plugin/fake-device-plugin.go new file mode 100644 index 000000000..38d976d94 --- /dev/null +++ b/scripts/testing/fake-device-plugin/fake-device-plugin.go @@ -0,0 +1,220 @@ +package main + +import ( + "context" + "fmt" + "log" + "net" + "os" + "os/signal" + "path/filepath" + "strings" + + "flag" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "sigs.k8s.io/yaml" + + pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" +) + +const ( + pluginDir = "/var/lib/kubelet/device-plugins" +) + +type Config struct { + ResourceName string `yaml:"resourceName"` + Devices []DeviceConfig `yaml:"devices"` +} + +type DeviceConfig struct { + ID string `yaml:"id"` + NUMANodes []int64 `yaml:"numaNodes,omitempty"` +} + +type FakeDevicePlugin struct { + pluginapi.UnimplementedDevicePluginServer + + resourceName string + devices []*pluginapi.Device + stop chan any +} + +func loadConfig(path string) (*Config, error) { + b, err := os.ReadFile(path) + if err != nil { + return nil, err + } + + var cfg Config + if err := yaml.Unmarshal(b, &cfg); err != nil { + return nil, err + } + + return &cfg, nil +} + +func newPlugin(cfg *Config) *FakeDevicePlugin { + p := &FakeDevicePlugin{ + resourceName: cfg.ResourceName, + stop: make(chan interface{}), + } + + for _, d := range cfg.Devices { + dev := &pluginapi.Device{ + ID: d.ID, + Health: pluginapi.Healthy, + } + if len(d.NUMANodes) > 0 { + numaNodes := []*pluginapi.NUMANode{} + for _, node := range d.NUMANodes { + numaNodes = append(numaNodes, &pluginapi.NUMANode{ID: node}) + } + dev.Topology = &pluginapi.TopologyInfo{ + Nodes: numaNodes, + } + } + p.devices = append(p.devices, dev) + } + log.Printf("newPlugin: %v[%v]", p.resourceName, p.devices) + + return p +} + +func (p *FakeDevicePlugin) GetDevicePluginOptions( + context.Context, + *pluginapi.Empty, +) (*pluginapi.DevicePluginOptions, error) { + + return &pluginapi.DevicePluginOptions{}, nil +} + +func (p *FakeDevicePlugin) ListAndWatch( + _ *pluginapi.Empty, + stream pluginapi.DevicePlugin_ListAndWatchServer, +) error { + + log.Printf("ListAndWatch: stream output %v", p.devices) + + _ = stream.Send(&pluginapi.ListAndWatchResponse{ + Devices: p.devices, + }) + + <-p.stop + return nil +} + +func (p *FakeDevicePlugin) Allocate( + ctx context.Context, + req *pluginapi.AllocateRequest, +) (*pluginapi.AllocateResponse, error) { + + log.Printf("Allocate input: %s", req.String()) + resp := &pluginapi.AllocateResponse{} + + for range req.ContainerRequests { + resp.ContainerResponses = append( + resp.ContainerResponses, + &pluginapi.ContainerAllocateResponse{}, + ) + } + log.Printf("Allocate output: %s", resp.String()) + + return resp, nil +} + +func (p *FakeDevicePlugin) PreStartContainer( + _ context.Context, + req *pluginapi.PreStartContainerRequest, +) (*pluginapi.PreStartContainerResponse, error) { + + log.Printf("PreStartContainer input: %s", req.String()) + + return &pluginapi.PreStartContainerResponse{}, nil +} + +func register(socket, resource string) error { + + conn, err := grpc.NewClient( + "unix://"+pluginapi.KubeletSocket, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + return err + } + defer func() { + _ = conn.Close() + }() + + client := pluginapi.NewRegistrationClient(conn) + + _, err = client.Register( + context.Background(), + &pluginapi.RegisterRequest{ + Version: pluginapi.Version, + Endpoint: socket, + ResourceName: resource, + }, + ) + + return err +} + +func main() { + log.SetPrefix("fake-device-plugin<> ") + configPath := flag.String("config", "fake-device-plugin.yaml", "Path to device plugin config file") + socketPath := flag.String("socket", "", "Unix socket path (optional, defaults to /fake-device-plugin-.sock)") + flag.Parse() + + cfg, err := loadConfig(*configPath) + if err != nil { + log.Fatal(err) + } + + log.SetPrefix(fmt.Sprintf("fake-device-plugin<%s> ", cfg.ResourceName)) + + plugin := newPlugin(cfg) + + var socket string + var socketPathAbs string + if *socketPath != "" { + socket = filepath.Base(*socketPath) + socketPathAbs = *socketPath + } else { + socket = fmt.Sprintf("fake-device-plugin-%s.sock", strings.ReplaceAll(cfg.ResourceName, "/", "-")) + socketPathAbs = filepath.Join(pluginDir, socket) + } + + _ = os.Remove(socketPathAbs) + + lis, err := net.Listen("unix", socketPathAbs) + if err != nil { + log.Fatal(err) + } + + grpcServer := grpc.NewServer() + + pluginapi.RegisterDevicePluginServer(grpcServer, plugin) + + go func() { + if err := grpcServer.Serve(lis); err != nil { + log.Printf("grpcServer.Serve error: %v", err) + } + }() + + if err := register(socket, cfg.ResourceName); err != nil { + log.Fatal(err) + } + + fmt.Printf("registered on %s\n", socketPathAbs) + + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, os.Interrupt) + <-sigChan + + fmt.Println("shutting down...") + grpcServer.Stop() + _ = os.Remove(socketPathAbs) +} From 0f8cb16583dbd2991098477252956043bb8b305c Mon Sep 17 00:00:00 2001 From: Antti Kervinen Date: Wed, 22 Jul 2026 15:29:14 +0300 Subject: [PATCH 3/4] e2e: add balloons preferCloseToDevices/podresourceapi tests Signed-off-by: Antti Kervinen --- .../balloons-podresources.cfg | 50 +++++ .../n4c16/test24-podresources/code.var.sh | 175 ++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/balloons-podresources.cfg create mode 100644 test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/code.var.sh diff --git a/test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/balloons-podresources.cfg b/test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/balloons-podresources.cfg new file mode 100644 index 000000000..c0bb27d0f --- /dev/null +++ b/test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/balloons-podresources.cfg @@ -0,0 +1,50 @@ +config: + agent: + nodeResourceTopology: true + podResourceAPI: true + allocatorTopologyBalancing: false + availableResources: + cpu: cpuset:0-7,9-14 + reservedResources: + cpu: cpuset:0 + + pinCPU: true + pinMemory: false + + balloonTypes: + - name: hp-near-tpu + cpuClass: pct-hp + preferCloseToDevices: + - podresourceapi:tech.com/* + preferNewBalloons: true + preferSpreadingPods: true + + - name: near-nic + preferCloseToDevices: + - podresourceapi:*.com/nic + preferNewBalloons: true + preferSpreadingPods: true + shareIdleCPUsInSame: package + + cpuClasses: + - name: pct-hp + minFreq: "turbo" + maxFreq: "turbo" + pctPriority: high + publishExtendedResource: true + + - name: default + minFreq: "min" + maxFreq: "base" + pctPriority: low + + log: + debug: + - policy + - nri-plugin + - cpu + - cpuclass +extraEnv: + OVERRIDE_SYS_CPUFREQ: '''[{"cpus": "0-15", "base": 2900000, "min": 800000, "max": 3800000}]''' + OVERRIDE_SST: '''{"supported": true, "clos_count": 4, "packages": [{"id": 0, "cpus": "0-7", "tf_supported": true, "cp_supported": true, "max_hp_cpus": 4}, {"id": 1, "cpus": "8-15", "tf_supported": true, "cp_supported": true, "max_hp_cpus": 4}]}''' + OVERRIDE_SST_STATE_DIR: "/tmp/nri-pct-mock" diff --git a/test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/code.var.sh b/test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/code.var.sh new file mode 100644 index 000000000..6d8933cef --- /dev/null +++ b/test/e2e/policies.test-suite/balloons/n4c16/test24-podresources/code.var.sh @@ -0,0 +1,175 @@ +# Test CPU affinity to devices published by device plugins and queried +# from kubelet's PodResourcesAPI. + +helm-terminate +helm_config=$TEST_DIR/balloons-podresources.cfg helm-launch balloons + +cleanup() { + vm-command 'pidof fake-device-plugin && kill $(pidof fake-device-plugin) && sleep 1' + vm-command "kubectl delete pods --all --now" || true +} + +# verify-podres-locality RESOURCE CONTAINER... +# +# Reads the balloons policy debug log to find, for each CONTAINER, the +# NUMA node(s) of the device instance matching RESOURCE that the +# kubelet device manager assigned to it (the log line is emitted by +# containerDeviceCpus when it resolves a "podresourceapi:" hint). It +# then verifies that all CPUs the container is allowed to run on belong +# to those NUMA node(s), that is, that the balloon's CPUs really landed +# close to the assigned device. +verify-podres-locality() { + local resource=$1 + shift + vm-command "kubectl -n kube-system logs ds/nri-resource-policy-balloons | grep 'pod-resource device \"$resource\"'" >/dev/null \ + || command-error "no device-locality log lines for resource $resource" + local log="$COMMAND_OUTPUT" + local ctr + for ctr in "$@"; do + # The pod resources API reports a device that spans several + # NUMA nodes as one entry per node, so collect the union of the + # NUMA nodes over all log lines mentioning this container. + local lines + lines=$(grep "/$ctr matches" <<< "$log") + [ -n "$lines" ] \ + || command-error "no device-locality log line for container $ctr (resource $resource)" + local numas + numas=$(sed -n 's/.*NUMA nodes \[\([0-9 ]*\)\].*/\1/p' <<< "$lines" | tr ' ' '\n' | grep -E '^[0-9]+$' | sort -un | tr '\n' ' ') + [ -n "$numas" ] \ + || command-error "could not parse NUMA nodes for container $ctr from: $lines" + local nodeset="" n + for n in $numas; do + nodeset="$nodeset\"node$n\"," + done + out "### container $ctr got $resource on NUMA node(s): $numas" + verify "len(nodes[\"$ctr\"]) > 0" \ + "nodes[\"$ctr\"].issubset({$nodeset})" + done +} + +# Install and (re)start fake-device-plugins +cleanup +vm-command "command -v fake-device-plugin" || { + HOST_DEVICE_PLUGIN=$OUTPUT_DIR/fake-device-plugin + + [ -f "$HOST_DEVICE_PLUGIN" ] || \ + GOARCH=amd64 go build -o "$HOST_DEVICE_PLUGIN" "${TEST_DIR%%/test/e2e/*}/scripts/testing/fake-device-plugin/fake-device-plugin.go" || \ + error "failed to build $HOST_DEVICE_PLUGIN" + + vm-put-file "$HOST_DEVICE_PLUGIN" "/usr/local/bin/$(basename "$HOST_DEVICE_PLUGIN")" +} + +vm-command "cat > fake-tpu.yaml <& fake-tpu.output &" + +vm-command "cat > fake-nic.yaml <& fake-nic.output &" +sleep 1 + + +# Wait until both fake device plugins have registered and their +# devices are Allocatable. +rounds=0 +while vm-command "kubectl describe node \$(hostname) | grep -E 'Capacity|Alloc|telco.com|tech.com'"; do + ( grep -A2 Allocatable <<< "$COMMAND_OUTPUT" | grep -qE 'tech.com/tpu:.*4' ) && \ + ( grep -A2 Allocatable <<< "$COMMAND_OUTPUT" | grep -qE 'telco.com/nic:.*2' ) && \ + break + rounds=$(( rounds + 1 )) + (( rounds > 60 )) && error "waiting for fake-device-plugin resources timed out" + sleep 1 +done + +# burstable containers +CPUREQ=2 CPULIM=4 MEMREQ=10M MEMLIM=50M \ + EXTREQ="telco.com/nic: \"1\"" \ + EXTLIM="telco.com/nic: \"1\"" \ + POD_ANNOTATION="balloon.balloons.resource-policy.nri.io: near-nic" \ + CONTCOUNT=2 \ + create balloons-busybox +report allowed + +# The two NIC-consuming containers matched "podresourceapi:*.com/nic". +# Each got a different telco.com/nic device, one on NUMA nodes {0,1} +# (package 0) and the other on NUMA nodes {2,3} (package 1). Verify that +# each container's CPUs landed on the NUMA node(s) of its assigned NIC, +# and that the two containers ended up on disjoint (different-package) +# node sets. +verify-podres-locality "telco.com/nic" pod0c0 pod0c1 +verify 'disjoint_sets(nodes["pod0c0"], nodes["pod0c1"])' + +# # Free the NIC balloons before the next phase so that the TPU pods +# # below can be placed on the NUMA nodes of their own devices without +# # competing for CPUs with the still-running NIC containers. Without +# # this, node2's only usable CPUs (10-11; cpu8 is off, cpu9 reserved) +# # are held by pod0c1's NIC balloon, leaving no room for the node2 TPU. +# vm-command "kubectl delete pod pod0 --now" +# vm-command "kubectl wait --for=delete pod/pod0 --timeout=30s" || true + +declare -a EXTREQ=( "tech.com/tpu: \"1\"" "cpuclass.balloons.nri.io/pct-hp: \"1\"" ) +declare -a EXTLIM=( "tech.com/tpu: \"1\"" "cpuclass.balloons.nri.io/pct-hp: \"1\"" ) +CPUREQ=1 CPULIM=1 MEMREQ=10M MEMLIM=10M \ + POD_ANNOTATION="balloon.balloons.resource-policy.nri.io: hp-near-tpu" \ + CONTCOUNT=4 \ + create balloons-busybox +report allowed + +# The four TPU-consuming containers matched "podresourceapi:tech.com/*". +# There are four tech.com/tpu devices, one on each of NUMA nodes 0..3, +# so the four containers get four distinct devices. Verify that each +# HP balloon's single CPU landed on the NUMA node of its assigned TPU, +# and that all four ended up on distinct NUMA nodes. +verify-podres-locality "tech.com/tpu" pod1c0 pod1c1 pod1c2 pod1c3 +verify 'disjoint_sets(nodes["pod1c0"], nodes["pod1c1"], nodes["pod1c2"], nodes["pod1c3"])' + +# Sanity check (cf. test19-pct): the hp-near-tpu balloons use the +# pct-hp cpuClass, so their CPUs must have been associated to the PCT +# high-priority CLOS 0. +vm-command "kubectl -n kube-system logs ds/nri-resource-policy-balloons | grep -E 'associated cpus .* to CLOS 0'" \ + || command-error "hp-near-tpu balloon CPUs were not associated to PCT HP CLOS 0" + +vm-command "kubectl delete pods --all --now" + +# Create pods that use both a tpu and a nic. Align pod2 near nic, pod3 near tpu. +declare -a EXTREQ=( "tech.com/tpu: \"1\"" "telco.com/nic: \"1\"" "cpuclass.balloons.nri.io/pct-hp: \"2\"" ) +declare -a EXTLIM=( "tech.com/tpu: \"1\"" "telco.com/nic: \"1\"" "cpuclass.balloons.nri.io/pct-hp: \"2\"" ) +CPUREQ=2 CPULIM=2 MEMREQ=10M MEMLIM=10M \ + POD_ANNOTATION="balloon.balloons.resource-policy.nri.io: hp-near-tpu" \ + CONTCOUNT=1 \ + create balloons-busybox +report allowed +verify-podres-locality "tech.com/tpu" pod2c0 + +declare -a EXTREQ=( "tech.com/tpu: \"1\"" "telco.com/nic: \"1\"" ) +declare -a EXTREQ=( "tech.com/tpu: \"1\"" "telco.com/nic: \"1\"" ) +CPUREQ=1 CPULIM=1 MEMREQ=10M MEMLIM=10M \ + POD_ANNOTATION="balloon.balloons.resource-policy.nri.io: near-nic" \ + CONTCOUNT=1 \ + create balloons-busybox +report allowed +verify-podres-locality "telco.com/nic" pod3c0 + +cleanup +vm-command "kubectl delete pods --all --now" +helm-terminate From ed093c0ca18fdfb8da329b85f525189d64891d04 Mon Sep 17 00:00:00 2001 From: Antti Kervinen Date: Wed, 22 Jul 2026 14:25:00 +0300 Subject: [PATCH 4/4] doc: add preferCloseToDevices/podresourceapi to balloons docs Signed-off-by: Antti Kervinen --- docs/resource-policy/policy/balloons.md | 44 ++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/docs/resource-policy/policy/balloons.md b/docs/resource-policy/policy/balloons.md index 1da518331..9618f517f 100644 --- a/docs/resource-policy/policy/balloons.md +++ b/docs/resource-policy/policy/balloons.md @@ -477,6 +477,17 @@ balloonTypes: - `/sys/class/drm/card0` - `/sys/devices/system/cpu/cpu14/cache/index2` - `/sys/devices/system/node/node0` +- Device-plugin devices assigned to a container via the kubelet pod + resources API, referenced by their extended resource name with the + `podresourceapi:` prefix, like + - `podresourceapi:nvidia.com/gpu` + - `podresourceapi:intel.com/sgx` + + The resource name may contain shell-style wildcards (`*` matches any + sequence of characters, `?` matches a single character) to match + several resources at once, like + - `podresourceapi:nvidia.com/*` + - `podresourceapi:*/gpu` - First device in list has highest priority. - Automatically adds anti-affinity between listed devices and other balloon types. @@ -487,7 +498,38 @@ balloonTypes: - /sys/class/drm/card0 - name: network-io preferCloseToDevices: - - /sys/class/net/eth0 + - podresourceapi:*.com/nic +``` + +Unlike sysfs device paths, whose CPU locality is fixed, a +`podresourceapi:` entry is resolved per container: when +a container that got such a device triggers creation of a new balloon +(`preferNewBalloons: true`), the CPUs of the new balloon are selected +close to the NUMA node(s) of the exact device instance that the +kubelet device manager assigned to that container. This enables, for +example, allocating CPUs from the socket that is closest to the +specific PCI device a container was given, when similar physical +devices providing the same logical resource exist on multiple +sockets. When the resource name contains wildcards, the CPUs are +selected close to the NUMA node(s) of all assigned device instances +whose resource name matches the pattern. + +Using `podresourceapi:` devices requires the pod resources API to be +enabled with the agent option `podResourceAPI: true`. The balloon type +should have `preferNewBalloons: true`. Otherwise the container might +be assigned to an existing balloon instance whose CPUs have been +selected based on affinity to another physical device in the system +because another container got that device instead. + +```yaml +config: + agent: + podResourceAPI: true + balloonTypes: + - name: gpu-containers + preferNewBalloons: true + preferCloseToDevices: + - podresourceapi:nvidia.com/gpu ``` #### Dynamic CPU Preferences