diff --git a/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache.go b/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache.go index f70124fd343..96d27fc4336 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache.go @@ -102,6 +102,7 @@ func (c *ExtendedResourceCache) OnAdd(obj interface{}, isInInitialList bool) { utilruntime.HandleErrorWithLogger(c.logger, nil, "Expected DeviceClass", "actual", fmt.Sprintf("%T", obj)) return } + c.logger.V(5).Info("DeviceClass added", "deviceClass", klog.KObj(deviceClass)) c.updateResourceName2class(deviceClass, nil) c.updateClass2ResourceName(deviceClass) @@ -122,6 +123,7 @@ func (c *ExtendedResourceCache) OnUpdate(oldObj, newObj interface{}) { utilruntime.HandleErrorWithLogger(c.logger, nil, "Expected DeviceClass", "actual", fmt.Sprintf("%T", oldObj)) return } + c.logger.V(5).Info("DeviceClass updated", "deviceClass", klog.KObj(deviceClass)) c.updateResourceName2class(deviceClass, oldDeviceClass) c.updateClass2ResourceName(deviceClass) @@ -140,6 +142,7 @@ func (c *ExtendedResourceCache) OnDelete(obj interface{}) { utilruntime.HandleErrorWithLogger(c.logger, nil, "Expected DeviceClass", "actual", fmt.Sprintf("%T", obj)) return } + c.logger.V(5).Info("DeviceClass deleted", "deviceClass", klog.KObj(deviceClass)) c.removeResourceName2class(deviceClass) c.removeClass2ResourceName(deviceClass) diff --git a/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache_test.go b/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache_test.go index 58171d77d26..61889507f9a 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache_test.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/deviceclass/extendedresourcecache/extendedresourcecache_test.go @@ -19,6 +19,7 @@ package extendedresourcecache import ( "context" "testing" + "testing/synctest" "time" v1 "k8s.io/api/core/v1" @@ -28,6 +29,7 @@ import ( "k8s.io/client-go/kubernetes/fake" "k8s.io/client-go/tools/cache" "k8s.io/klog/v2/ktesting" + _ "k8s.io/klog/v2/ktesting/init" // Add command line flags. "k8s.io/utils/ptr" ) @@ -140,25 +142,9 @@ func TestHandlers(t *testing.T) { erCache.OnDelete(updatedClass) } -func TestExtendedResourceCache(t *testing.T) { - logger, ctx := ktesting.NewTestContext(t) - tCtx, tCancel := context.WithCancel(ctx) - client := fake.NewClientset() - informerFactory := informers.NewSharedInformerFactory(client, 0) - - deviceClassInformer := informerFactory.Resource().V1().DeviceClasses() - cache := NewExtendedResourceCache(logger) - if _, err := deviceClassInformer.Informer().AddEventHandler(cache); err != nil { - logger.Error(err, "Failed to add event handler for device classes") - } - informerFactory.Start(tCtx.Done()) - t.Cleanup(func() { - // Need to cancel before waiting for the shutdown. - tCancel() - // Now we can wait for all goroutines to stop. - informerFactory.Shutdown() - }) - informerFactory.WaitForCacheSync(tCtx.Done()) +func TestExtendedResourceCache(t *testing.T) { synctest.Test(t, testExtendedResourceCache) } +func testExtendedResourceCache(t *testing.T) { + tCtx, client, cache := setup(t) // Test with a device class that has an explicit extended resource name now := time.Now() @@ -226,7 +212,7 @@ func TestExtendedResourceCache(t *testing.T) { if err != nil { t.Fatalf("Failed to create device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() // Verify explicit mapping deviceClass := cache.GetDeviceClass("example.com/gpu") @@ -252,7 +238,7 @@ func TestExtendedResourceCache(t *testing.T) { if err != nil { t.Fatalf("Failed to create device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() // should keep deviceClass1, since it is newer than deviceClass3 deviceClass = cache.GetDeviceClass("example.com/gpu") @@ -265,7 +251,7 @@ func TestExtendedResourceCache(t *testing.T) { if err != nil { t.Fatalf("Failed to create device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() // deviceClass4 replaces deviceClass1, since it is newer with the same example.com/gpu extended resource name deviceClass = cache.GetDeviceClass("example.com/gpu") @@ -279,7 +265,7 @@ func TestExtendedResourceCache(t *testing.T) { if err != nil { t.Fatalf("Failed to create device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() // deviceClass0 replaces deviceClass4, it is created at the same time as deviceClass4, but its name is // alphabetically ordered earlier @@ -295,7 +281,7 @@ func TestExtendedResourceCache(t *testing.T) { if err != nil { t.Fatalf("Failed to update device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() // Should have the new mapping deviceClass = cache.GetDeviceClass("test.com/gpu") @@ -312,7 +298,7 @@ func TestExtendedResourceCache(t *testing.T) { if err != nil { t.Fatalf("Failed to delete device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() deviceClass = cache.GetDeviceClass("test.com/gpu") if deviceClass != nil { @@ -324,24 +310,9 @@ func TestExtendedResourceCache(t *testing.T) { } } -func TestDeviceClassMapping(t *testing.T) { - logger, ctx := ktesting.NewTestContext(t) - tCtx, tCancel := context.WithCancel(ctx) - - client := fake.NewClientset() - informerFactory := informers.NewSharedInformerFactory(client, 0) - cache := NewExtendedResourceCache(logger) - if _, err := informerFactory.Resource().V1().DeviceClasses().Informer().AddEventHandler(cache); err != nil { - logger.Error(err, "failed to add device class informer event handler") - } - informerFactory.Start(tCtx.Done()) - t.Cleanup(func() { - // Need to cancel before waiting for the shutdown. - tCancel() - // Now we can wait for all goroutines to stop. - informerFactory.Shutdown() - }) - informerFactory.WaitForCacheSync(tCtx.Done()) +func TestDeviceClassMapping(t *testing.T) { synctest.Test(t, testDeviceClassMapping) } +func testDeviceClassMapping(t *testing.T) { + tCtx, client, cache := setup(t) deviceClass1 := &resourceapi.DeviceClass{ ObjectMeta: metav1.ObjectMeta{ @@ -368,7 +339,8 @@ func TestDeviceClassMapping(t *testing.T) { t.Fatalf("Failed to create device class: %v", err) } - time.Sleep(1 * time.Second) + // Wait for background goroutines to handle the new classes. + synctest.Wait() name := cache.GetExtendedResource("gpu-class") if name != "example.com/gpu" { t.Errorf("Expected to find device class 'gpu-class', got %s", name) @@ -386,7 +358,7 @@ func TestDeviceClassMapping(t *testing.T) { t.Fatalf("Failed to update device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() name = cache.GetExtendedResource("gpu-class") if name != "my.com/gpu" { t.Errorf("Expected to find device class 'gpu-class' with 'my.com/gpu' after modification, got %s", name) @@ -398,9 +370,49 @@ func TestDeviceClassMapping(t *testing.T) { t.Fatalf("Failed to delete device class: %v", err) } - time.Sleep(1 * time.Second) + synctest.Wait() name = cache.GetExtendedResource("gpu-class") if name != "" { t.Error("Expected 'gpu-class' not found after deletion") } } + +func setup(t *testing.T) (context.Context, *fake.Clientset, *ExtendedResourceCache) { + logger, ctx := ktesting.NewTestContext(t) + ctx, cancel := context.WithCancel(ctx) + t.Cleanup(cancel) + + client := fake.NewClientset() + informerFactory := informers.NewSharedInformerFactory(client, 0) + ec := NewExtendedResourceCache(logger) + handle, err := informerFactory.Resource().V1().DeviceClasses().Informer().AddEventHandler(ec) + if err != nil { + t.Fatalf("failed to add device class informer event handler: %v", err) + } + informerFactory.Start(ctx.Done()) + t.Cleanup(func() { + // Need to cancel before waiting for the shutdown. + cancel() + // Now we can wait for all goroutines to stop. + informerFactory.Shutdown() + }) + informerFactory.WaitForCacheSync(ctx.Done()) + cache.WaitForNamedCacheSyncWithContext(ctx, handle.HasSynced) + + // fake.Clientset suffers from a race condition related to informers: + // it does not implement resource version support in its Watch + // implementation and instead assumes that watches are set up + // before further changes are made. + // + // If a test waits for caches to be synced and then immediately + // adds an object, that new object will never be seen by event handlers + // if the race goes wrong and the Watch call hadn't completed yet + // (can be triggered by adding a sleep before https://github.com/kubernetes/kubernetes/blob/b53b9fb5573323484af9a19cf3f5bfe80760abba/staging/src/k8s.io/client-go/tools/cache/reflector.go#L431). + // + // To work around that, we wait here for the goroutines which + // are involved in setting up the watch *before* creating + // DeviceClasses. + synctest.Wait() + + return ctx, client, ec +}