Merge pull request #625 from Iceber/convert_deleted_object

add ConvertDeletedObject method for ResourceStorage

Author: Iceber

pkg/storage/internalstorage/resource_storage.go

@@ -20,6 +20,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 	genericstorage "k8s.io/apiserver/pkg/storage"
+	"k8s.io/client-go/tools/cache"
 
 	internal "github.com/clusterpedia-io/api/clusterpedia"
 	"github.com/clusterpedia-io/clusterpedia/pkg/storage"
@@ -125,6 +126,22 @@ func (s *ResourceStorage) Update(ctx context.Context, cluster string, obj runtim
 	return InterpretResourceDBError(cluster, metaobj.GetName(), result.Error)
 }
 
+func (c *ResourceStorage) ConvertDeletedObject(obj interface{}) (runtime.Object, error) {
+	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		return nil, err
+	}
+
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return nil, err
+	}
+
+	// Since it is not necessary to save the complete deleted object to the queue,
+	// we convert the object to `PartialObjectMetadata`
+	return &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}}, nil
+}
+
 func (s *ResourceStorage) deleteObject(cluster, namespace, name string) *gorm.DB {
 	return s.db.Model(&Resource{}).Where(map[string]interface{}{
 		"cluster":   cluster,

pkg/storage/memorystorage/memory_resource_storage.go

@@ -8,11 +8,13 @@ import (
 	"sync"
 
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/conversion"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/watch"
+	kubecache "k8s.io/client-go/tools/cache"
 
 	internal "github.com/clusterpedia-io/api/clusterpedia"
 	"github.com/clusterpedia-io/clusterpedia/pkg/storage"
@@ -82,6 +84,24 @@ func (s *ResourceStorage) Update(ctx context.Context, cluster string, obj runtim
 	return nil
 }
 
+func (s *ResourceStorage) ConvertDeletedObject(obj interface{}) (runobj runtime.Object, _ error) {
+	if d, ok := obj.(kubecache.DeletedFinalStateUnknown); ok {
+		if obj, ok := d.Obj.(runtime.Object); ok {
+			return obj, nil
+		}
+		namespace, name, err := kubecache.SplitMetaNamespaceKey(d.Key)
+		if err != nil {
+			return nil, err
+		}
+		return &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}}, nil
+	}
+
+	if obj, ok := obj.(runtime.Object); ok {
+		return obj, nil
+	}
+	return nil, fmt.Errorf("Invalid Type(%T): couldn't convert deleted object", obj)
+}
+
 func (s *ResourceStorage) Delete(ctx context.Context, cluster string, obj runtime.Object) error {
 	resourceVersion, err := s.CrvSynchro.UpdateClusterResourceVersion(obj, cluster)
 	if err != nil {

pkg/storage/storage.go

@@ -36,6 +36,8 @@ type ResourceStorage interface {
 
 	Create(ctx context.Context, cluster string, obj runtime.Object) error
 	Update(ctx context.Context, cluster string, obj runtime.Object) error
+
+	ConvertDeletedObject(obj interface{}) (runtime.Object, error)
 	Delete(ctx context.Context, cluster string, obj runtime.Object) error
 }
 

pkg/synchromanager/clustersynchro/resource_synchro.go

@@ -346,19 +346,14 @@ func (synchro *ResourceSynchro) OnDelete(obj interface{}) {
 	if !synchro.isRunnableForStorage.Load() {
 		return
 	}
-
-	if d, ok := obj.(cache.DeletedFinalStateUnknown); ok {
-		if obj, ok = d.Obj.(*unstructured.Unstructured); !ok {
-			namespace, name, err := cache.SplitMetaNamespaceKey(d.Key)
-			if err != nil {
-				return
-			}
-			obj = &metav1.PartialObjectMetadata{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}}
-		}
-	}
 	if o, ok := obj.(*unstructured.Unstructured); ok {
 		synchro.pruneObject(o)
 	}
+
+	obj, err := synchro.storage.ConvertDeletedObject(obj)
+	if err != nil {
+		return
+	}
 	_ = synchro.queue.Delete(obj)
 }