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)}) } } 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 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) +} 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