From fb5ba3acafbce384e410dbaecc0027fefe59afe6 Mon Sep 17 00:00:00 2001 From: Jian Zhu Date: Thu, 5 Jun 2025 09:58:40 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Use=20syncmap=20for=20the=20reso?= =?UTF-8?q?urce=20cache=20(#1023)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Use syncmap for the resource cache Signed-off-by: zhujian * update unit tests Signed-off-by: zhujian * fix unit test Signed-off-by: zhujian * use sync.map directly Signed-off-by: zhujian --------- Signed-off-by: zhujian --- pkg/work/spoke/apply/resource_cache.go | 191 ++++++++++++++++++++ pkg/work/spoke/apply/resource_cache_test.go | 86 +++++++++ pkg/work/spoke/apply/update_apply.go | 4 +- pkg/work/spoke/apply/update_apply_test.go | 3 +- 4 files changed, 280 insertions(+), 4 deletions(-) create mode 100644 pkg/work/spoke/apply/resource_cache.go create mode 100644 pkg/work/spoke/apply/resource_cache_test.go diff --git a/pkg/work/spoke/apply/resource_cache.go b/pkg/work/spoke/apply/resource_cache.go new file mode 100644 index 000000000..9d1131b9b --- /dev/null +++ b/pkg/work/spoke/apply/resource_cache.go @@ -0,0 +1,191 @@ +package apply + +import ( + "fmt" + "reflect" + "sync" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/klog/v2" +) + +type cachedVersionKey struct { + name string + namespace string + kind schema.GroupKind +} + +// record of resource metadata used to determine if its safe to return early from an ApplyFoo +// resourceHash is an ms5 hash of the required in an ApplyFoo that is computed in case the input changes +// resourceVersion is the received resourceVersion from the apiserver in response to an update that is comparable to the GET +type cachedResource struct { + resourceHash, resourceVersion string +} + +type resourceCache struct { + cache sync.Map // use syncmap for concurrent access +} + +// NewResourceCache creates a new resource cache instance. +// TODO: currently only work agent uses this syncmap cache, consider using this in other components +func NewResourceCache() *resourceCache { + return &resourceCache{ + cache: sync.Map{}, + } +} + +func getResourceMetadata(obj runtime.Object) (schema.GroupKind, string, string, string, error) { + if obj == nil { + return schema.GroupKind{}, "", "", "", fmt.Errorf("nil object has no metadata") + } + metadata, err := meta.Accessor(obj) + if err != nil { + return schema.GroupKind{}, "", "", "", err + } + if metadata == nil || reflect.ValueOf(metadata).IsNil() { + return schema.GroupKind{}, "", "", "", fmt.Errorf("object has no metadata") + } + resourceHash := hashOfResourceStruct(obj) + + // retrieve kind, sometimes this can be done via the accesor, sometimes not (depends on the type) + kind := schema.GroupKind{} + gvk := obj.GetObjectKind().GroupVersionKind() + if len(gvk.Kind) > 0 { + kind = gvk.GroupKind() + } else { + if currKind := getCoreGroupKind(obj); currKind != nil { + kind = *currKind + } + } + if len(kind.Kind) == 0 { + return schema.GroupKind{}, "", "", "", fmt.Errorf("unable to determine GroupKind of %T", obj) + } + + return kind, metadata.GetName(), metadata.GetNamespace(), resourceHash, nil +} + +func getResourceVersion(obj runtime.Object) (string, error) { + if obj == nil { + return "", fmt.Errorf("nil object has no resourceVersion") + } + metadata, err := meta.Accessor(obj) + if err != nil { + return "", err + } + if metadata == nil || reflect.ValueOf(metadata).IsNil() { + return "", fmt.Errorf("object has no metadata") + } + rv := metadata.GetResourceVersion() + if len(rv) == 0 { + return "", fmt.Errorf("missing resourceVersion") + } + + return rv, nil +} + +func (c *resourceCache) UpdateCachedResourceMetadata(required runtime.Object, actual runtime.Object) { + if c == nil { + return + } + if required == nil || actual == nil { + return + } + kind, name, namespace, resourceHash, err := getResourceMetadata(required) + if err != nil { + return + } + cacheKey := cachedVersionKey{ + name: name, + namespace: namespace, + kind: kind, + } + + resourceVersion, err := getResourceVersion(actual) + if err != nil { + klog.V(4).Infof("error reading resourceVersion %s:%s:%s %s", name, kind, namespace, err) + return + } + + c.cache.Store(cacheKey, cachedResource{resourceHash, resourceVersion}) + klog.V(7).Infof("updated resourceVersion of %s:%s:%s %s", name, kind, namespace, resourceVersion) +} + +// in the circumstance that an ApplyFoo's 'required' is the same one which was previously +// applied for a given (name, kind, namespace) and the existing resource (if any), +// hasn't been modified since the ApplyFoo last updated that resource, then return true (we don't +// need to reapply the resource). Otherwise return false. +func (c *resourceCache) SafeToSkipApply(required runtime.Object, existing runtime.Object) bool { + if c == nil { + return false + } + if required == nil || existing == nil { + return false + } + kind, name, namespace, resourceHash, err := getResourceMetadata(required) + if err != nil { + return false + } + cacheKey := cachedVersionKey{ + name: name, + namespace: namespace, + kind: kind, + } + + resourceVersion, err := getResourceVersion(existing) + if err != nil { + return false + } + + var versionMatch, hashMatch bool + + if value, ok := c.cache.Load(cacheKey); ok { + if cached, ok := value.(cachedResource); ok { + versionMatch = cached.resourceVersion == resourceVersion + hashMatch = cached.resourceHash == resourceHash + if versionMatch && hashMatch { + klog.V(4).Infof("found matching resourceVersion & manifest hash") + return true + } + } + } + + return false +} + +// TODO find way to create a registry of these based on struct mapping or some such that forces users to get this right +// +// for creating an ApplyGeneric +// Perhaps a struct containing the apply function and the getKind +func getCoreGroupKind(obj runtime.Object) *schema.GroupKind { + switch obj.(type) { + case *corev1.Namespace: + return &schema.GroupKind{ + Kind: "Namespace", + } + case *corev1.Service: + return &schema.GroupKind{ + Kind: "Service", + } + case *corev1.Pod: + return &schema.GroupKind{ + Kind: "Pod", + } + case *corev1.ServiceAccount: + return &schema.GroupKind{ + Kind: "ServiceAccount", + } + case *corev1.ConfigMap: + return &schema.GroupKind{ + Kind: "ConfigMap", + } + case *corev1.Secret: + return &schema.GroupKind{ + Kind: "Secret", + } + default: + return nil + } +} diff --git a/pkg/work/spoke/apply/resource_cache_test.go b/pkg/work/spoke/apply/resource_cache_test.go new file mode 100644 index 000000000..e5e5ceca2 --- /dev/null +++ b/pkg/work/spoke/apply/resource_cache_test.go @@ -0,0 +1,86 @@ +package apply + +import ( + "fmt" + "testing" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +func TestCache(t *testing.T) { + cache := NewResourceCache() + if cache == nil { + t.Fatal("expected non-nil resource cache") + } + + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "apps", + Version: "v1", + Kind: "Deployment", + }) + obj.SetResourceVersion("12345") + + obj.SetName("test") + obj.SetNamespace("default") + + // Test UpdateCachedResourceMetadata + cache.UpdateCachedResourceMetadata(obj, obj) + + // Test SafeToSkipApply + if !cache.SafeToSkipApply(obj, obj) { + t.Fatal("expected SafeToSkipApply to return true for identical objects") + } + + // Test SafeToSkipApply with different objects + obj2 := &unstructured.Unstructured{} + obj2.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "apps", + Version: "v1", + Kind: "Deployment", + }) + obj2.SetName("test2") + if cache.SafeToSkipApply(obj, obj2) { + t.Fatal("expected SafeToSkipApply to return false for different objects") + } + + obj3 := obj.DeepCopy() + obj3.SetResourceVersion("54321") + if cache.SafeToSkipApply(obj, obj3) { + t.Fatal("expected SafeToSkipApply to return false for objects with different resource versions") + } + cache.UpdateCachedResourceMetadata(obj, obj3) + if !cache.SafeToSkipApply(obj, obj3) { + t.Fatal("expected SafeToSkipApply to return true after updating cache with new resource version") + } +} + +func TestCurrentReadWriteCache(t *testing.T) { + // cache := resourceapply.NewResourceCache() + cache := NewResourceCache() + if cache == nil { + t.Fatal("expected non-nil resource cache") + } + + for i := range 1000 { + go func() { + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(schema.GroupVersionKind{ + Group: "apps", + Version: "v1", + Kind: "Deployment", + }) + + obj.SetNamespace("default") + obj.SetName("test") + obj.SetResourceVersion(fmt.Sprintf("12345%d", i)) + + cache.UpdateCachedResourceMetadata(obj, obj) + cache.SafeToSkipApply(obj, obj) + }() + } + + // if the code can run here without panic, it means the cache is thread-safe + t.Log("Cache operations completed without panic") +} diff --git a/pkg/work/spoke/apply/update_apply.go b/pkg/work/spoke/apply/update_apply.go index c183dda4b..27306daf3 100644 --- a/pkg/work/spoke/apply/update_apply.go +++ b/pkg/work/spoke/apply/update_apply.go @@ -36,8 +36,8 @@ func NewUpdateApply(dynamicClient dynamic.Interface, kubeclient kubernetes.Inter kubeclient: kubeclient, apiExtensionClient: apiExtensionClient, // TODO we did not gc resources in cache, which may cause more memory usage. It - // should be refactored using own cache implementation in the future. - staticResourceCache: resourceapply.NewResourceCache(), + // should be refactored in the future. + staticResourceCache: NewResourceCache(), } } diff --git a/pkg/work/spoke/apply/update_apply_test.go b/pkg/work/spoke/apply/update_apply_test.go index 869deec97..47124e947 100644 --- a/pkg/work/spoke/apply/update_apply_test.go +++ b/pkg/work/spoke/apply/update_apply_test.go @@ -4,7 +4,6 @@ import ( "context" "testing" - "github.com/openshift/library-go/pkg/operator/resource/resourceapply" corev1 "k8s.io/api/core/v1" apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" fakeapiextensions "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake" @@ -325,7 +324,7 @@ func TestApplyUnstructred(t *testing.T) { c.required.SetOwnerReferences([]metav1.OwnerReference{c.owner}) syncContext := testingcommon.NewFakeSyncContext(t, "test") - cache := resourceapply.NewResourceCache() + cache := NewResourceCache() cache.UpdateCachedResourceMetadata(c.required, c.existing) _, _, err := applier.applyUnstructured( context.TODO(), c.required, c.gvr, syncContext.Recorder(), cache)