From 2e966244ed88653fc351fb5e771092a2191189fe Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Tue, 1 Jul 2025 15:38:05 +0200 Subject: [PATCH] DRA resourceslice controller: fix recreation after quick delete If a ResourceSlice got published by the ResourceSlice controller in a DRA driver and then that ResourceSlice got deleted quickly (within one minute, the mutation cache TTL) by someone (for example, the kubelet because of a restart), then the controller did not react properly to the deletion unless some other event triggered the syncing of the pool. Found while adding upgrade/downgrade tests with a driver which keeps running across the upgrade/downgrade. The exact sequence leading to this were: - controller adds ResourceSlice, schedules a sync for one minute in the future (the TTL) - someone else deletes the ResourceSlice - add and delete events schedule another sync 30 seconds in the future (the delay), *overwriting* the other scheduled sync - sync runs once, finds deleted slices in the mutation cache, does not re-create them, and also does not run again One possible fix would be to set a resync period. But then work is done periodically, whether it's necessary or not. Another fix is to ensure that the TTL is shorter than the delay. Then when a sync occurs, all locally stored additional slices are expired. But that renders the whole storing of recently created slices in the cache pointless. So the fix used here is to keep track of when another sync has to run because of added slices. At the end of each sync, the next sync gets scheduled if (and only if) needed, until eventually syncing can stop. --- .../resourceslice/resourceslicecontroller.go | 69 +++- test/integration/dra/dra_test.go | 324 ++++++++++++++---- 2 files changed, 316 insertions(+), 77 deletions(-) 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)) +}