[WIP] sync resource with the last resource version

go.mod

@@ -22,6 +22,7 @@ require (
 	k8s.io/component-base v0.22.4
 	k8s.io/klog/v2 v2.9.0
 	k8s.io/kubernetes v1.22.4
+	k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a
 	sigs.k8s.io/controller-runtime v0.10.3
 	sigs.k8s.io/controller-tools v0.7.0
 )

pkg/synchromanager/clustersynchro/cluster_synchro.go

@@ -153,6 +153,7 @@ func (s *ClusterSynchro) initWithResourceVersions(resourceversions map[schema.Gr
 }
 
 type syncConfig struct {
+	kind            string
 	syncResource    schema.GroupVersionResource
 	storageResource schema.GroupVersionResource
 	convertor       runtime.ObjectConvertor
@@ -258,6 +259,7 @@ func (s *ClusterSynchro) SetResources(clusterResources []clustersv1alpha1.Cluste
 				storageResource := storageConfig.StorageGroupResource.WithVersion(storageConfig.StorageVersion.Version)
 				if _, ok := configs[storageResource]; !ok {
 					config := &syncConfig{
+						kind:            info.Kind,
 						syncResource:    syncResource,
 						storageResource: storageResource,
 						storageConfig:   storageConfig,
@@ -340,7 +342,8 @@ func (s *ClusterSynchro) SetResources(clusterResources []clustersv1alpha1.Cluste
 			s.resourceVersionCaches[gvr] = resourceVersionCache
 		}
 
-		synchro := newResourceSynchro(s.name,
+		syncKind := config.syncResource.GroupVersion().WithKind(config.kind)
+		synchro := newResourceSynchro(s.name, syncKind,
 			s.listerWatcherFactory.ForResource(metav1.NamespaceAll, config.syncResource),
 			resourceVersionCache,
 			config.convertor,

pkg/synchromanager/clustersynchro/informer/named_controller.go

@@ -14,24 +14,34 @@ type controller struct {
 	config cache.Config
 
 	reflectorMutex sync.RWMutex
-	reflector      *cache.Reflector
-	queue          cache.Queue
+	reflector      *Reflector
+
+	lastResourceVersion string
 }
 
-func NewNamedController(name string, config *cache.Config) cache.Controller {
+func NewNamedController(name string, config *cache.Config) *controller {
 	return &controller{
 		name:   name,
 		config: *config,
 	}
 }
 
+func (c *controller) SetLastResourceVersion(lastResourceVersion string) {
+	c.reflectorMutex.Lock()
+	defer c.reflectorMutex.Unlock()
+	if c.reflector != nil {
+		panic("controller is running, connot set last resource version")
+	}
+	c.lastResourceVersion = lastResourceVersion
+}
+
 func (c *controller) Run(stopCh <-chan struct{}) {
 	defer utilruntime.HandleCrash()
 	go func() {
 		<-stopCh
 		c.config.Queue.Close()
 	}()
-	r := cache.NewNamedReflector(
+	r := NewNamedReflector(
 		c.name,
 		c.config.ListerWatcher,
 		c.config.ObjectType,
@@ -42,6 +52,9 @@ func (c *controller) Run(stopCh <-chan struct{}) {
 	r.WatchListPageSize = c.config.WatchListPageSize
 
 	c.reflectorMutex.Lock()
+	if c.lastResourceVersion != "" {
+		r.lastSyncResourceVersion = c.lastResourceVersion
+	}
 	c.reflector = r
 	c.reflectorMutex.Unlock()
 
@@ -52,6 +65,27 @@ func (c *controller) Run(stopCh <-chan struct{}) {
 	wg.Wait()
 }
 
+/*
+func (c *controller) setLastResourceVersionForReflector(reflector *cache.Reflector) {
+	if c.resourceVersionGetter == nil {
+		return
+	}
+
+	rv := c.resourceVersionGetter.LastResourceVersion()
+	if rv == "" || rv == "0" {
+		return
+	}
+	rvValue := reflect.ValueOf(rv)
+
+	field := reflect.ValueOf(reflector).Elem().FieldByName("lastSyncResourceVersion")
+	value := reflect.NewAt(field.Type(), unsafe.Pointer(field.UnsafeAddr())).Elem()
+	if value.Kind() != rvValue.Kind() {
+		panic(fmt.Sprintf("reflector.lastSyncResourceVersion's value kind is %v", value.Kind()))
+	}
+	value.Set(rvValue)
+}
+*/
+
 func (c *controller) processLoop() {
 	for {
 		obj, err := c.config.Queue.Pop(cache.PopProcessFunc(c.config.Process))
@@ -71,10 +105,7 @@ func (c *controller) HasSynced() bool {
 	c.reflectorMutex.RLock()
 	defer c.reflectorMutex.RUnlock()
 
-	if c.queue == nil {
-		return false
-	}
-	return c.queue.HasSynced()
+	return c.config.Queue.HasSynced()
 }
 
 func (c *controller) LastSyncResourceVersion() string {