🐛 Use syncmap for the resource cache (#1023)

* Use syncmap for the resource cache Signed-off-by: zhujian <jiazhu@redhat.com> * update unit tests Signed-off-by: zhujian <jiazhu@redhat.com> * fix unit test Signed-off-by: zhujian <jiazhu@redhat.com> * use sync.map directly Signed-off-by: zhujian <jiazhu@redhat.com> --------- Signed-off-by: zhujian <jiazhu@redhat.com>
2026-08-23 22:26:49 +00:00 · 2025-06-05 01:58:40 +00:00
parent 8faa1b2327
commit fb5ba3acaf
4 changed files with 280 additions and 4 deletions
@@ -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
+	}
+}
@@ -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")
+}
@@ -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(),
 	}
 }

@@ -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)