From 39fefb9634df58164b85f820003d68be23e6953b Mon Sep 17 00:00:00 2001 From: Alvaro Aleman Date: Wed, 5 Mar 2025 21:05:54 -0500 Subject: [PATCH] bug: Fakeclient: Fix dataraces when writing to the scheme We have a scheme write lock but plenty of other codeptaths that read from the scheme and that don't do looking, resulting in dataraces if the two happen in parallel. This change introduces a simple RW lock and makes the fakeclient acquire read locking for all its operations except when needing the write lock. This isn't particularly smart, but given that we only have one codepath that writes to the scheme, it seems good enough. --- pkg/client/fake/client.go | 24 ++++++++-- pkg/client/fake/client_test.go | 88 ++++++++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 4 deletions(-) diff --git a/pkg/client/fake/client.go b/pkg/client/fake/client.go index 0c4300d548..793219c723 100644 --- a/pkg/client/fake/client.go +++ b/pkg/client/fake/client.go @@ -75,8 +75,8 @@ type fakeClient struct { trackerWriteLock sync.Mutex tracker versionedTracker - schemeWriteLock sync.Mutex - scheme *runtime.Scheme + schemeLock sync.RWMutex + scheme *runtime.Scheme restMapper meta.RESTMapper withStatusSubresource sets.Set[schema.GroupVersionKind] @@ -512,6 +512,8 @@ func (t versionedTracker) updateObject(gvr schema.GroupVersionResource, obj runt } func (c *fakeClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + c.schemeLock.RLock() + defer c.schemeLock.RUnlock() gvr, err := getGVRFromObject(obj, c.scheme) if err != nil { return err @@ -561,6 +563,8 @@ func (c *fakeClient) Watch(ctx context.Context, list client.ObjectList, opts ... } func (c *fakeClient) List(ctx context.Context, obj client.ObjectList, opts ...client.ListOption) error { + c.schemeLock.RLock() + defer c.schemeLock.RUnlock() gvk, err := apiutil.GVKForObject(obj, c.scheme) if err != nil { return err @@ -573,9 +577,11 @@ func (c *fakeClient) List(ctx context.Context, obj client.ObjectList, opts ...cl if _, isUnstructuredList := obj.(runtime.Unstructured); isUnstructuredList && !c.scheme.Recognizes(gvk) { // We need to register the ListKind with UnstructuredList: // https://github.com/kubernetes/kubernetes/blob/7b2776b89fb1be28d4e9203bdeec079be903c103/staging/src/k8s.io/client-go/dynamic/fake/simple.go#L44-L51 - c.schemeWriteLock.Lock() + c.schemeLock.RUnlock() + c.schemeLock.Lock() c.scheme.AddKnownTypeWithName(gvk.GroupVersion().WithKind(gvk.Kind+"List"), &unstructured.UnstructuredList{}) - c.schemeWriteLock.Unlock() + c.schemeLock.Unlock() + c.schemeLock.RLock() } listOpts := client.ListOptions{} @@ -726,6 +732,8 @@ func (c *fakeClient) IsObjectNamespaced(obj runtime.Object) (bool, error) { } func (c *fakeClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { + c.schemeLock.RLock() + defer c.schemeLock.RUnlock() createOptions := &client.CreateOptions{} createOptions.ApplyOptions(opts) @@ -762,6 +770,8 @@ func (c *fakeClient) Create(ctx context.Context, obj client.Object, opts ...clie } func (c *fakeClient) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error { + c.schemeLock.RLock() + defer c.schemeLock.RUnlock() gvr, err := getGVRFromObject(obj, c.scheme) if err != nil { return err @@ -807,6 +817,8 @@ func (c *fakeClient) Delete(ctx context.Context, obj client.Object, opts ...clie } func (c *fakeClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error { + c.schemeLock.RLock() + defer c.schemeLock.RUnlock() gvk, err := apiutil.GVKForObject(obj, c.scheme) if err != nil { return err @@ -856,6 +868,8 @@ func (c *fakeClient) Update(ctx context.Context, obj client.Object, opts ...clie } func (c *fakeClient) update(obj client.Object, isStatus bool, opts ...client.UpdateOption) error { + c.schemeLock.RLock() + defer c.schemeLock.RUnlock() updateOptions := &client.UpdateOptions{} updateOptions.ApplyOptions(opts) @@ -884,6 +898,8 @@ func (c *fakeClient) Patch(ctx context.Context, obj client.Object, patch client. } func (c *fakeClient) patch(obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + c.schemeLock.RLock() + defer c.schemeLock.RUnlock() patchOptions := &client.PatchOptions{} patchOptions.ApplyOptions(opts) diff --git a/pkg/client/fake/client_test.go b/pkg/client/fake/client_test.go index db768cca37..f6a493f54d 100644 --- a/pkg/client/fake/client_test.go +++ b/pkg/client/fake/client_test.go @@ -2516,6 +2516,93 @@ var _ = Describe("Fake client", func() { Expect(cl.SubResource(subResourceScale).Update(context.Background(), obj, client.WithSubResourceBody(scale)).Error()).To(Equal(expectedErr)) }) + It("is threadsafe", func() { + cl := NewClientBuilder().Build() + + u := func() *unstructured.Unstructured { + u := &unstructured.Unstructured{} + u.SetAPIVersion("custom/v1") + u.SetKind("Version") + u.SetName("foo") + return u + } + + uList := func() *unstructured.UnstructuredList { + u := &unstructured.UnstructuredList{} + u.SetAPIVersion("custom/v1") + u.SetKind("Version") + + return u + } + + meta := func() *metav1.PartialObjectMetadata { + return &metav1.PartialObjectMetadata{ + ObjectMeta: metav1.ObjectMeta{ + Name: "foo", + Namespace: "default", + }, + TypeMeta: metav1.TypeMeta{ + APIVersion: "custom/v1", + Kind: "Version", + }, + } + } + metaList := func() *metav1.PartialObjectMetadataList { + return &metav1.PartialObjectMetadataList{ + TypeMeta: metav1.TypeMeta{ + + APIVersion: "custom/v1", + Kind: "Version", + }, + } + } + + pod := func() *corev1.Pod { + return &corev1.Pod{ObjectMeta: metav1.ObjectMeta{ + Name: "foo", + Namespace: "default", + }} + } + + ctx := context.Background() + ops := []func(){ + func() { _ = cl.Create(ctx, u()) }, + func() { _ = cl.Get(ctx, client.ObjectKeyFromObject(u()), u()) }, + func() { _ = cl.Update(ctx, u()) }, + func() { _ = cl.Patch(ctx, u(), client.RawPatch(types.StrategicMergePatchType, []byte("foo"))) }, + func() { _ = cl.Delete(ctx, u()) }, + func() { _ = cl.DeleteAllOf(ctx, u(), client.HasLabels{"foo"}) }, + func() { _ = cl.List(ctx, uList()) }, + + func() { _ = cl.Create(ctx, meta()) }, + func() { _ = cl.Get(ctx, client.ObjectKeyFromObject(meta()), meta()) }, + func() { _ = cl.Update(ctx, meta()) }, + func() { _ = cl.Patch(ctx, meta(), client.RawPatch(types.StrategicMergePatchType, []byte("foo"))) }, + func() { _ = cl.Delete(ctx, meta()) }, + func() { _ = cl.DeleteAllOf(ctx, meta(), client.HasLabels{"foo"}) }, + func() { _ = cl.List(ctx, metaList()) }, + + func() { _ = cl.Create(ctx, pod()) }, + func() { _ = cl.Get(ctx, client.ObjectKeyFromObject(pod()), pod()) }, + func() { _ = cl.Update(ctx, pod()) }, + func() { _ = cl.Patch(ctx, pod(), client.RawPatch(types.StrategicMergePatchType, []byte("foo"))) }, + func() { _ = cl.Delete(ctx, pod()) }, + func() { _ = cl.DeleteAllOf(ctx, pod(), client.HasLabels{"foo"}) }, + func() { _ = cl.List(ctx, &corev1.PodList{}) }, + } + + wg := sync.WaitGroup{} + wg.Add(len(ops)) + for _, op := range ops { + go func() { + defer wg.Done() + op() + }() + } + + wg.Wait() + }) + scalableObjs := []client.Object{ &appsv1.Deployment{ ObjectMeta: metav1.ObjectMeta{ @@ -2604,6 +2691,7 @@ var _ = Describe("Fake client", func() { scaleExpected.ResourceVersion = scaleActual.ResourceVersion Expect(cmp.Diff(scaleExpected, scaleActual)).To(BeEmpty()) }) + } })