diff --git a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go index b006db12749..b1ba2e46895 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go @@ -473,6 +473,27 @@ func DRAService(enabled bool) Option { } } +// ReconcilePoolWithName limits reconciliation to slices with Spec.Pool.Name +// equal to name. +// +// If set, the controller enqueues only ResourceSlices with a matching +// Spec.Pool.Name and does not set Spec.NodeName (even for Node owners). +// This enables node-owned slices that remain cluster-visible via +// NodeSelector or AllNodes. +// +// Beware that this has a performance impact on the cluster +// because all nodes have to receive all ResourceSlices of +// the driver. Without this option, each node only receives +// its own ResourceSlices. +// +// Empty means the default behavior. +func ReconcilePoolWithName(name string) Option { + return func(o *options) error { + o.reconcilePoolWithName = name + return nil + } +} + type options struct { logger klog.Logger grpcVerbosity int @@ -494,6 +515,7 @@ type options struct { registrationService bool draService bool healthService *bool + reconcilePoolWithName string } // Helper combines the kubelet registration service and the DRA node plugin @@ -502,19 +524,20 @@ type Helper struct { // backgroundCtx is for activities that are started later. backgroundCtx context.Context // cancel cancels the backgroundCtx. - cancel func(cause error) - wg sync.WaitGroup - registrar *nodeRegistrar - pluginServer *grpcServer - plugin DRAPlugin - driverName string - nodeName string - nodeUID types.UID - kubeClient kubernetes.Interface - resourceClient cgoresource.ResourceV1Interface - serialize bool - grpcMutex sync.Mutex - grpcLockFilePath string + cancel func(cause error) + wg sync.WaitGroup + registrar *nodeRegistrar + pluginServer *grpcServer + plugin DRAPlugin + driverName string + nodeName string + nodeUID types.UID + kubeClient kubernetes.Interface + resourceClient cgoresource.ResourceV1Interface + serialize bool + grpcMutex sync.Mutex + grpcLockFilePath string + reconcilePoolWithName string // Information about resource publishing changes concurrently and thus // must be protected by the mutex. The controller gets started only @@ -578,13 +601,14 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe } d := &Helper{ - driverName: o.driverName, - nodeName: o.nodeName, - nodeUID: o.nodeUID, - kubeClient: o.kubeClient, - resourceClient: draclient.New(o.kubeClient), - serialize: o.serialize, - plugin: plugin, + driverName: o.driverName, + nodeName: o.nodeName, + nodeUID: o.nodeUID, + kubeClient: o.kubeClient, + resourceClient: draclient.New(o.kubeClient), + serialize: o.serialize, + plugin: plugin, + reconcilePoolWithName: o.reconcilePoolWithName, } if o.rollingUpdateUID != "" { dir := o.pluginDataDirectoryPath @@ -802,6 +826,7 @@ func (d *Helper) PublishResources(_ context.Context, resources resourceslice.Dri // -> all errors are recoverable. d.plugin.HandleError(ctx, recoverableError{error: err}, msg) }, + ReconcilePoolWithName: d.reconcilePoolWithName, }); err != nil { return fmt.Errorf("start ResourceSlice controller: %w", err) } diff --git a/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller.go b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller.go index 9d8a3046f1b..56a4e98e894 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller.go @@ -123,6 +123,9 @@ type Controller struct { // so it is okay to not do a deep copy of it when reading it. Only reading // the pointer itself must be protected by a read lock. resources *DriverResources + + // Optional pool name to reconcile. + reconcilePoolWithName string } // +k8s:deepcopy-gen=true @@ -144,10 +147,13 @@ type DriverResources struct { // Pool is the collection of devices belonging to the same pool. type Pool struct { // NodeSelector may be different for each pool. Must not get set together - // with Resources.NodeName. If nil and Resources.NodeName is not set, - // then devices are available on all nodes. + // with Resources.NodeName, Slices.PerDeviceNodeSelection or AllNodes. NodeSelector *v1.NodeSelector + // AllNodes may be different for each pool. Must not get set together + // with Resources.NodeName, Slices.PerDeviceNodeSelection or NodeSelector. + AllNodes bool + // Generation can be left at zero. It gets bumped up automatically // by the controller. Generation int64 @@ -282,6 +288,21 @@ type Options struct { // The default is [utilruntime.HandleErrorWithContext] which just logs // the problem. ErrorHandler func(ctx context.Context, err error, msg string) + + // ReconcilePoolWithName limits reconciliation to a single pool. + // + // If set, the controller enqueues only ResourceSlices with a matching + // Spec.Pool.Name and does not set Spec.NodeName (even for Node owners). + // This enables node-owned slices that remain cluster-visible via + // NodeSelector or AllNodes. + // + // Beware that this has a performance impact on the cluster + // because all nodes have to receive all ResourceSlices of + // the driver. Without this option, each node only receives + // its own ResourceSlices. + // + // Empty means the default behavior. + ReconcilePoolWithName string } // DroppedFieldsError is reported through the ErrorHandler in [Options] if @@ -379,6 +400,20 @@ func (c *Controller) Update(resources *DriverResources) { if resources == nil { c.resources = &DriverResources{} } else { + // If reconcilePoolWithName is set, we expect to reconcile only a single pool. + // Having additional pools is considered an error. However, an empty pool list + // is intentionally allowed and treated as "no slices to publish", which matches + // the default controller behavior. + if c.reconcilePoolWithName != "" { + _, ok := resources.Pools[c.reconcilePoolWithName] + if (ok && len(resources.Pools) > 1) || !ok && len(resources.Pools) > 0 { + c.errorHandler(context.Background(), + fmt.Errorf("ReconcilePoolWithName=%q, but found %d pools; expected exactly one pool with this name", c.reconcilePoolWithName, len(resources.Pools)), + "processing update DriverResources") + return + } + } + c.resources = resources.DeepCopy() roundTaintTimeAdded(c.resources) } @@ -437,16 +472,17 @@ func newController(ctx context.Context, options Options) (*Controller, error) { ctx, cancel := context.WithCancelCause(ctx) c := &Controller{ - cancel: cancel, - resourceClient: draclient.New(options.KubeClient), - coreClient: options.KubeClient.CoreV1(), - driverName: options.DriverName, - owner: options.Owner.DeepCopy(), - queue: options.Queue, - mutationCacheTTL: ptr.Deref(options.MutationCacheTTL, DefaultMutationCacheTTL), - syncDelay: ptr.Deref(options.SyncDelay, DefaultSyncDelay), - errorHandler: options.ErrorHandler, - lastAddByPool: make(map[string]time.Time), + cancel: cancel, + resourceClient: draclient.New(options.KubeClient), + coreClient: options.KubeClient.CoreV1(), + driverName: options.DriverName, + owner: options.Owner.DeepCopy(), + queue: options.Queue, + mutationCacheTTL: ptr.Deref(options.MutationCacheTTL, DefaultMutationCacheTTL), + syncDelay: ptr.Deref(options.SyncDelay, DefaultSyncDelay), + errorHandler: options.ErrorHandler, + lastAddByPool: make(map[string]time.Time), + reconcilePoolWithName: options.ReconcilePoolWithName, } if c.queue == nil { c.queue = workqueue.NewTypedRateLimitingQueueWithConfig( @@ -477,7 +513,10 @@ func (c *Controller) initInformer(ctx context.Context) error { resourceapi.ResourceSliceSelectorDriver: c.driverName, resourceapi.ResourceSliceSelectorNodeName: "", } - if c.owner != nil && c.owner.APIVersion == "v1" && c.owner.Kind == "Node" { + // TODO: We can list/watch ResourceSlices with field selectors for NodeName or Driver. + // There is no field selector for PoolName, so we apply additional client-side filtering in list/watch. Issue for adding a PoolName field selector: │ + // https://github.com/kubernetes/kubernetes/issues/137413 + if c.owner != nil && c.owner.APIVersion == "v1" && c.owner.Kind == "Node" && c.reconcilePoolWithName == "" { selector[resourceapi.ResourceSliceSelectorNodeName] = c.owner.Name } tweakListOptions := func(options *metav1.ListOptions) { @@ -502,12 +541,21 @@ func (c *Controller) initInformer(ctx context.Context) error { } else { logger.V(5).Info("Listed ResourceSlices", "resourceAPI", c.resourceClient.CurrentAPI(), "err", err) } + + if c.reconcilePoolWithName != "" { + retainSlicesByPoolName(slices, c.reconcilePoolWithName) + } return slices, err }, WatchFuncWithContext: func(ctx context.Context, options metav1.ListOptions) (watch.Interface, error) { tweakListOptions(&options) w, err := c.resourceClient.ResourceSlices().Watch(ctx, options) logger.V(5).Info("Started watching ResourceSlices", "resourceAPI", c.resourceClient.CurrentAPI(), "err", err) + + if c.reconcilePoolWithName != "" { + return filterSliceWatchByPoolName(ctx, w, c.reconcilePoolWithName), nil + } + return w, err }, }, c.resourceClient), @@ -661,7 +709,9 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error { // the controller runs. var nodeName string if c.owner != nil && c.owner.APIVersion == "v1" && c.owner.Kind == "Node" { - nodeName = c.owner.Name + if c.reconcilePoolWithName == "" { + nodeName = c.owner.Name + } if c.owner.UID == "" { node, err := c.coreClient.Nodes().Get(ctx, c.owner.Name, metav1.GetOptions{}) if err != nil { @@ -719,7 +769,6 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error { Generation: generation, // May get updated later. ResourceSliceCount: int64(resourceSliceCount), } - desiredAllNodes := pool.NodeSelector == nil && nodeName == "" // Now for each desired slice, figure out which of them are changed. changedDesiredSlices := sets.New[int]() @@ -728,7 +777,7 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error { // entries are the same. if !apiequality.Semantic.DeepEqual(¤tSlice.Spec.Pool, &desiredPool) || !apiequality.Semantic.DeepEqual(currentSlice.Spec.NodeSelector, pool.NodeSelector) || - ptr.Deref(currentSlice.Spec.AllNodes, false) != desiredAllNodes || + ptr.Deref(currentSlice.Spec.AllNodes, false) != pool.AllNodes || !DevicesDeepEqual(currentSlice.Spec.Devices, pool.Slices[i].Devices) || !apiequality.Semantic.DeepEqual(currentSlice.Spec.SharedCounters, pool.Slices[i].SharedCounters) || !apiequality.Semantic.DeepEqual(currentSlice.Spec.PerDeviceNodeSelection, pool.Slices[i].PerDeviceNodeSelection) { @@ -778,7 +827,7 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error { // // When adding new fields here, then also extend sliceStored. slice.Spec.NodeSelector = pool.NodeSelector - slice.Spec.AllNodes = refIfNotZero(desiredAllNodes) + slice.Spec.AllNodes = refIfNotZero(pool.AllNodes) slice.Spec.SharedCounters = pool.Slices[i].SharedCounters slice.Spec.PerDeviceNodeSelection = pool.Slices[i].PerDeviceNodeSelection // Preserve TimeAdded from existing device, if there is a matching device and taint. @@ -830,7 +879,7 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error { Pool: desiredPool, NodeName: refIfNotZero(nodeName), NodeSelector: pool.NodeSelector, - AllNodes: refIfNotZero(desiredAllNodes), + AllNodes: refIfNotZero(pool.AllNodes), Devices: pool.Slices[i].Devices, SharedCounters: pool.Slices[i].SharedCounters, PerDeviceNodeSelection: pool.Slices[i].PerDeviceNodeSelection, @@ -1081,3 +1130,16 @@ func decodeIndex(name string, expectedLength int) (int, error) { func encodeIndex(index int, expectedLength int) string { return fmt.Sprintf("%0*x", expectedLength, index) } + +func retainSlicesByPoolName(sliceList *resourceapi.ResourceSliceList, poolName string) { + sliceList.Items = slices.DeleteFunc(sliceList.Items, func(slice resourceapi.ResourceSlice) bool { + return slice.Spec.Pool.Name != poolName + }) +} + +func filterSliceWatchByPoolName(ctx context.Context, w watch.Interface, poolName string) watch.Interface { + return newWrapWatcher(ctx, w, func(event watch.Event) bool { + resourceSlice, ok := event.Object.(*resourceapi.ResourceSlice) + return ok && resourceSlice.Spec.Pool.Name == poolName + }) +} diff --git a/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller_test.go b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller_test.go index 39d91fa0b7f..201c189ebbd 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller_test.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller_test.go @@ -105,6 +105,9 @@ func TestControllerSyncPool(t *testing.T) { nodeUID types.UID // noOwner completely disables setting an owner. noOwner bool + // reconcilePoolWithName limits reconciliation to a single pool (issue #137011). + // When set, NodeName is not set on slices even for Node owner. + reconcilePoolWithName string // initialObjects is a list of initial resource slices to be used in the test. initialObjects []runtime.Object initialOtherObjects []runtime.Object @@ -802,7 +805,8 @@ func TestControllerSyncPool(t *testing.T) { inputDriverResources: &DriverResources{ Pools: map[string]Pool{ poolName: { - Slices: []Slice{{Devices: []resourceapi.Device{newDevice(deviceName)}}}, + Slices: []Slice{{Devices: []resourceapi.Device{newDevice(deviceName)}}}, + AllNodes: true, }, }, }, @@ -813,6 +817,9 @@ func TestControllerSyncPool(t *testing.T) { *MakeResourceSlice().Name(encodeIndex(0, resourceSliceIndexMinLength) + "-" + driverName + "-0"). GenerateName(encodeIndex(0, resourceSliceIndexMinLength) + "-" + driverName + "-"). AllNodes(true). + NodeName(""). + NodeSelector(nil). + PerDeviceNodeSelection(false). Driver(driverName).Devices([]resourceapi.Device{newDevice(deviceName)}). Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 1}).Obj(), }, @@ -837,6 +844,71 @@ func TestControllerSyncPool(t *testing.T) { Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 1}).Obj(), }, }, + "reconcile-with-pool-name-create-slice-no-node-name": { + nodeUID: nodeUID, + reconcilePoolWithName: poolName, + initialObjects: []runtime.Object{}, + inputDriverResources: &DriverResources{ + Pools: map[string]Pool{ + poolName: { + Slices: []Slice{{Devices: []resourceapi.Device{}}}, + AllNodes: true, + }, + }, + }, + expectedStats: Stats{NumCreates: 1}, + expectedResourceSlices: []resourceapi.ResourceSlice{ + *MakeResourceSlice().Name(resourceSlice1).GenerateName(generateName1). + NodeOwnerReferences(ownerName, string(nodeUID)). + AllNodes(true). + NodeName(""). + NodeSelector(nil). + PerDeviceNodeSelection(false). + Driver(driverName).Devices([]resourceapi.Device{}). + Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 1}).Obj(), + }, + }, + "reconcile-with-pool-name-with-node-selector": { + nodeUID: nodeUID, + reconcilePoolWithName: poolName, + initialObjects: []runtime.Object{}, + inputDriverResources: &DriverResources{ + Pools: map[string]Pool{ + poolName: { + NodeSelector: nodeSelector, + Slices: []Slice{{Devices: []resourceapi.Device{newDevice(deviceName)}}}, + }, + }, + }, + expectedStats: Stats{NumCreates: 1}, + expectedResourceSlices: []resourceapi.ResourceSlice{ + *MakeResourceSlice().Name(resourceSlice1).GenerateName(generateName1). + NodeOwnerReferences(ownerName, string(nodeUID)). + NodeSelector(nodeSelector). + Driver(driverName).Devices([]resourceapi.Device{newDevice(deviceName)}). + Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 1}).Obj(), + }, + }, + "reconcile-with-pool-name-with-perdevice-node-selector": { + nodeUID: nodeUID, + reconcilePoolWithName: poolName, + initialObjects: []runtime.Object{}, + inputDriverResources: &DriverResources{ + Pools: map[string]Pool{ + poolName: { + Slices: []Slice{{Devices: []resourceapi.Device{newDevice(deviceName, nodeSelector)}, PerDeviceNodeSelection: ptr.To(true)}}, + }, + }, + }, + expectedStats: Stats{NumCreates: 1}, + expectedResourceSlices: []resourceapi.ResourceSlice{ + *MakeResourceSlice().Name(resourceSlice1).GenerateName(generateName1). + NodeOwnerReferences(ownerName, string(nodeUID)). + PerDeviceNodeSelection(true). + Driver(driverName).Devices([]resourceapi.Device{newDevice(deviceName, nodeSelector)}). + Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 1}).Obj(), + }, + }, "update-node-selector": { initialObjects: []runtime.Object{ MakeResourceSlice().Name(resourceSlice1).UID(resourceSlice1). @@ -863,7 +935,6 @@ func TestControllerSyncPool(t *testing.T) { }, }, "create-partitionable-devices": { - nodeUID: nodeUID, inputDriverResources: &DriverResources{ Pools: map[string]Pool{ poolName: { @@ -902,7 +973,10 @@ func TestControllerSyncPool(t *testing.T) { }, expectedResourceSlices: []resourceapi.ResourceSlice{ *MakeResourceSlice().Name(resourceSlice1).GenerateName(generateName1). - NodeOwnerReferences(ownerName, string(nodeUID)).NodeName(ownerName). + AppOwnerReferences(ownerName). + AllNodes(false). + NodeName(""). + NodeSelector(nil). PerDeviceNodeSelection(true). SharedCounters([]resourceapi.CounterSet{{ Name: "gpu-0", @@ -914,7 +988,10 @@ func TestControllerSyncPool(t *testing.T) { Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 2}). Obj(), *MakeResourceSlice().Name(resourceSlice2).GenerateName(generateName2). - NodeOwnerReferences(ownerName, string(nodeUID)).NodeName(ownerName). + AppOwnerReferences(ownerName). + AllNodes(false). + NodeName(""). + NodeSelector(nil). PerDeviceNodeSelection(true). Driver(driverName). Devices([]resourceapi.Device{ @@ -935,7 +1012,6 @@ func TestControllerSyncPool(t *testing.T) { }, "drop-partitionable-devices": { features: features{disablePartitionableDevices: true}, - nodeUID: nodeUID, inputDriverResources: &DriverResources{ Pools: map[string]Pool{ poolName: { @@ -974,12 +1050,20 @@ func TestControllerSyncPool(t *testing.T) { }, expectedResourceSlices: []resourceapi.ResourceSlice{ *MakeResourceSlice().Name(resourceSlice1).GenerateName(generateName1). - NodeOwnerReferences(ownerName, string(nodeUID)).NodeName(ownerName). + AppOwnerReferences(ownerName). + AllNodes(false). + NodeName(""). + NodeSelector(nil). + PerDeviceNodeSelection(false). // Should be dropped. Driver(driverName). Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 2}). Obj(), *MakeResourceSlice().Name(resourceSlice2).GenerateName(generateName2). - NodeOwnerReferences(ownerName, string(nodeUID)).NodeName(ownerName). + AppOwnerReferences(ownerName). + AllNodes(false). + NodeName(""). + NodeSelector(nil). + PerDeviceNodeSelection(false). // Should be dropped. Driver(driverName). Devices([]resourceapi.Device{newDevice(deviceName)}). Pool(resourceapi.ResourcePool{Name: poolName, Generation: 1, ResourceSliceCount: 2}). @@ -1327,6 +1411,7 @@ func TestControllerSyncPool(t *testing.T) { ErrorHandler: func(ctx context.Context, err error, msg string) { controllerErrors = append(controllerErrors, fmt.Errorf("%s: %w", msg, err)) }, + ReconcilePoolWithName: test.reconcilePoolWithName, }) defer ctrl.Stop() require.NoError(t, err, "unexpected controller creation error") @@ -1400,6 +1485,69 @@ func TestControllerSyncPool(t *testing.T) { } } +// TestControllerUpdateReconcilePoolWithNameValidation verifies that Update rejects +// invalid pool sets when ReconcilePoolWithName is set +func TestControllerUpdateReconcilePoolWithNameValidation(t *testing.T) { + const poolName = "pool" + + testcases := map[string]struct { + resources *DriverResources + expectedErrors []string + }{ + "multiple pools returns error": { + resources: &DriverResources{ + Pools: map[string]Pool{ + poolName: {Slices: []Slice{{Devices: []resourceapi.Device{}}}}, + "other-pool": {Slices: []Slice{{Devices: []resourceapi.Device{}}}}, + }, + }, + expectedErrors: []string{"found 2 pools; expected exactly one pool with this name"}, + }, + + "wrong pool only returns error": { + resources: &DriverResources{ + Pools: map[string]Pool{ + "other-pool": {Slices: []Slice{{Devices: []resourceapi.Device{}}}}, + }, + }, + expectedErrors: []string{"found 1 pools; expected exactly one pool with this name"}, + }, + + "empty pools succeeds": { + resources: &DriverResources{Pools: map[string]Pool{}}, + }, + + "single matching pool succeeds": { + resources: &DriverResources{ + Pools: map[string]Pool{ + poolName: {Slices: []Slice{{Devices: []resourceapi.Device{}}}}, + }, + }, + }, + } + + for name, tc := range testcases { + t.Run(name, func(t *testing.T) { + ctrl := &Controller{ + reconcilePoolWithName: poolName, + queue: ptr.To(workqueue.Mock[string]{}), + errorHandler: func(_ context.Context, err error, _ string) { + if len(tc.expectedErrors) > 0 { + require.Error(t, err) + for _, expectedError := range tc.expectedErrors { + assert.Contains(t, err.Error(), expectedError) + } + return + } + + require.NoError(t, err) + }, + } + ctrl.Update(tc.resources) + }) + } +} + func joinErrors(errors []string) string { return strings.Join(errors, "\n ") } @@ -1682,6 +1830,8 @@ func newDevice(name string, fields ...any) resourceapi.Device { device.NodeName = ptr.To(string(f)) case allowMultipleAllocationsField: device.AllowMultipleAllocations = ptr.To(bool(f)) + case *v1.NodeSelector: + device.NodeSelector = f default: panic(fmt.Sprintf("unsupported resourceapi.Device field type %T", field)) } diff --git a/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/watcher.go b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/watcher.go new file mode 100644 index 00000000000..97585e9b958 --- /dev/null +++ b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/watcher.go @@ -0,0 +1,86 @@ +/* +Copyright The Kubernetes Authors. + +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 resourceslice + +import ( + "context" + + "k8s.io/apimachinery/pkg/watch" +) + +func newWrapWatcher(ctx context.Context, w watch.Interface, match func(event watch.Event) bool) *wrapWatcher { + ctx, cancel := context.WithCancel(ctx) + + watcher := &wrapWatcher{ + watcher: w, + match: match, + ctx: ctx, + cancel: cancel, + result: make(chan watch.Event), + done: make(chan struct{}), + } + go watcher.receive(ctx) + + return watcher +} + +type wrapWatcher struct { + watcher watch.Interface + match func(event watch.Event) bool + + ctx context.Context + cancel context.CancelFunc + result chan watch.Event + done chan struct{} +} + +func (w *wrapWatcher) receive(ctx context.Context) { + defer close(w.result) + defer close(w.done) + resultChan := w.watcher.ResultChan() + for { + select { + case <-ctx.Done(): + return + case event, ok := <-resultChan: + if !ok { + return + } + if w.match == nil || w.match(event) { + select { + case <-ctx.Done(): + return + case w.result <- event: + } + } + } + } +} + +func (w *wrapWatcher) ResultChan() <-chan watch.Event { + return w.result +} + +func (w *wrapWatcher) Stop() { + select { + case <-w.ctx.Done(): + default: + w.watcher.Stop() + w.cancel() + } + <-w.done +} diff --git a/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/watcher_test.go b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/watcher_test.go new file mode 100644 index 00000000000..4d77ac7128a --- /dev/null +++ b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/watcher_test.go @@ -0,0 +1,106 @@ +/* +Copyright The Kubernetes Authors. + +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 resourceslice + +import ( + "testing" + "testing/synctest" + + "github.com/stretchr/testify/require" + + resourceapi "k8s.io/api/resource/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" +) + +func TestFilterSliceWatchByPoolName_FiltersByPoolName(t *testing.T) { + synctest.Test(t, testFilterSliceWatchByPoolNameFiltersByPoolName) +} + +func testFilterSliceWatchByPoolNameFiltersByPoolName(t *testing.T) { + poolName := "pool-a" + + source := watch.NewFakeWithOptions(watch.FakeOptions{ChannelSize: 3}) + defer source.Stop() + + wrapped := filterSliceWatchByPoolName(t.Context(), source, poolName) + defer wrapped.Stop() + + source.Add(resourceSliceForPool("pool-a", "slice-a")) + source.Add(resourceSliceForPool("pool-b", "slice-b")) + source.Add(resourceSliceForPool("pool-a", "slice-a2")) + + event1 := <-wrapped.ResultChan() + require.Equal(t, watch.Added, event1.Type) + require.Equal(t, "slice-a", event1.Object.(*resourceapi.ResourceSlice).Name) + require.Equal(t, poolName, event1.Object.(*resourceapi.ResourceSlice).Spec.Pool.Name) + + event2 := <-wrapped.ResultChan() + require.Equal(t, watch.Added, event2.Type) + require.Equal(t, "slice-a2", event2.Object.(*resourceapi.ResourceSlice).Name) +} + +func TestWrapWatcher_NilMatchPassesAll(t *testing.T) { + synctest.Test(t, testWrapWatcherNilMatchPassesAll) +} + +func testWrapWatcherNilMatchPassesAll(t *testing.T) { + source := watch.NewFakeWithOptions(watch.FakeOptions{ChannelSize: 2}) + defer source.Stop() + + wrapped := newWrapWatcher(t.Context(), source, nil) + defer wrapped.Stop() + + source.Add(resourceSliceForPool("pool-a", "slice-a")) + source.Add(resourceSliceForPool("pool-b", "slice-b")) + + event1 := <-wrapped.ResultChan() + require.Equal(t, "slice-a", event1.Object.(*resourceapi.ResourceSlice).Name) + + event2 := <-wrapped.ResultChan() + require.Equal(t, "slice-b", event2.Object.(*resourceapi.ResourceSlice).Name) +} + +func TestWrapWatcher_StopClosesResultChan(t *testing.T) { + synctest.Test(t, testWrapWatcherStopClosesResultChan) +} + +func testWrapWatcherStopClosesResultChan(t *testing.T) { + source := watch.NewFakeWithOptions(watch.FakeOptions{ChannelSize: 1}) + wrapped := newWrapWatcher(t.Context(), source, nil) + defer wrapped.Stop() + + source.Add(resourceSliceForPool("pool-a", "slice-a")) + + _, ok := <-wrapped.ResultChan() + require.True(t, ok) + + wrapped.Stop() + + _, ok = <-wrapped.ResultChan() + require.False(t, ok) +} + +func resourceSliceForPool(poolName, name string) *resourceapi.ResourceSlice { + return &resourceapi.ResourceSlice{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: resourceapi.ResourceSliceSpec{ + Driver: "test-driver", + Pool: resourceapi.ResourcePool{Name: poolName}, + }, + } +} diff --git a/test/integration/dra/core.go b/test/integration/dra/core.go index 835225654f2..50edadf73cc 100644 --- a/test/integration/dra/core.go +++ b/test/integration/dra/core.go @@ -231,6 +231,7 @@ func testPublishResourceSlices(tCtx ktesting.TContext, haveLatestAPI bool, disab }, }, }, + AllNodes: true, }, }, } diff --git a/test/integration/dra/resourceslicecontroller.go b/test/integration/dra/resourceslicecontroller.go index 53903ae8633..39f0879f633 100644 --- a/test/integration/dra/resourceslicecontroller.go +++ b/test/integration/dra/resourceslicecontroller.go @@ -42,7 +42,8 @@ func TestCreateResourceSlices(tCtx ktesting.TContext, numSlices int) { domain := strings.Repeat("x", resourceapi.DeviceMaxDomainLength-len(domainSuffix)) + domainSuffix stringValue := strings.Repeat("v", resourceapi.DeviceAttributeMaxValueLength) pool := resourceslice.Pool{ - Slices: make([]resourceslice.Slice, numSlices), + Slices: make([]resourceslice.Slice, numSlices), + AllNodes: true, } numDevices := 0 for i := range numSlices { @@ -115,7 +116,7 @@ func TestCreateResourceSlices(tCtx ktesting.TContext, numSlices int) { // Ask the controller to delete all slices except for one empty slice. tCtx.Log("Deleting slices") resources = resources.DeepCopy() - resources.Pools[poolName] = resourceslice.Pool{Slices: []resourceslice.Slice{{}}} + resources.Pools[poolName] = resourceslice.Pool{Slices: []resourceslice.Slice{{}}, AllNodes: true} controller.Update(resources) // A slice should be updated to be empty, and the rest should be deleted.