diff --git a/deployment/helm/topology-aware/templates/deviceclass.yaml b/deployment/helm/topology-aware/templates/deviceclass.yaml new file mode 100644 index 000000000..2c75d7791 --- /dev/null +++ b/deployment/helm/topology-aware/templates/deviceclass.yaml @@ -0,0 +1,12 @@ +{{- if .Values.config.dra.enabled }} +apiVersion: resource.k8s.io/v1 +kind: DeviceClass +metadata: + name: topology-aware.nri.io + labels: + {{- include "nri-plugin.labels" . | nindent 4 }} +spec: + selectors: + - cel: + expression: device.driver == "topology-aware.nri.io" +{{- end }} diff --git a/docs/resource-policy/policy/topology-aware.md b/docs/resource-policy/policy/topology-aware.md index 8ece885dd..711c180ba 100644 --- a/docs/resource-policy/policy/topology-aware.md +++ b/docs/resource-policy/policy/topology-aware.md @@ -122,6 +122,8 @@ The `topology-aware` policy has the following features: - Support for DRAM, PMEM (such as Intel® Optane™ memory), and [HBM](https://en.wikipedia.org/wiki/High_Bandwidth_Memory) - **Cold start support** - Pin workload exclusively to PMEM for an initial warm-up period +- **Dynamic Resource Allocation (DRA)** + - Publish NUMA nodes as DRA devices and allocate CPUs to claims ### Organization of this document @@ -129,6 +131,7 @@ This document is organized as follows: - **Installation and configuration** describes installation and basic configuration - **Configuration options** covers detailed configuration options +- **Dynamic Resource Allocation** describes CPU allocation to DRA claims - **Cookbook** provides recipes for common use cases - **Troubleshooting** offers troubleshooting guidance @@ -1421,6 +1424,182 @@ metadata: [configuration]: ../configuration.md +## Dynamic Resource Allocation (DRA) + +The topology-aware policy can allocate CPUs to Kubernetes +[Dynamic Resource Allocation](https://kubernetes.io/docs/concepts/scheduling-eviction/dynamic-resource-allocation/) +(DRA) claims. With DRA enabled (`config.dra.enabled: true` in the Helm +chart), the plugin registers the DRA driver `topology-aware.nri.io`, and the +chart adds a `DeviceClass` of the same name. A pod claims CPUs of a NUMA node +with a `ResourceClaim`. The scheduler picks a node with enough capacity, and +the policy runs the containers of the pod on the claimed CPUs. + +### Requirements + +- Kubernetes 1.34 or later, with the `DRAConsumableCapacity` feature gate + enabled. The gate is on by default from Kubernetes 1.36. It lets several + claims share a device, each consuming a number of CPUs. Without it, no claim + can be prepared. +- The `DRANodeAllocatableResources` feature gate (alpha) is optional. With it, + the scheduler counts claimed CPUs against the `cpu` resource of the node, so + claims and CPU requests together cannot overcommit the node. Without it, + allocation works, but the scheduler does not count claimed CPUs against the + `cpu` of the node. +- The `DRAListTypeAttributes` feature gate (alpha) is optional. The plugin + checks at start whether the API server accepts list attributes, and + publishes the `pcieRoot` attribute only then. + +### Devices + +The policy publishes one device per NUMA node, named `cpudevnuma000`, +`cpudevnuma001` and so on. The capacity `dra.cpu/cpu` of a device is the +number of CPUs of the node, excluding the reserved CPUs. The capacity does not +change when containers or claims use the CPUs. The devices have these +attributes: + +| Attribute | Type | Value | +|---|---|---| +| `resource.kubernetes.io/numaNode` | int | the ID of the NUMA node | +| `dra.cpu/socketID` | int | the ID of the socket the node belongs to | +| `dra.cpu/smtEnabled` | bool | whether the CPUs have hyperthreading enabled | +| `resource.kubernetes.io/pcieRoot` | list of strings | the PCIe roots of the socket, with `DRAListTypeAttributes` only | + +A request consumes a number of whole CPUs of one device. A request without an +amount gets one CPU. The policy picks the CPUs by topology: two CPUs are the +two threads of one core when possible. + +### Claims + +A claim selects a NUMA node with its `numaNode` attribute, or leaves the +choice to the scheduler. This claim asks for three CPUs of NUMA node 1: + +```yaml +apiVersion: resource.k8s.io/v1 +kind: ResourceClaimTemplate +metadata: + name: three-cpus +spec: + spec: + devices: + requests: + - name: cpus + exactly: + deviceClassName: topology-aware.nri.io + selectors: + - cel: + expression: device.attributes["resource.kubernetes.io"].numaNode == 1 + capacity: + requests: + dra.cpu/cpu: "3" +``` + +A request gets CPUs of one NUMA node. A claim which needs CPUs of several +nodes has one request per node. The constraints of a claim work as for any +DRA device, for instance `matchAttribute: dra.cpu/socketID` keeps the +requests of a claim in one socket. + +The policy allocates the CPUs of a claim when the kubelet prepares it, from +the free CPUs of the node. If the node has too few free CPUs, preparing fails, +and the pod events say why. The claimed CPUs stay allocated until the claim is +released, and no other container runs on them. + +### Containers + +A container which holds a claim runs on the CPUs of the claim. A container +with a CPU request of its own runs on both its own CPUs and the claimed ones. +A container which names only some requests of a claim runs on the CPUs of +those requests. The memory of the container comes from the NUMA nodes of its +CPUs, as for any other container. + +The policy tells a container which CPUs its claims got in +`/.nri-resource-policy/resources.sh`, for instance `CLAIMED_CPUS="2-3"`. + +### Memory Policy of a Claim + +A claim, or its device class, can set the Linux memory policy of the +containers holding the claim. The policy is given in a +`TopologyAwareClaimConfig` in the opaque configuration of the claim. This +claim takes one CPU of NUMA node 0 and one of NUMA node 1, and interleaves +the memory of its containers over the two nodes: + +```yaml +apiVersion: resource.k8s.io/v1 +kind: ResourceClaimTemplate +metadata: + name: interleaved +spec: + spec: + devices: + requests: + - name: first + exactly: + deviceClassName: topology-aware.nri.io + selectors: + - cel: + expression: device.attributes["resource.kubernetes.io"].numaNode == 0 + - name: second + exactly: + deviceClassName: topology-aware.nri.io + selectors: + - cel: + expression: device.attributes["resource.kubernetes.io"].numaNode == 1 + config: + - opaque: + driver: topology-aware.nri.io + parameters: + apiVersion: config.nri/v1alpha1 + kind: TopologyAwareClaimConfig + memoryPolicy: + mode: interleave +``` + +The `memoryPolicy` has these fields: + +- `mode`: `default`, `preferred`, `bind`, `interleave`, `local`, + `preferred-many` or `weighted-interleave`. +- `nodes`: the memory nodes of the policy. `all` means all nodes, + `allowed-mems` the memory nodes of the container, `cpu-nodes` the nodes of + the CPUs of the container, `cpu-packages` all nodes in the sockets of those + CPUs, `max-dist:` the nodes within the distance of `cpu-nodes`, + and a list such as `0,2-3` those nodes. Unset means `allowed-mems`. +- `flags`: `static-nodes`, `relative-nodes` or `numa-balancing`. + `numa-balancing` works only with `bind` and `preferred-many`, and `local` + takes neither `static-nodes` nor `relative-nodes`. + +The mode, the nodes and the flags are the same as in the +`policy.memory-policy.nri.io` annotation of the memory-policy plugin, with +short names instead of the `MPOL_` names: `interleave` stands for +`MPOL_INTERLEAVE`, and `static-nodes` for `MPOL_F_STATIC_NODES`. The memory-policy plugin and a claim both set the +memory policy of a container. Use one or the other for a container. + +The nodes are resolved for each container, from the CPUs it runs on and the +memory nodes it is allowed to use. If the memory of the container is not +pinned (`pinMemory: false`), all nodes are allowed. A container which names +only some requests of a claim gets the nodes of those requests. A config with +a `requests` list applies to the containers holding those requests only, so +the requests of a claim can have different memory policies. A container which +holds requests with different memory policies is refused. + +The memory policy is set when the container is created. If the policy later +changes the memory nodes of the container, the kernel remaps the nodes of the +memory policy onto the new nodes, but it does not add nodes. A memory policy +for nodes 0-1 stays on two nodes when the memory nodes grow to 0-3. + +A claim with a bad configuration fails to prepare, and the pod events say +why. + +### Known Limitations + +- The plugin forgets its claims when it restarts. Containers which hold a + claim are placed again without it, so they lose the claimed CPUs, and new + containers can get those CPUs. +- The CDI spec files of claims which ended while the plugin was not running + are left behind. +- The capacity of a device is the total of the node, not its free CPUs. + Containers with CPU requests take CPUs from the same supply, and the + scheduler does not see that, so a claim can be placed on a node with too + few free CPUs and fail to prepare. + ## Cookbook ### Mixed Workloads with Different QoS Requirements diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 90ebfc5e3..1f75218f8 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -130,8 +130,10 @@ type Agent struct { cfgIf ConfigInterface // custom resource access interface k8sCli *client.Client // wrapped kubernetes client + REST config + HTTP client - nrtCli *nrtapi.Client // NRT custom resources client - nrtLock sync.Mutex // serialize NRT custom resource updates + nrtCli nrtapi.Interface // NRT custom resources client + nrtLock sync.Mutex // lock accessing the NRT client and desired state + nrtWant *nrtState // NRT custom resource desired state + nrtWake chan struct{} // NRT reconciliation trigger podResCli *podresapi.Client // pod resources API client notifyFn NotifyFn // config resource change notification callback @@ -165,6 +167,7 @@ func New(cfgIf ConfigInterface, options ...Option) (*Agent, error) { groupLabel: defaultGroupLabel, cfgIf: cfgIf, extResWake: make(chan struct{}, 1), + nrtWake: make(chan struct{}, 1), stopC: make(chan struct{}), } @@ -205,7 +208,8 @@ func (a *Agent) Start(notifyFn NotifyFn) error { return w.ResultChan() } - go a.reconcileExtendedResourcesLoop() + go a.reconcileLoop("extended resources", a.extResWake, a.reconcileNodeExtendedResources) + go a.reconcileLoop("node resource topology CR", a.nrtWake, a.reconcileNrtCR) for { select { @@ -259,6 +263,62 @@ func (a *Agent) Start(notifyFn NotifyFn) error { } } +// resyncInterval bounds how often a reconciler re-checks the cluster +// against the desired state without an explicit request. It retries +// transient API failures and heals external drift. +var resyncInterval = 30 * time.Second + +// retrySyncInterval is time between reconciliation rounds when the +// previous round failed. Avoids flooding the API server. +var retrySyncInterval = 5 * time.Second + +// wakeUp wakes a reconciler, unless a wakeup is already pending. +func wakeUp(wake chan struct{}) { + select { + case wake <- struct{}{}: + default: + } +} + +// reconcileLoop drives some cluster state towards the most recently +// requested desired state with the given reconcile function, on every +// wakeup and periodically, until the agent stops. +func (a *Agent) reconcileLoop(what string, wake <-chan struct{}, reconcile func() error) { + // One goroutine runs the loop, so reconciles never overlap and an + // older state is never written after a newer one. The periodic run + // retries a failed write and heals changes made by someone else. + timer := time.NewTimer(resyncInterval) + defer timer.Stop() + + for { + select { + case <-a.stopC: + return + case <-wake: + case <-timer.C: + } + + // We are about to reconcile, so cancel any pending timer + // fire and recompute the next deadline from now below. This + // drains the channel if the timer already fired but we were + // woken by something else, keeping the single-fire invariant. + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + + nextInterval := resyncInterval + if err := reconcile(); err != nil { + log.Errorf("failed to reconcile %s: %v", what, err) + nextInterval = retrySyncInterval + } + + timer.Reset(nextInterval) + } +} + // Stop asks the agent to stop. It does not wait for the event loop to finish; // Start() returns once that has happened. func (a *Agent) Stop() { @@ -291,6 +351,7 @@ func (a *Agent) configure(newConfig metav1.Object) { cfg := getAgentConfig(newConfig) // Failure to create a client is not a fatal error. + a.nrtLock.Lock() switch { case cfg.NodeResourceTopology && a.nrtCli == nil: log.Infof("enabling NRT client") @@ -309,6 +370,7 @@ func (a *Agent) configure(newConfig metav1.Object) { log.Infof("disabling NRT client") a.nrtCli = nil } + a.nrtLock.Unlock() // Reconfigure pod resource client, both on initial startup and reconfiguration. // Failure to create a client is not a fatal error. diff --git a/pkg/agent/node-extended-resources.go b/pkg/agent/node-extended-resources.go index bd4eacc57..787c2e7e5 100644 --- a/pkg/agent/node-extended-resources.go +++ b/pkg/agent/node-extended-resources.go @@ -29,15 +29,6 @@ import ( "k8s.io/apimachinery/pkg/types" ) -// extResResyncInterval bounds how often the reconciler re-checks -// the Node against the desired state without an explicit request. -// It retries transient API failures and heals external drift. -const extResResyncInterval = 30 * time.Second - -// extResRetrySyncInterval is time between reconciliation rounds -// when previous round failed. Avoids flooding the API server. -const extResRetrySyncInterval = 5 * time.Second - // nriExtendedResourceGlob restricts which extended resources this // agent is ever allowed to publish or remove. Regardless of what a // policy claims to own or wants to publish, only resources whose @@ -94,67 +85,23 @@ func (a *Agent) UpdateNodeExtendedResources(resources map[string]*resource.Quant a.extResWant = snapshot a.extResLock.Unlock() - // Coalescing wakeup: if a signal is already pending, the - // reconciler will pick up the latest desired state anyway. - select { - case a.extResWake <- struct{}{}: - default: - } + wakeUp(a.extResWake) return nil } -// reconcileExtendedResourcesLoop is the single long-lived worker -// that drives the Node's status.capacity towards the most recently -// requested desired state. Running as a single goroutine, it -// guarantees in-order, non-overlapping reconciles (no last-writer -// race) and at most one in-flight GET+PATCH. It reconciles on every -// wakeup and periodically, so transient API failures and external -// drift are corrected even without new requests. -func (a *Agent) reconcileExtendedResourcesLoop() { - timer := time.NewTimer(extResResyncInterval) - defer timer.Stop() - - for { - select { - case <-a.stopC: - return - case <-a.extResWake: - case <-timer.C: - } - - // We are about to reconcile, so cancel any pending timer - // fire and recompute the next deadline from now below. This - // drains the channel if the timer already fired but we were - // woken by something else, keeping the single-fire invariant. - if !timer.Stop() { - select { - case <-timer.C: - default: - } - } - - nextInterval := extResResyncInterval - - a.extResLock.Lock() - resources := a.extResWant - a.extResLock.Unlock() - if len(resources) > 0 { - // Only touch the API server when there is a desired - // state to reconcile towards. - if err := a.reconcileNodeExtendedResources(); err != nil { - log.Errorf("failed to reconcile extended resources: %v", err) - nextInterval = extResRetrySyncInterval - } - } - - timer.Reset(nextInterval) - } -} - // reconcileNodeExtendedResources computes and applies the JSON // patch that brings Node.status.capacity in line with the most // recently requested desired state. func (a *Agent) reconcileNodeExtendedResources() error { + // Only touch the API server when there is a desired state to + // reconcile towards. + a.extResLock.Lock() + want := len(a.extResWant) + a.extResLock.Unlock() + if want == 0 { + return nil + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() node, err := a.k8sCli.CoreV1().Nodes().Get(ctx, a.nodeName, metav1.GetOptions{}) diff --git a/pkg/agent/node-resource-topology.go b/pkg/agent/node-resource-topology.go index f1aa39e44..fa4670346 100644 --- a/pkg/agent/node-resource-topology.go +++ b/pkg/agent/node-resource-topology.go @@ -17,7 +17,9 @@ package agent import ( "context" "fmt" + "time" + "k8s.io/apimachinery/pkg/api/equality" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -25,41 +27,47 @@ import ( nrt "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/apis/topology/v1alpha2" ) -// UpdateNrtCR updates the node's node resource topology CR using the given data. +// nrtState is the desired content of the node resource topology CR. +type nrtState struct { + policy string + zones []*policyapi.TopologyZone +} + +// UpdateNrtCR records the given data as the desired state of the node's +// node resource topology CR and wakes the reconciler to write it. +// +// A newer request replaces older ones. The reconciler writes the latest +// snapshot only, so bursts coalesce and an old snapshot never overwrites a +// new one. func (a *Agent) UpdateNrtCR(policy string, zones []*policyapi.TopologyZone) error { if a.hasLocalConfig() { return nil } - if a.nrtCli == nil { - return fmt.Errorf("no node resource topology client, can't update CR") - } - - log.Infof("updating node resource topology CR") - - // To minimize the risk of an NRI request timeout (and the plugin getting - // kicked out) we do the update asynchronously. We can rework this to use - // a single goroutine that reads update requests from a channel to mimic - // the rest if necessary. - // XXX TODO(klihub): We can't/don't propagate update errors now back - // to the caller. We could do that (using a channel) if necessary... - go func() { - if err := a.updateNrtCR(policy, zones); err != nil { - log.Errorf("failed to update topology CR: %v", err) - } - }() + a.nrtLock.Lock() + a.nrtWant = &nrtState{policy: policy, zones: zones} + a.nrtLock.Unlock() + wakeUp(a.nrtWake) return nil } -// updateNrtCR updates the node's node resource topology CR using the given data. -func (a *Agent) updateNrtCR(policy string, zones []*policyapi.TopologyZone) error { +// reconcileNrtCR writes the most recently requested desired state to the +// node's node resource topology CR. It does nothing without an NRT client +// or without a desired state. +func (a *Agent) reconcileNrtCR() error { a.nrtLock.Lock() - defer a.nrtLock.Unlock() + cli, want := a.nrtCli, a.nrtWant + a.nrtLock.Unlock() + if cli == nil || want == nil { + return nil + } - cli := a.nrtCli.NodeResourceTopologies() - ctx := context.Background() - cr, err := cli.Get(ctx, a.nodeName, metav1.GetOptions{}) + nrtCli := cli.NodeResourceTopologies() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + cr, err := nrtCli.Get(ctx, a.nodeName, metav1.GetOptions{}) if err != nil { cr = nil if !errors.IsNotFound(err) { @@ -67,24 +75,30 @@ func (a *Agent) updateNrtCR(policy string, zones []*policyapi.TopologyZone) erro } } + attributes := nrt.AttributeList{ + nrt.AttributeInfo{ + Name: "TopologyPolicy", + Value: want.policy, + }, + } + zones := zonesToNrt(want.zones) + // XXX TODO Deletion should be handled differently: // 1. add expiration timestamp to nrt.NodeResourceTopology // 2. GC CRs that are past their expiration time (for instance by NFD) - // 3. make sure we refresh our CR (either here or preferably/easier - // by triggering in resmgr an updateTopologyZones() during longer - // periods of inactivity) // update CR if one exists if cr != nil { - cr.Attributes = nrt.AttributeList{ - nrt.AttributeInfo{ - Name: "TopologyPolicy", - Value: policy, - }, + // A periodic resync only writes when the CR differs. + if equality.Semantic.DeepEqual(cr.Attributes, attributes) && equality.Semantic.DeepEqual(cr.Zones, zones) { + return nil } - cr.Zones = zonesToNrt(zones) + log.Infof("updating node resource topology CR") - _, err = cli.Update(ctx, cr, metav1.UpdateOptions{}) + cr.Attributes = attributes + cr.Zones = zones + + _, err = nrtCli.Update(ctx, cr, metav1.UpdateOptions{}) if err != nil { return fmt.Errorf("failed to update node resource topology CR: %w", err) } @@ -93,21 +107,17 @@ func (a *Agent) updateNrtCR(policy string, zones []*policyapi.TopologyZone) erro } // or create a new one + log.Infof("creating node resource topology CR") + cr = &nrt.NodeResourceTopology{ ObjectMeta: metav1.ObjectMeta{ Name: a.nodeName, }, - - Attributes: nrt.AttributeList{ - nrt.AttributeInfo{ - Name: "TopologyPolicy", - Value: policy, - }, - }, - Zones: zonesToNrt(zones), + Attributes: attributes, + Zones: zones, } - _, err = cli.Create(ctx, cr, metav1.CreateOptions{}) + _, err = nrtCli.Create(ctx, cr, metav1.CreateOptions{}) if err != nil { return fmt.Errorf("failed to create node resource topology CR: %w", err) } diff --git a/pkg/agent/node-resource-topology_test.go b/pkg/agent/node-resource-topology_test.go new file mode 100644 index 000000000..85be3f2de --- /dev/null +++ b/pkg/agent/node-resource-topology_test.go @@ -0,0 +1,168 @@ +// Copyright The NRI Plugins Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package agent + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + policyapi "github.com/containers/nri-plugins/pkg/resmgr/policy" + nrt "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/apis/topology/v1alpha2" + nrtfake "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/clientset/versioned/fake" +) + +const testNode = "node0" + +// newNrtTestAgent returns an agent with a fake NRT client and no worker. +func newNrtTestAgent() (*Agent, *nrtfake.Clientset) { + cs := nrtfake.NewSimpleClientset() + a := &Agent{ + nodeName: testNode, + nrtCli: cs.TopologyV1alpha2(), + nrtWake: make(chan struct{}, 1), + stopC: make(chan struct{}), + } + return a, cs +} + +// startNrtWorker runs the NRT reconciler until the agent is stopped. It +// returns a channel closed when the worker has exited. +func startNrtWorker(a *Agent) <-chan struct{} { + done := make(chan struct{}) + go func() { + defer close(done) + a.reconcileLoop("node resource topology CR", a.nrtWake, a.reconcileNrtCR) + }() + return done +} + +func zones(names ...string) []*policyapi.TopologyZone { + var out []*policyapi.TopologyZone + for _, name := range names { + out = append(out, &policyapi.TopologyZone{Name: name, Type: "Node"}) + } + return out +} + +func getCR(t *testing.T, cs *nrtfake.Clientset) *nrt.NodeResourceTopology { + cr, err := cs.TopologyV1alpha2().NodeResourceTopologies().Get(context.Background(), testNode, metav1.GetOptions{}) + require.NoError(t, err) + return cr +} + +// waitForZones waits until the CR has the given zones. +func waitForZones(t *testing.T, cs *nrtfake.Clientset, want []*policyapi.TopologyZone) { + require.Eventually(t, func() bool { + cr, err := cs.TopologyV1alpha2().NodeResourceTopologies().Get(context.Background(), testNode, metav1.GetOptions{}) + if err != nil { + return false + } + return len(cr.Zones) == len(want) && cr.Zones[0].Name == want[0].Name + }, 5*time.Second, 5*time.Millisecond) +} + +func waitForStop(t *testing.T, a *Agent, done <-chan struct{}) { + a.Stop() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("worker did not stop") + } +} + +func TestNrtCRLatestUpdateWins(t *testing.T) { + a, cs := newNrtTestAgent() + done := startNrtWorker(a) + + require.NoError(t, a.UpdateNrtCR("policy", zones("old"))) + require.NoError(t, a.UpdateNrtCR("policy", zones("new"))) + waitForZones(t, cs, zones("new")) + waitForStop(t, a, done) + + cr := getCR(t, cs) + require.Equal(t, "new", cr.Zones[0].Name) + require.Equal(t, "policy", cr.Attributes[0].Value) +} + +func TestNrtCRBurstCoalesces(t *testing.T) { + a, cs := newNrtTestAgent() + + for i := 0; i < 5; i++ { + require.NoError(t, a.UpdateNrtCR("policy", zones("zone"))) + } + require.Len(t, a.nrtWake, 1, "a burst leaves one pending wakeup") + + require.NoError(t, a.reconcileNrtCR()) + writes := 0 + for _, action := range cs.Actions() { + if action.GetVerb() == "create" || action.GetVerb() == "update" { + writes++ + } + } + require.Equal(t, 1, writes, "one write for the whole burst") +} + +func TestNrtCRUnchangedIsNotWritten(t *testing.T) { + a, cs := newNrtTestAgent() + + zone := &policyapi.TopologyZone{ + Name: "zone", + Type: "Node", + Resources: []*policyapi.ZoneResource{{Name: "cpu", Capacity: resource.MustParse("4")}}, + } + require.NoError(t, a.UpdateNrtCR("policy", []*policyapi.TopologyZone{zone})) + require.NoError(t, a.reconcileNrtCR()) + require.NoError(t, a.reconcileNrtCR()) + + writes := 0 + for _, action := range cs.Actions() { + if action.GetVerb() == "create" || action.GetVerb() == "update" { + writes++ + } + } + require.Equal(t, 1, writes, "a resync of an unchanged CR writes nothing") +} + +func TestNrtCRResyncHealsDrift(t *testing.T) { + resyncInterval = 10 * time.Millisecond + t.Cleanup(func() { resyncInterval = 30 * time.Second }) + + a, cs := newNrtTestAgent() + done := startNrtWorker(a) + + require.NoError(t, a.UpdateNrtCR("policy", zones("ours"))) + waitForZones(t, cs, zones("ours")) + + cr := getCR(t, cs) + cr.Zones[0].Name = "theirs" + _, err := cs.TopologyV1alpha2().NodeResourceTopologies().Update(context.Background(), cr, metav1.UpdateOptions{}) + require.NoError(t, err) + + waitForZones(t, cs, zones("ours")) + waitForStop(t, a, done) +} + +func TestNrtCRWithoutClient(t *testing.T) { + a, _ := newNrtTestAgent() + a.nrtCli = nil + + require.NoError(t, a.UpdateNrtCR("policy", zones("zone"))) + require.NoError(t, a.reconcileNrtCR()) +} diff --git a/pkg/agent/nrtapi/nrtapi.go b/pkg/agent/nrtapi/nrtapi.go index 54ab9a7ce..12a1ad96e 100644 --- a/pkg/agent/nrtapi/nrtapi.go +++ b/pkg/agent/nrtapi/nrtapi.go @@ -24,9 +24,12 @@ import ( client "github.com/k8stopologyawareschedwg/noderesourcetopology-api/pkg/generated/clientset/versioned/typed/topology/v1alpha2" ) -// Clientset is the client for accessing node resource topology custom resources. +// Client is the client for accessing node resource topology custom resources. type Client = client.TopologyV1alpha2Client +// Interface is the interface of Client, for fakes in tests. +type Interface = client.TopologyV1alpha2Interface + // NewForConfigAndClient creates a new Clientset for the given config and http client. func NewForConfigAndClient(c *rest.Config, httpCli *http.Client) (*Client, error) { restCfg := *c diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh b/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh index 5bb7b71aa..598b21a32 100644 --- a/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh +++ b/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/code.var.sh @@ -8,7 +8,6 @@ cleanup() { vm-command "kubectl delete pods --all --now --wait" || : vm-command "kubectl delete resourceclaimtemplates --all --now" || : - vm-command "kubectl delete deviceclass topology-aware.nri.io --ignore-not-found" || : helm-terminate || : } @@ -137,8 +136,6 @@ for node in 0 1 2 3; do echo "$name: numaNode $numa, socketID $socket, smtEnabled $smt, $cpus CPUs" done -wait="" create deviceclass - # pod0 claims 3 CPUs on node 3. It gets all node CPUs except the reserved # CPU 15. claim pod0 3 3 diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/deviceclass.yaml.in b/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/deviceclass.yaml.in deleted file mode 100644 index 1e8b34d83..000000000 --- a/test/e2e/policies.test-suite/topology-aware/n4c16/test42-dra-devices/deviceclass.yaml.in +++ /dev/null @@ -1,8 +0,0 @@ -apiVersion: resource.k8s.io/v1 -kind: DeviceClass -metadata: - name: topology-aware.nri.io -spec: - selectors: - - cel: - expression: device.driver == \"topology-aware.nri.io\" diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test45-dra-memory-policy/code.var.sh b/test/e2e/policies.test-suite/topology-aware/n4c16/test45-dra-memory-policy/code.var.sh index ca14406cd..76c57c4a0 100644 --- a/test/e2e/policies.test-suite/topology-aware/n4c16/test45-dra-memory-policy/code.var.sh +++ b/test/e2e/policies.test-suite/topology-aware/n4c16/test45-dra-memory-policy/code.var.sh @@ -7,7 +7,6 @@ cleanup() { vm-command "kubectl delete pods --all --now --wait" || : vm-command "kubectl delete resourceclaimtemplates --all --now" || : - vm-command "kubectl delete deviceclass topology-aware.nri.io --ignore-not-found" || : helm-terminate || : } @@ -43,8 +42,6 @@ if [ "$(slice ".spec.devices[0].allowMultipleAllocations")" != "true" ]; then exit 0 fi -wait="" create deviceclass - # pod0's claim asks for one CPU of node 0 and one of node 1, both in package # 0. Its config interleaves memory on the nodes the containers may use. # pod0c0 holds both requests, so it interleaves on nodes 0 and 1. pod0c1 diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test45-dra-memory-policy/deviceclass.yaml.in b/test/e2e/policies.test-suite/topology-aware/n4c16/test45-dra-memory-policy/deviceclass.yaml.in deleted file mode 100644 index 1e8b34d83..000000000 --- a/test/e2e/policies.test-suite/topology-aware/n4c16/test45-dra-memory-policy/deviceclass.yaml.in +++ /dev/null @@ -1,8 +0,0 @@ -apiVersion: resource.k8s.io/v1 -kind: DeviceClass -metadata: - name: topology-aware.nri.io -spec: - selectors: - - cel: - expression: device.driver == \"topology-aware.nri.io\"