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 a9d213a0d38..c37c61fd096 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/resourceslice/resourceslicecontroller.go @@ -59,14 +59,14 @@ const ( // Then the obsolete slice remains in the mutation cache. // // To mitigate this, we use a TTL and check a pool again once added slices expire. - defaultMutationCacheTTL = time.Minute + DefaultMutationCacheTTL = time.Minute - // defaultSyncDelay defines how long to wait between receiving the most recent + // DefaultSyncDelay defines how long to wait between receiving the most recent // informer event and syncing again. This is long enough that the informer cache - // should be up-to-date (matter mostly for deletes because an out-dated cache + // should be up-to-date (matters mostly for deletes because an out-dated cache // causes redundant delete API calls) and not too long that a human mistake // doesn't get fixed while that human is waiting for it. - defaultSyncDelay = 30 * time.Second + DefaultSyncDelay = 30 * time.Second ) // Controller synchronizes information about resources of one driver with @@ -87,6 +87,21 @@ type Controller struct { syncDelay time.Duration errorHandler func(ctx context.Context, err error, msg string) + // Last time that a ResourceSlice of a pool was created. + // At that time + cache mutation TTL do we have to sync again + // because the locally cached slice might have stayed in the + // cache erronously (not removed on delete by someone else) + // and we have to check again once it has been removed from + // the cache. + // + // It's not sufficient to schedule a delayed sync because + // another sync scheduled by an event overwrites the older + // one, so we would sync too soon and then not again. + // + // The key is the pool name. This makes each time entry + // unique for syncPool calls for the pool. + lastAddByPool map[string]time.Time + // Must use atomic access... numCreates int64 numUpdates int64 @@ -368,9 +383,10 @@ func newController(ctx context.Context, options Options) (*Controller, error) { driverName: options.DriverName, owner: options.Owner.DeepCopy(), queue: options.Queue, - mutationCacheTTL: ptr.Deref(options.MutationCacheTTL, defaultMutationCacheTTL), - syncDelay: ptr.Deref(options.SyncDelay, defaultSyncDelay), + mutationCacheTTL: ptr.Deref(options.MutationCacheTTL, DefaultMutationCacheTTL), + syncDelay: ptr.Deref(options.SyncDelay, DefaultSyncDelay), errorHandler: options.ErrorHandler, + lastAddByPool: make(map[string]time.Time), } if c.queue == nil { c.queue = workqueue.NewTypedRateLimitingQueueWithConfig( @@ -444,6 +460,7 @@ func (c *Controller) initInformer(ctx context.Context) error { } logger.V(5).Info("ResourceSlice add", "slice", klog.KObj(slice)) c.queue.AddAfter(slice.Spec.Pool.Name, c.syncDelay) + logger.V(5).Info("Scheduled sync", "poolName", slice.Spec.Pool.Name, "at", time.Now().Add(c.syncDelay)) }, UpdateFunc: func(old, new any) { oldSlice, ok := old.(*resourceapi.ResourceSlice) @@ -460,7 +477,11 @@ func (c *Controller) initInformer(ctx context.Context) error { logger.V(5).Info("ResourceSlice update", "slice", klog.KObj(newSlice)) } c.queue.AddAfter(oldSlice.Spec.Pool.Name, c.syncDelay) - c.queue.AddAfter(newSlice.Spec.Pool.Name, c.syncDelay) + logger.V(5).Info("Scheduled sync", "pool", oldSlice.Spec.Pool.Name, "at", time.Now().Add(c.syncDelay)) + if oldSlice.Spec.Pool.Name != newSlice.Spec.Pool.Name { + c.queue.AddAfter(newSlice.Spec.Pool.Name, c.syncDelay) + logger.V(5).Info("Scheduled sync", "poolName", newSlice.Spec.Pool.Name, "at", time.Now().Add(c.syncDelay)) + } }, DeleteFunc: func(obj any) { if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { @@ -472,6 +493,7 @@ func (c *Controller) initInformer(ctx context.Context) error { } logger.V(5).Info("ResourceSlice delete", "slice", klog.KObj(slice)) c.queue.AddAfter(slice.Spec.Pool.Name, c.syncDelay) + logger.V(5).Info("Scheduled sync", "poolName", slice.Spec.Pool.Name, "at", time.Now().Add(c.syncDelay)) }, }) if err != nil { @@ -530,6 +552,7 @@ func (c *Controller) processNextWorkItem(ctx context.Context) bool { // be updated at any time by the user of the controller. func (c *Controller) syncPool(ctx context.Context, poolName string) error { logger := klog.FromContext(ctx) + start := time.Now() // Gather information about the actual and desired state. var slices []*resourceapi.ResourceSlice @@ -724,11 +747,11 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error { // Preserve TimeAdded from existing device, if there is a matching device and taint. slice.Spec.Devices = copyTaintTimeAdded(slice.Spec.Devices, pool.Slices[i].Devices) - logger.V(5).Info("Updating existing resource slice", "slice", klog.KObj(slice)) actualSlice, err := c.resourceClient.ResourceSlices().Update(ctx, slice, metav1.UpdateOptions{}) if err != nil { return fmt.Errorf("update resource slice: %w", err) } + logger.V(5).Info("Updated existing resource slice", "slice", klog.KObj(slice)) atomic.AddInt64(&c.numUpdates, 1) c.sliceStored(ctx, "update ResourceSlice", poolName, pool, i, slice, actualSlice) } @@ -782,19 +805,39 @@ func (c *Controller) syncPool(ctx context.Context, poolName string) error { // // Using a https://pkg.go.dev/k8s.io/client-go/tools/cache#MutationCache // avoids that. - logger.V(5).Info("Creating new resource slice") actualSlice, err := c.resourceClient.ResourceSlices().Create(ctx, slice, metav1.CreateOptions{}) if err != nil { return fmt.Errorf("create resource slice: %w", err) } + logger.V(5).Info("Created new resource slice", "slice", klog.KObj(actualSlice)) atomic.AddInt64(&c.numCreates, 1) added = true c.sliceStored(ctx, "create ResourceSlice", poolName, pool, i, slice, actualSlice) } + + now := time.Now() if added { - // Check that the recently added slice(s) really exist even - // after they expired from the mutation cache. - c.queue.AddAfter(poolName, c.mutationCacheTTL) + c.lastAddByPool[poolName] = now + logger.V(5).Info("Added slices") + } else if lastAdd, ok := c.lastAddByPool[poolName]; ok && start.After(lastAdd.Add(c.mutationCacheTTL)) { + // This sync started after the last add expired from the cache, + // so we are done and don't need to check again. + delete(c.lastAddByPool, poolName) + logger.V(5).Info("Done with re-syncing") + } + if lastAdd, ok := c.lastAddByPool[poolName]; ok { + // Need to check again. + // + // Scheduling the resync races with scheduling them in informer events, but that's okay: + // what matters is that we sync at all at some point. + // + // lastAdd was taked by time.Now() above and thus is slightly higher or equal + // to the time taken by the mutation cache when the slice was added, so we + // can be sure that any sync running at this time will not see the added + // slice because it will be expired. + when := lastAdd.Add(c.mutationCacheTTL) + c.queue.AddAfter(poolName, when.Sub(now)) + logger.V(5).Info("Scheduled re-sync", "at", when) } return nil @@ -818,10 +861,10 @@ func (c *Controller) removeSlices(ctx context.Context, slices []*resourceapi.Res // If this happens, we get a "not found error" and nothing // changes on the server. The only downside is the extra API // call. This isn't as bad as extra creates. - logger.V(5).Info("Deleting obsolete resource slice", "slice", klog.KObj(slice), "deleteOptions", options) err := c.resourceClient.ResourceSlices().Delete(ctx, slice.Name, options) switch { case err == nil: + logger.V(5).Info("Deleted obsolete resource slice", "slice", klog.KObj(slice), "deleteOptions", options) atomic.AddInt64(&c.numDeletes, 1) case apierrors.IsNotFound(err): logger.V(5).Info("Resource slice was already deleted earlier", "slice", klog.KObj(slice)) diff --git a/test/integration/dra/dra_test.go b/test/integration/dra/dra_test.go index 5a2074397fa..38a77a52f1b 100644 --- a/test/integration/dra/dra_test.go +++ b/test/integration/dra/dra_test.go @@ -33,6 +33,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/onsi/gomega" "github.com/onsi/gomega/gstruct" + gtypes "github.com/onsi/gomega/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -213,7 +214,7 @@ func TestDRA(t *testing.T) { tCtx.Run("PrioritizedList", func(tCtx ktesting.TContext) { testPrioritizedList(tCtx, false) }) tCtx.Run("Pod", func(tCtx ktesting.TContext) { testPod(tCtx, true) }) tCtx.Run("PublishResourceSlices", func(tCtx ktesting.TContext) { - testPublishResourceSlices(tCtx, features.DRADeviceTaints, features.DRAPartitionableDevices) + testPublishResourceSlices(tCtx, true, features.DRADeviceTaints, features.DRAPartitionableDevices) }) tCtx.Run("ResourceClaimDeviceStatus", func(tCtx ktesting.TContext) { testResourceClaimDeviceStatus(tCtx, false) }) }, @@ -229,7 +230,7 @@ func TestDRA(t *testing.T) { tCtx.Run("PrioritizedList", func(tCtx ktesting.TContext) { testPrioritizedList(tCtx, false) }) tCtx.Run("Pod", func(tCtx ktesting.TContext) { testPod(tCtx, true) }) tCtx.Run("PublishResourceSlices", func(tCtx ktesting.TContext) { - testPublishResourceSlices(tCtx, features.DRADeviceTaints, features.DRAPartitionableDevices) + testPublishResourceSlices(tCtx, true, features.DRADeviceTaints, features.DRAPartitionableDevices) }) tCtx.Run("ResourceClaimDeviceStatus", func(tCtx ktesting.TContext) { testResourceClaimDeviceStatus(tCtx, true) }) }, @@ -241,7 +242,7 @@ func TestDRA(t *testing.T) { features: map[featuregate.Feature]bool{features.DynamicResourceAllocation: true}, f: func(tCtx ktesting.TContext) { tCtx.Run("PublishResourceSlices", func(tCtx ktesting.TContext) { - testPublishResourceSlices(tCtx, features.DRADeviceTaints, features.DRAPartitionableDevices) + testPublishResourceSlices(tCtx, false, features.DRADeviceTaints, features.DRAPartitionableDevices) }) }, }, @@ -252,7 +253,7 @@ func TestDRA(t *testing.T) { features: map[featuregate.Feature]bool{features.DynamicResourceAllocation: true}, f: func(tCtx ktesting.TContext) { tCtx.Run("PublishResourceSlices", func(tCtx ktesting.TContext) { - testPublishResourceSlices(tCtx, features.DRADeviceTaints, features.DRAPartitionableDevices) + testPublishResourceSlices(tCtx, true, features.DRADeviceTaints, features.DRAPartitionableDevices) }) }, }, @@ -276,7 +277,7 @@ func TestDRA(t *testing.T) { tCtx.Run("AdminAccess", func(tCtx ktesting.TContext) { testAdminAccess(tCtx, true) }) tCtx.Run("Convert", testConvert) tCtx.Run("PrioritizedList", func(tCtx ktesting.TContext) { testPrioritizedList(tCtx, true) }) - tCtx.Run("PublishResourceSlices", func(tCtx ktesting.TContext) { testPublishResourceSlices(tCtx) }) + tCtx.Run("PublishResourceSlices", func(tCtx ktesting.TContext) { testPublishResourceSlices(tCtx, true) }) tCtx.Run("ResourceClaimDeviceStatus", func(tCtx ktesting.TContext) { testResourceClaimDeviceStatus(tCtx, true) }) tCtx.Run("MaxResourceSlice", testMaxResourceSlice) }, @@ -573,12 +574,14 @@ func testPrioritizedList(tCtx ktesting.TContext, enabled bool) { }).WithTimeout(10 * time.Second).WithPolling(time.Second).Should(schedulingAttempted) } -func testPublishResourceSlices(tCtx ktesting.TContext, disabledFeatures ...featuregate.Feature) { +func testPublishResourceSlices(tCtx ktesting.TContext, haveLatestAPI bool, disabledFeatures ...featuregate.Feature) { tCtx.Parallel() - tCtx = ktesting.WithTimeout(tCtx, 30*time.Second, "test timed out") namespace := createTestNamespace(tCtx, nil) driverName := namespace + ".example.com" + listDriverSlices := metav1.ListOptions{ + FieldSelector: resourcev1beta2.ResourceSliceSelectorDriver + "=" + driverName, + } poolName := "global" resources := &resourceslice.DriverResources{ Pools: map[string]resourceslice.Pool{ @@ -614,7 +617,7 @@ func testPublishResourceSlices(tCtx ktesting.TContext, disabledFeatures ...featu Key: "dra.example.com/taint", Value: "taint-value", Effect: resourcev1beta2.DeviceTaintEffectNoExecute, - TimeAdded: ptr.To(metav1.Now()), + TimeAdded: ptr.To(metav1.Time{Time: time.Now().Truncate(time.Second)}), }}, }, { @@ -632,68 +635,254 @@ func testPublishResourceSlices(tCtx ktesting.TContext, disabledFeatures ...featu }, }, } + + // Manually turn into the expected slices, considering that some fields get dropped. + expectedResources := resources.DeepCopy() + var expectedSliceSpecs []resourcev1beta2.ResourceSliceSpec + for _, sl := range expectedResources.Pools[poolName].Slices { + expectedSliceSpecs = append(expectedSliceSpecs, resourcev1beta2.ResourceSliceSpec{ + Driver: driverName, + Pool: resourcev1beta2.ResourcePool{ + Name: poolName, + ResourceSliceCount: int64(len(expectedResources.Pools[poolName].Slices)), + }, + AllNodes: ptr.To(true), + SharedCounters: sl.SharedCounters, + Devices: sl.Devices, + }) + } + for _, disabled := range disabledFeatures { + switch disabled { + case features.DRADeviceTaints: + for i := range expectedSliceSpecs { + for e := range expectedSliceSpecs[i].Devices { + expectedSliceSpecs[i].Devices[e].Taints = nil + } + } + case features.DRAPartitionableDevices: + for i := range expectedSliceSpecs { + expectedSliceSpecs[i].SharedCounters = nil + for e := range expectedSliceSpecs[i].Devices { + expectedSliceSpecs[i].Devices[e].ConsumesCounters = nil + } + } + default: + tCtx.Fatalf("faulty test, case for %s missing", disabled) + } + } + var expectedSlices []any + for _, spec := range expectedSliceSpecs { + // The matcher is precise and matches all fields, except for those few which are known to + // be not exactly as sent by the client. New fields have to be added when extending the API. + expectedSlices = append(expectedSlices, gomega.HaveField("Spec", gstruct.MatchAllFields(gstruct.Fields{ + "Driver": gomega.Equal(driverName), + "Pool": gstruct.MatchAllFields(gstruct.Fields{ + "Name": gomega.Equal(poolName), + "Generation": gomega.BeNumerically(">=", int64(1)), + "ResourceSliceCount": gomega.Equal(int64(len(expectedResources.Pools[poolName].Slices))), + }), + "NodeName": matchPointer(spec.NodeName), + "NodeSelector": matchPointer(spec.NodeSelector), + "AllNodes": gstruct.PointTo(gomega.BeTrue()), + "Devices": gomega.HaveExactElements(func() []any { + var expected []any + for _, device := range spec.Devices { + expected = append(expected, gstruct.MatchAllFields(gstruct.Fields{ + "Name": gomega.Equal(device.Name), + "Attributes": gomega.Equal(device.Attributes), + "Capacity": gomega.Equal(device.Capacity), + "ConsumesCounters": gomega.Equal(device.ConsumesCounters), + "NodeName": matchPointer(device.NodeName), + "NodeSelector": matchPointer(device.NodeSelector), + "AllNodes": matchPointer(device.AllNodes), + "Taints": gomega.HaveExactElements(func() []any { + var expected []any + for _, taint := range device.Taints { + if taint.TimeAdded != nil { + // Can do exact match. + expected = append(expected, gomega.Equal(taint)) + } else { + // Ignore TimeAdded value. + expected = append(expected, gstruct.MatchAllFields(gstruct.Fields{ + "Key": gomega.Equal(taint.Key), + "Value": gomega.Equal(taint.Value), + "Effect": gomega.Equal(taint.Effect), + "TimeAdded": gomega.Not(gomega.BeNil()), + })) + } + } + return expected + }()...), + })) + } + return expected + }()...), + "PerDeviceNodeSelection": matchPointer(spec.PerDeviceNodeSelection), + "SharedCounters": gomega.Equal(spec.SharedCounters), + }))) + } + + expectSlices := func(tCtx ktesting.TContext) { + tCtx.Helper() + + if !haveLatestAPI { + return + } + slices, err := tCtx.Client().ResourceV1beta2().ResourceSlices().List(tCtx, listDriverSlices) + tCtx.ExpectNoError(err, "list slices") + gomega.NewGomegaWithT(tCtx).Expect(slices.Items).Should(gomega.ConsistOf(expectedSlices...)) + } + + deleteSlices := func(tCtx ktesting.TContext) { + tCtx.Helper() + + // At least one of the APIs must be enabled... + var err error + err = tCtx.Client().ResourceV1beta1().ResourceSlices().DeleteCollection(tCtx, metav1.DeleteOptions{}, listDriverSlices) + if err == nil { + return + } + err = tCtx.Client().ResourceV1beta2().ResourceSlices().DeleteCollection(tCtx, metav1.DeleteOptions{}, listDriverSlices) + if err == nil { + return + } + tCtx.ExpectNoError(err, "delete slices") + } + + // Speed up testing a bit... + factor := time.Duration(10) + mutationCacheTTL := resourceslice.DefaultMutationCacheTTL / factor + syncDelay := resourceslice.DefaultSyncDelay / factor + quiesencePeriod := syncDelay + if mutationCacheTTL > quiesencePeriod { + quiesencePeriod = mutationCacheTTL + } + quiesencePeriod += 10 * time.Second + var gotDroppedFieldError atomic.Bool var gotValidationError atomic.Bool var validationErrorsOkay atomic.Bool - opts := resourceslice.Options{ - DriverName: driverName, - KubeClient: tCtx.Client(), - SyncDelay: ptr.To(0 * time.Second), - Resources: resources, - ErrorHandler: func(ctx context.Context, err error, msg string) { - klog.FromContext(ctx).Info("ErrorHandler called", "err", err, "msg", msg) - if !validationErrorsOkay.Load() && len(disabledFeatures) == 0 { - assert.NoError(tCtx, err, msg) - return - } - var droppedFields *resourceslice.DroppedFieldsError - if errors.As(err, &droppedFields) { - var disabled []string - for _, feature := range disabledFeatures { - disabled = append(disabled, string(feature)) + setup := func(tCtx ktesting.TContext) (*resourceslice.Controller, func(tCtx ktesting.TContext) resourceslice.Stats, resourceslice.Stats) { + tCtx.Helper() + + tCtx.CleanupCtx(deleteSlices) + + gotDroppedFieldError.Store(false) + gotValidationError.Store(false) + validationErrorsOkay.Store(false) + + opts := resourceslice.Options{ + DriverName: driverName, + KubeClient: tCtx.Client(), + Resources: resources, + MutationCacheTTL: &mutationCacheTTL, + SyncDelay: &syncDelay, + ErrorHandler: func(ctx context.Context, err error, msg string) { + klog.FromContext(ctx).Info("ErrorHandler called", "err", err, "msg", msg) + if !validationErrorsOkay.Load() && len(disabledFeatures) == 0 { + assert.NoError(tCtx, err, msg) + return } - assert.ErrorContains(tCtx, err, fmt.Sprintf("pool %q, slice #1: some fields were dropped by the apiserver, probably because these features are disabled: %s", poolName, strings.Join(disabled, " "))) - gotDroppedFieldError.Store(true) - } else if validationErrorsOkay.Load() && apierrors.IsInvalid(err) { - gotValidationError.Store(true) - } else { - tCtx.Errorf("unexpected error: %v", err) + + var droppedFields *resourceslice.DroppedFieldsError + if errors.As(err, &droppedFields) { + var disabled []string + for _, feature := range disabledFeatures { + disabled = append(disabled, string(feature)) + } + assert.ErrorContains(tCtx, err, fmt.Sprintf("pool %q, slice #1: some fields were dropped by the apiserver, probably because these features are disabled: %s", poolName, strings.Join(disabled, " "))) + gotDroppedFieldError.Store(true) + } else if validationErrorsOkay.Load() && apierrors.IsInvalid(err) { + gotValidationError.Store(true) + } else { + tCtx.Errorf("unexpected error: %v", err) + } + }, + } + + controller, err := resourceslice.StartController(tCtx, opts) + tCtx.ExpectNoError(err, "start controller") + tCtx.Cleanup(controller.Stop) + + expectedStats := resourceslice.Stats{ + NumCreates: int64(len(expectedSlices)), + } + getStats := func(tCtx ktesting.TContext) resourceslice.Stats { + return controller.GetStats() + } + ktesting.Eventually(tCtx, getStats).WithTimeout(syncDelay + 5*time.Second).Should(gomega.Equal(expectedStats)) + expectSlices(tCtx) + + return controller, getStats, expectedStats + } + + // Each sub-test starts with no slices and must clean up after itself. + + tCtx.Run("create", func(tCtx ktesting.TContext) { + controller, getStats, expectedStats := setup(tCtx) + + // No further changes necessary. + ktesting.Consistently(tCtx, getStats).WithTimeout(quiesencePeriod).Should(gomega.Equal(expectedStats)) + + if len(disabledFeatures) > 0 && !gotDroppedFieldError.Load() { + tCtx.Error("expected dropped fields error, got none") + } + + // Now switch to one invalid slice. + resources := resources.DeepCopy() + pool := resources.Pools[poolName] + pool.Slices = pool.Slices[:1] + pool.Slices[0].Devices[0].Attributes = map[resourcev1beta2.QualifiedName]resourcev1beta2.DeviceAttribute{"empty": {}} + resources.Pools[poolName] = pool + validationErrorsOkay.Store(true) + controller.Update(resources) + ktesting.Eventually(tCtx, getStats).WithTimeout(10*time.Second).Should(gomega.HaveField("NumDeletes", gomega.BeNumerically(">=", int64(1))), "Slice should have been removed.") + ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) bool { + return gotValidationError.Load() + }).WithTimeout(time.Minute).Should(gomega.BeTrueBecause("Should have gotten another error because the slice is invalid.")) + }) + + if !haveLatestAPI { + return + } + + tCtx.Run("recreate-after-delete", func(tCtx ktesting.TContext) { + _, getStats, expectedStats := setup(tCtx) + ktesting.Consistently(tCtx, getStats).WithTimeout(quiesencePeriod).Should(gomega.Equal(expectedStats)) + + // Stress the controller by repeatedly deleting the slices. + // One delete occurs after the sync period is over (because of the Consistently), + // the second before (because it's done as quickly as possible). + for i := 0; i < 2; i++ { + tCtx.Log("deleting ResourceSlices") + tCtx.ExpectNoError(tCtx.Client().ResourceV1beta2().ResourceSlices().DeleteCollection(tCtx, metav1.DeleteOptions{}, listDriverSlices), "delete driver slices") + expectedStats.NumCreates += int64(len(expectedSlices)) + ktesting.Eventually(tCtx, getStats).WithTimeout(syncDelay + 5*time.Second).Should(gomega.Equal(expectedStats)) + expectSlices(tCtx) + } + }) + + tCtx.Run("fix-after-update", func(tCtx ktesting.TContext) { + _, getStats, expectedStats := setup(tCtx) + + // Stress the controller by repeatedly updatings the slices. + for i := 0; i < 2; i++ { + slices, err := tCtx.Client().ResourceV1beta2().ResourceSlices().List(tCtx, listDriverSlices) + tCtx.ExpectNoError(err, "list slices") + for _, slice := range slices.Items { + if slice.Spec.Devices[0].Attributes == nil { + slice.Spec.Devices[0].Attributes = make(map[resourcev1beta2.QualifiedName]resourcev1beta2.DeviceAttribute) + } + slice.Spec.Devices[0].Attributes["someUnwantedAttribute"] = resourcev1beta2.DeviceAttribute{BoolValue: ptr.To(true)} + _, err := tCtx.Client().ResourceV1beta2().ResourceSlices().Update(tCtx, &slice, metav1.UpdateOptions{}) + tCtx.ExpectNoError(err, "update slice") } - }, - } - controller, err := resourceslice.StartController(tCtx, opts) - tCtx.ExpectNoError(err, "start controller") - defer controller.Stop() - - // Two create calls should be all that are needed. - expectedStats := resourceslice.Stats{ - NumCreates: 2, - } - getStats := func(tCtx ktesting.TContext) resourceslice.Stats { - return controller.GetStats() - } - ktesting.Eventually(tCtx, getStats).WithTimeout(10 * time.Second).Should(gomega.Equal(expectedStats)) - - // No further changes necessary. - ktesting.Consistently(tCtx, getStats).WithTimeout(10 * time.Second).Should(gomega.Equal(expectedStats)) - - if len(disabledFeatures) > 0 && !gotDroppedFieldError.Load() { - tCtx.Error("expected dropped fields error, got none") - } - - // Now switch to one invalid slice. - pool := resources.Pools[poolName] - pool.Slices = pool.Slices[:1] - pool.Slices[0].Devices[0].Attributes = map[resourcev1beta2.QualifiedName]resourcev1beta2.DeviceAttribute{"empty": {}} - resources.Pools[poolName] = pool - validationErrorsOkay.Store(true) - controller.Update(resources) - ktesting.Eventually(tCtx, getStats).WithTimeout(10*time.Second).Should(gomega.HaveField("NumDeletes", gomega.BeNumerically(">=", int64(1))), "Slice should have been removed.") - ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) bool { - return gotValidationError.Load() - }).WithTimeout(time.Minute).Should(gomega.BeTrueBecause("Should have gotten another error because the slice is invalid.")) - + expectedStats.NumUpdates += int64(len(expectedSlices)) + ktesting.Eventually(tCtx, getStats).WithTimeout(syncDelay + 5*time.Second).Should(gomega.Equal(expectedStats)) + expectSlices(tCtx) + } + }) } // testResourceClaimDeviceStatus creates a ResourceClaim with an invalid device (not allocated device) @@ -890,3 +1079,10 @@ func testMaxResourceSlice(tCtx ktesting.TContext) { tCtx.Errorf("ResourceSliceSpec got modified during Create (- want, + got):\n%s", diff) } } + +func matchPointer[T any](p *T) gtypes.GomegaMatcher { + if p == nil { + return gomega.BeNil() + } + return gstruct.PointTo(gomega.Equal(*p)) +}