Merge pull request #137365 from yaroslavborbat/dra-resourceslice-reconcile-only-pool-name

DRA: add ReconcileOnlyPoolName to ResourceSlice controller

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

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(&currentSlice.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
+	})
+}

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

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

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

test/integration/dra/core.go

@@ -231,6 +231,7 @@ func testPublishResourceSlices(tCtx ktesting.TContext, haveLatestAPI bool, disab
 						},
 					},
 				},
+				AllNodes: true,
 			},
 		},
 	}

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.