From 1047c6d97b45ff8a2d43474df86bc5364bac6ceb Mon Sep 17 00:00:00 2001 From: Iceber Gu Date: Tue, 18 Jun 2024 18:20:33 +0800 Subject: [PATCH] support for storage to implement custom resource synchro Signed-off-by: Iceber Gu --- .../app/controllermanager.go | 2 +- pkg/controller/dependentresource_manager.go | 2 +- pkg/discovery/controller/crd.go | 2 +- .../clustersynchro/cluster_synchro.go | 55 +++++++----- ...synchro.go => default_resource_synchro.go} | 88 +++++++++---------- .../informer/cache/.reflector.go | 0 .../informer/event_handler.go | 0 .../informer/informer.go | 0 .../informer/listwatch.go | 0 .../informer/named_controller.go | 0 .../informer/pager/.pager.go.copy | 0 .../informer/pager/pager.go | 0 .../informer/pager/result_stream.go | 0 .../informer/queue.go | 0 .../informer/reflector.go | 2 +- .../informer/resourceversion_informer.go | 0 .../informer/resourceversion_storage.go | 0 .../queue/event.go | 0 .../queue/event_test.go | 0 .../queue/pressurequeue.go | 0 .../queue/queue.go | 0 .../resourcesynchro/resource_synchro.go | 48 ++++++++++ 22 files changed, 131 insertions(+), 68 deletions(-) rename pkg/synchromanager/clustersynchro/{resource_synchro.go => default_resource_synchro.go} (87%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/cache/.reflector.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/event_handler.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/informer.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/listwatch.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/named_controller.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/pager/.pager.go.copy (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/pager/pager.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/pager/result_stream.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/queue.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/reflector.go (99%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/resourceversion_informer.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/informer/resourceversion_storage.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/queue/event.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/queue/event_test.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/queue/pressurequeue.go (100%) rename pkg/synchromanager/{clustersynchro => resourcesynchro}/queue/queue.go (100%) create mode 100644 pkg/synchromanager/resourcesynchro/resource_synchro.go diff --git a/cmd/controller-manager/app/controllermanager.go b/cmd/controller-manager/app/controllermanager.go index 7f3177655..ffd1eb5e3 100644 --- a/cmd/controller-manager/app/controllermanager.go +++ b/cmd/controller-manager/app/controllermanager.go @@ -29,7 +29,7 @@ import ( "github.com/clusterpedia-io/clusterpedia/pkg/controller/pediaclusterlifecycle" clientset "github.com/clusterpedia-io/clusterpedia/pkg/generated/clientset/versioned" "github.com/clusterpedia-io/clusterpedia/pkg/generated/informers/externalversions" - "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/clustersynchro/informer" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro/informer" clusterpediafeature "github.com/clusterpedia-io/clusterpedia/pkg/utils/feature" "github.com/clusterpedia-io/clusterpedia/pkg/version/verflag" ) diff --git a/pkg/controller/dependentresource_manager.go b/pkg/controller/dependentresource_manager.go index 32b280735..21c20eaba 100644 --- a/pkg/controller/dependentresource_manager.go +++ b/pkg/controller/dependentresource_manager.go @@ -16,7 +16,7 @@ import ( "k8s.io/klog/v2" policyv1alpha1 "github.com/clusterpedia-io/api/policy/v1alpha1" - "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/clustersynchro/informer" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro/informer" ) type DependentResourceManager struct { diff --git a/pkg/discovery/controller/crd.go b/pkg/discovery/controller/crd.go index da2b0d784..66cc97c78 100644 --- a/pkg/discovery/controller/crd.go +++ b/pkg/discovery/controller/crd.go @@ -18,7 +18,7 @@ import ( "k8s.io/klog/v2" "github.com/clusterpedia-io/clusterpedia/pkg/scheme" - "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/clustersynchro/informer" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro/informer" ) type CRDController struct { diff --git a/pkg/synchromanager/clustersynchro/cluster_synchro.go b/pkg/synchromanager/clustersynchro/cluster_synchro.go index 527c7a259..69ba8aa74 100644 --- a/pkg/synchromanager/clustersynchro/cluster_synchro.go +++ b/pkg/synchromanager/clustersynchro/cluster_synchro.go @@ -20,8 +20,9 @@ import ( kubestatemetrics "github.com/clusterpedia-io/clusterpedia/pkg/kube_state_metrics" "github.com/clusterpedia-io/clusterpedia/pkg/storage" "github.com/clusterpedia-io/clusterpedia/pkg/storageconfig" - "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/clustersynchro/informer" "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/features" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro/informer" clusterpediafeature "github.com/clusterpedia-io/clusterpedia/pkg/utils/feature" ) @@ -36,11 +37,12 @@ type ClusterSynchro struct { RESTConfig *rest.Config ClusterStatusUpdater ClusterStatusUpdater - storage storage.StorageFactory - syncConfig ClusterSyncConfig - healthChecker *healthChecker - dynamicDiscovery discovery.DynamicDiscoveryInterface - listerWatcherFactory informer.DynamicListerWatcherFactory + storage storage.StorageFactory + resourceSynchroFactory resourcesynchro.SynchroFactory + syncConfig ClusterSyncConfig + healthChecker *healthChecker + dynamicDiscovery discovery.DynamicDiscoveryInterface + listerWatcherFactory informer.DynamicListerWatcherFactory closeOnce sync.Once closer chan struct{} @@ -123,6 +125,12 @@ func New(name string, config *rest.Config, storage storage.StorageFactory, updat storageResourceVersions: make(map[schema.GroupVersionResource]map[string]interface{}), } + if factory, ok := storage.(resourcesynchro.SynchroFactory); ok { + synchro.resourceSynchroFactory = factory + } else { + synchro.resourceSynchroFactory = DefaultResourceSynchroFactory{} + } + var refresherOnce sync.Once synchro.dynamicDiscovery.Prepare(discovery.PrepareConfig{ ResourceMutationHandler: synchro.resetSyncResources, @@ -167,8 +175,9 @@ func New(name string, config *rest.Config, storage storage.StorageFactory, updat func (s *ClusterSynchro) GetMetricsWriterList() (writers metricsstore.MetricsWriterList) { s.storageResourceSynchros.Range(func(_, value interface{}) bool { - if synchro := value.(*ResourceSynchro); synchro.metricsWriter != nil { - writers = append(writers, synchro.metricsWriter) + synchro := value.(resourcesynchro.Synchro) + if writer := synchro.GetMetricsWriter(); writer != nil { + writers = append(writers, writer) } return true }) @@ -237,7 +246,7 @@ func (s *ClusterSynchro) Shutdown(updateStatus bool) { shutdownCount := 0 statuses := make(map[string][]string) s.storageResourceSynchros.Range(func(key, value interface{}) bool { - synchro := value.(*ResourceSynchro) + synchro := value.(resourcesynchro.Synchro) status := synchro.Status() if status.Status == clusterv1alpha2.ResourceSyncStatusStop && status.Reason == "" { shutdownCount++ @@ -245,7 +254,7 @@ func (s *ClusterSynchro) Shutdown(updateStatus bool) { } gvr := key.(schema.GroupVersionResource) - sr := fmt.Sprintf("%s,%s,%s", status.Status, status.Reason, synchro.runningStage) + sr := fmt.Sprintf("%s,%s,%s", status.Status, status.Reason, synchro.Stage()) msg := fmt.Sprintf("%s/%s/%s", gvr.Group, gvr.Version, gvr.Resource) statuses[sr] = append(statuses[sr], msg) return true @@ -359,18 +368,23 @@ func (s *ClusterSynchro) refreshSyncResources() { if s.syncConfig.MetricsStoreBuilder != nil { metricsStore = s.syncConfig.MetricsStoreBuilder.GetMetricStore(s.name, config.syncResource) } - synchro := newResourceSynchro(s.name, - ResourceSynchroConfig{ + synchro, err := s.resourceSynchroFactory.NewResourceSynchro(s.name, + resourcesynchro.Config{ GroupVersionResource: config.syncResource, Kind: config.kind, ListerWatcher: s.listerWatcherFactory.ForResource(metav1.NamespaceAll, config.syncResource), ObjectConvertor: config.convertor, - ResourceStorage: resourceStorage, MetricsStore: metricsStore, ResourceVersions: rvs, PageSizeForInformer: s.syncConfig.PageSizeForResourceSync, + ResourceStorage: resourceStorage, }, ) + if err != nil { + klog.ErrorS(err, "Failed to create resource synchro", "cluster", s.name, "storage resource", storageGVR) + updateSyncConditions(storageGVR, clusterv1alpha2.ResourceSyncStatusPending, "SynchroCreateFailed", fmt.Sprintf("new resource synchro failed: %s", err)) + continue + } s.waitGroup.StartWithChannel(s.closer, synchro.Run) s.storageResourceSynchros.Store(storageGVR, synchro) @@ -400,7 +414,7 @@ func (s *ClusterSynchro) refreshSyncResources() { for storageGVR := range removedStorageGVRs { if synchro, ok := s.storageResourceSynchros.Load(storageGVR); ok { select { - case <-synchro.(*ResourceSynchro).Close(): + case <-synchro.(resourcesynchro.Synchro).Close(): case <-s.closer: return } @@ -475,7 +489,7 @@ func (s *ClusterSynchro) runner() { go s.dynamicDiscovery.Start(s.handlerStopCh) s.storageResourceSynchros.Range(func(_, value interface{}) bool { - go value.(*ResourceSynchro).Start(s.handlerStopCh) + go value.(resourcesynchro.Synchro).Start(s.handlerStopCh) return true }) }() @@ -542,12 +556,13 @@ func (s *ClusterSynchro) genClusterStatus() *clusterv1alpha2.ClusterStatus { gr := schema.GroupResource{Group: status.Group, Resource: resource.Name} storageGVR := cond.StorageGVR(gr) if value, ok := s.storageResourceSynchros.Load(storageGVR); ok { - synchro := value.(*ResourceSynchro) - if gr != synchro.syncResource.GroupResource() { - cond.SyncResource = synchro.syncResource.GroupResource().String() + synchro := value.(resourcesynchro.Synchro) + syncedGVR := synchro.GroupVersionResource() + if gr != syncedGVR.GroupResource() { + cond.SyncResource = syncedGVR.GroupResource().String() } - if cond.Version != synchro.syncResource.Version { - cond.SyncVersion = synchro.syncResource.Version + if cond.Version != syncedGVR.Version { + cond.SyncVersion = syncedGVR.Version } status := synchro.Status() diff --git a/pkg/synchromanager/clustersynchro/resource_synchro.go b/pkg/synchromanager/clustersynchro/default_resource_synchro.go similarity index 87% rename from pkg/synchromanager/clustersynchro/resource_synchro.go rename to pkg/synchromanager/clustersynchro/default_resource_synchro.go index dbf0a37a6..dabfdc038 100644 --- a/pkg/synchromanager/clustersynchro/resource_synchro.go +++ b/pkg/synchromanager/clustersynchro/default_resource_synchro.go @@ -19,34 +19,16 @@ import ( metricsstore "k8s.io/kube-state-metrics/v2/pkg/metrics_store" clusterv1alpha2 "github.com/clusterpedia-io/api/cluster/v1alpha2" - kubestatemetrics "github.com/clusterpedia-io/clusterpedia/pkg/kube_state_metrics" "github.com/clusterpedia-io/clusterpedia/pkg/storage" - "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/clustersynchro/informer" - "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/clustersynchro/queue" "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/features" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro/informer" + "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro/queue" "github.com/clusterpedia-io/clusterpedia/pkg/utils" clusterpediafeature "github.com/clusterpedia-io/clusterpedia/pkg/utils/feature" ) -type ResourceSynchroConfig struct { - schema.GroupVersionResource - Kind string - - cache.ListerWatcher - runtime.ObjectConvertor - storage.ResourceStorage - - *kubestatemetrics.MetricsStore - - ResourceVersions map[string]interface{} - PageSizeForInformer int64 -} - -func (c ResourceSynchroConfig) GroupVersionKind() schema.GroupVersionKind { - return c.GroupVersionResource.GroupVersion().WithKind(c.Kind) -} - -type ResourceSynchro struct { +type resourceSynchro struct { cluster string example runtime.Object @@ -88,9 +70,11 @@ type ResourceSynchro struct { runningStage string } -func newResourceSynchro(cluster string, config ResourceSynchroConfig) *ResourceSynchro { +type DefaultResourceSynchroFactory struct{} + +func (factory DefaultResourceSynchroFactory) NewResourceSynchro(cluster string, config resourcesynchro.Config) (resourcesynchro.Synchro, error) { storageConfig := config.ResourceStorage.GetStorageConfig() - synchro := &ResourceSynchro{ + synchro := &resourceSynchro{ cluster: cluster, syncResource: config.GroupVersionResource, storageResource: storageConfig.StorageGroupResource.WithVersion(storageConfig.StorageVersion.Version), @@ -127,10 +111,26 @@ func newResourceSynchro(cluster string, config ResourceSynchroConfig) *ResourceS } synchro.setStatus(clusterv1alpha2.ResourceSyncStatusPending, "", "") - return synchro + return synchro, nil +} + +func (synchro *resourceSynchro) Stage() string { + return synchro.runningStage +} + +func (synchro *resourceSynchro) GroupVersionResource() schema.GroupVersionResource { + return synchro.syncResource +} + +func (synchro *resourceSynchro) StoragedGroupVersionResource() schema.GroupVersionResource { + return synchro.storageResource +} + +func (synchro *resourceSynchro) GetMetricsWriter() *metricsstore.MetricsWriter { + return synchro.metricsWriter } -func (synchro *ResourceSynchro) Run(shutdown <-chan struct{}) { +func (synchro *resourceSynchro) Run(shutdown <-chan struct{}) { defer close(synchro.closed) go func() { select { @@ -158,7 +158,7 @@ func (synchro *ResourceSynchro) Run(shutdown <-chan struct{}) { synchro.runningStage = "shutdown" } -func (synchro *ResourceSynchro) Close() <-chan struct{} { +func (synchro *resourceSynchro) Close() <-chan struct{} { synchro.closeOnce.Do(func() { close(synchro.closer) synchro.queue.Close() @@ -167,7 +167,7 @@ func (synchro *ResourceSynchro) Close() <-chan struct{} { return synchro.closed } -func (synchro *ResourceSynchro) Start(stopCh <-chan struct{}) { +func (synchro *resourceSynchro) Start(stopCh <-chan struct{}) { synchro.startlock.Lock() stopped := synchro.stopped // avoid race synchro.startlock.Unlock() @@ -288,7 +288,7 @@ func (synchro *ResourceSynchro) Start(stopCh <-chan struct{}) { const LastAppliedConfigurationAnnotation = "kubectl.kubernetes.io/last-applied-configuration" -func (synchro *ResourceSynchro) pruneObject(obj *unstructured.Unstructured) { +func (synchro *resourceSynchro) pruneObject(obj *unstructured.Unstructured) { if clusterpediafeature.FeatureGate.Enabled(features.PruneManagedFields) { obj.SetManagedFields(nil) } @@ -305,7 +305,7 @@ func (synchro *ResourceSynchro) pruneObject(obj *unstructured.Unstructured) { } } -func (synchro *ResourceSynchro) OnAdd(obj interface{}, isInInitialList bool) { +func (synchro *resourceSynchro) OnAdd(obj interface{}, isInInitialList bool) { if !synchro.isRunnableForStorage.Load() { return } @@ -327,7 +327,7 @@ func (synchro *ResourceSynchro) OnAdd(obj interface{}, isInInitialList bool) { _ = synchro.queue.Add(obj) } -func (synchro *ResourceSynchro) OnUpdate(_, obj interface{}) { +func (synchro *resourceSynchro) OnUpdate(_, obj interface{}) { if !synchro.isRunnableForStorage.Load() { return } @@ -342,7 +342,7 @@ func (synchro *ResourceSynchro) OnUpdate(_, obj interface{}) { _ = synchro.queue.Update(obj) } -func (synchro *ResourceSynchro) OnDelete(obj interface{}) { +func (synchro *resourceSynchro) OnDelete(obj interface{}) { if !synchro.isRunnableForStorage.Load() { return } @@ -357,9 +357,9 @@ func (synchro *ResourceSynchro) OnDelete(obj interface{}) { _ = synchro.queue.Delete(obj) } -func (synchro *ResourceSynchro) OnSync(obj interface{}) {} +func (synchro *resourceSynchro) OnSync(obj interface{}) {} -func (synchro *ResourceSynchro) processResources() { +func (synchro *resourceSynchro) processResources() { for { select { case <-synchro.closer: @@ -381,7 +381,7 @@ func (synchro *ResourceSynchro) processResources() { } } -func (synchro *ResourceSynchro) handleResourceEvent(event *queue.Event) { +func (synchro *resourceSynchro) handleResourceEvent(event *queue.Event) { defer func() { _ = synchro.queue.Done(event) }() obj, ok := event.Object.(runtime.Object) @@ -476,7 +476,7 @@ func (synchro *ResourceSynchro) handleResourceEvent(event *queue.Event) { } } -func (synchro *ResourceSynchro) setRunnableForStorage() { +func (synchro *resourceSynchro) setRunnableForStorage() { synchro.isRunnableForStorage.Store(true) synchro.forStorageLock.Lock() @@ -494,7 +494,7 @@ func (synchro *ResourceSynchro) setRunnableForStorage() { } } -func (synchro *ResourceSynchro) setStopForStorage() { +func (synchro *resourceSynchro) setStopForStorage() { synchro.isRunnableForStorage.Store(false) synchro.forStorageLock.Lock() @@ -512,7 +512,7 @@ func (synchro *ResourceSynchro) setStopForStorage() { } } -func (synchro *ResourceSynchro) convertToStorageVersion(obj runtime.Object) (runtime.Object, error) { +func (synchro *resourceSynchro) convertToStorageVersion(obj runtime.Object) (runtime.Object, error) { if synchro.syncResource == synchro.storageResource || synchro.convertor == nil { return obj, nil } @@ -535,7 +535,7 @@ func (synchro *ResourceSynchro) convertToStorageVersion(obj runtime.Object) (run return obj, nil } -func (synchro *ResourceSynchro) createOrUpdateResource(ctx context.Context, obj runtime.Object) error { +func (synchro *resourceSynchro) createOrUpdateResource(ctx context.Context, obj runtime.Object) error { err := synchro.storage.Create(ctx, synchro.cluster, obj) if genericstorage.IsExist(err) { return synchro.storage.Update(ctx, synchro.cluster, obj) @@ -543,7 +543,7 @@ func (synchro *ResourceSynchro) createOrUpdateResource(ctx context.Context, obj return err } -func (synchro *ResourceSynchro) updateOrCreateResource(ctx context.Context, obj runtime.Object) error { +func (synchro *resourceSynchro) updateOrCreateResource(ctx context.Context, obj runtime.Object) error { err := synchro.storage.Update(ctx, synchro.cluster, obj) if genericstorage.IsNotFound(err) { return synchro.storage.Create(ctx, synchro.cluster, obj) @@ -551,11 +551,11 @@ func (synchro *ResourceSynchro) updateOrCreateResource(ctx context.Context, obj return err } -func (synchro *ResourceSynchro) deleteResource(ctx context.Context, obj runtime.Object) error { +func (synchro *resourceSynchro) deleteResource(ctx context.Context, obj runtime.Object) error { return synchro.storage.Delete(ctx, synchro.cluster, obj) } -func (synchro *ResourceSynchro) setStatus(status string, reason, message string) { +func (synchro *resourceSynchro) setStatus(status string, reason, message string) { synchro.status.Store(clusterv1alpha2.ClusterResourceSyncCondition{ Status: status, Reason: reason, @@ -564,11 +564,11 @@ func (synchro *ResourceSynchro) setStatus(status string, reason, message string) }) } -func (synchro *ResourceSynchro) Status() clusterv1alpha2.ClusterResourceSyncCondition { +func (synchro *resourceSynchro) Status() clusterv1alpha2.ClusterResourceSyncCondition { return synchro.status.Load().(clusterv1alpha2.ClusterResourceSyncCondition) } -func (synchro *ResourceSynchro) ErrorHandler(r *informer.Reflector, err error) { +func (synchro *resourceSynchro) ErrorHandler(r *informer.Reflector, err error) { if err != nil { // TODO(iceber): Use `k8s.io/apimachinery/pkg/api/errors` to resolve the error type and update it to `status.Reason` synchro.setStatus(clusterv1alpha2.ResourceSyncStatusError, "ResourceWatchFailed", err.Error()) diff --git a/pkg/synchromanager/clustersynchro/informer/cache/.reflector.go b/pkg/synchromanager/resourcesynchro/informer/cache/.reflector.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/cache/.reflector.go rename to pkg/synchromanager/resourcesynchro/informer/cache/.reflector.go diff --git a/pkg/synchromanager/clustersynchro/informer/event_handler.go b/pkg/synchromanager/resourcesynchro/informer/event_handler.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/event_handler.go rename to pkg/synchromanager/resourcesynchro/informer/event_handler.go diff --git a/pkg/synchromanager/clustersynchro/informer/informer.go b/pkg/synchromanager/resourcesynchro/informer/informer.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/informer.go rename to pkg/synchromanager/resourcesynchro/informer/informer.go diff --git a/pkg/synchromanager/clustersynchro/informer/listwatch.go b/pkg/synchromanager/resourcesynchro/informer/listwatch.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/listwatch.go rename to pkg/synchromanager/resourcesynchro/informer/listwatch.go diff --git a/pkg/synchromanager/clustersynchro/informer/named_controller.go b/pkg/synchromanager/resourcesynchro/informer/named_controller.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/named_controller.go rename to pkg/synchromanager/resourcesynchro/informer/named_controller.go diff --git a/pkg/synchromanager/clustersynchro/informer/pager/.pager.go.copy b/pkg/synchromanager/resourcesynchro/informer/pager/.pager.go.copy similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/pager/.pager.go.copy rename to pkg/synchromanager/resourcesynchro/informer/pager/.pager.go.copy diff --git a/pkg/synchromanager/clustersynchro/informer/pager/pager.go b/pkg/synchromanager/resourcesynchro/informer/pager/pager.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/pager/pager.go rename to pkg/synchromanager/resourcesynchro/informer/pager/pager.go diff --git a/pkg/synchromanager/clustersynchro/informer/pager/result_stream.go b/pkg/synchromanager/resourcesynchro/informer/pager/result_stream.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/pager/result_stream.go rename to pkg/synchromanager/resourcesynchro/informer/pager/result_stream.go diff --git a/pkg/synchromanager/clustersynchro/informer/queue.go b/pkg/synchromanager/resourcesynchro/informer/queue.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/queue.go rename to pkg/synchromanager/resourcesynchro/informer/queue.go diff --git a/pkg/synchromanager/clustersynchro/informer/reflector.go b/pkg/synchromanager/resourcesynchro/informer/reflector.go similarity index 99% rename from pkg/synchromanager/clustersynchro/informer/reflector.go rename to pkg/synchromanager/resourcesynchro/informer/reflector.go index 701a0a8ee..41e72853e 100644 --- a/pkg/synchromanager/clustersynchro/informer/reflector.go +++ b/pkg/synchromanager/resourcesynchro/informer/reflector.go @@ -27,7 +27,7 @@ import ( "k8s.io/utils/clock" "k8s.io/utils/trace" - clspager "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/clustersynchro/informer/pager" + clspager "github.com/clusterpedia-io/clusterpedia/pkg/synchromanager/resourcesynchro/informer/pager" ) const defaultExpectedTypeName = "" diff --git a/pkg/synchromanager/clustersynchro/informer/resourceversion_informer.go b/pkg/synchromanager/resourcesynchro/informer/resourceversion_informer.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/resourceversion_informer.go rename to pkg/synchromanager/resourcesynchro/informer/resourceversion_informer.go diff --git a/pkg/synchromanager/clustersynchro/informer/resourceversion_storage.go b/pkg/synchromanager/resourcesynchro/informer/resourceversion_storage.go similarity index 100% rename from pkg/synchromanager/clustersynchro/informer/resourceversion_storage.go rename to pkg/synchromanager/resourcesynchro/informer/resourceversion_storage.go diff --git a/pkg/synchromanager/clustersynchro/queue/event.go b/pkg/synchromanager/resourcesynchro/queue/event.go similarity index 100% rename from pkg/synchromanager/clustersynchro/queue/event.go rename to pkg/synchromanager/resourcesynchro/queue/event.go diff --git a/pkg/synchromanager/clustersynchro/queue/event_test.go b/pkg/synchromanager/resourcesynchro/queue/event_test.go similarity index 100% rename from pkg/synchromanager/clustersynchro/queue/event_test.go rename to pkg/synchromanager/resourcesynchro/queue/event_test.go diff --git a/pkg/synchromanager/clustersynchro/queue/pressurequeue.go b/pkg/synchromanager/resourcesynchro/queue/pressurequeue.go similarity index 100% rename from pkg/synchromanager/clustersynchro/queue/pressurequeue.go rename to pkg/synchromanager/resourcesynchro/queue/pressurequeue.go diff --git a/pkg/synchromanager/clustersynchro/queue/queue.go b/pkg/synchromanager/resourcesynchro/queue/queue.go similarity index 100% rename from pkg/synchromanager/clustersynchro/queue/queue.go rename to pkg/synchromanager/resourcesynchro/queue/queue.go diff --git a/pkg/synchromanager/resourcesynchro/resource_synchro.go b/pkg/synchromanager/resourcesynchro/resource_synchro.go new file mode 100644 index 000000000..3be83ad87 --- /dev/null +++ b/pkg/synchromanager/resourcesynchro/resource_synchro.go @@ -0,0 +1,48 @@ +package resourcesynchro + +import ( + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/tools/cache" + metricsstore "k8s.io/kube-state-metrics/v2/pkg/metrics_store" + + clusterv1alpha2 "github.com/clusterpedia-io/api/cluster/v1alpha2" + kubestatemetrics "github.com/clusterpedia-io/clusterpedia/pkg/kube_state_metrics" + "github.com/clusterpedia-io/clusterpedia/pkg/storage" +) + +type SynchroFactory interface { + NewResourceSynchro(name string, config Config) (Synchro, error) +} + +type Synchro interface { + Run(shutdown <-chan struct{}) + Start(stopCh <-chan struct{}) + Close() <-chan struct{} + + GroupVersionResource() schema.GroupVersionResource + StoragedGroupVersionResource() schema.GroupVersionResource + + GetMetricsWriter() *metricsstore.MetricsWriter + + Stage() string + Status() clusterv1alpha2.ClusterResourceSyncCondition +} + +type Config struct { + schema.GroupVersionResource + Kind string + + ListerWatcher cache.ListerWatcher + ObjectConvertor runtime.ObjectConvertor + + ResourceVersions map[string]interface{} + PageSizeForInformer int64 + + MetricsStore *kubestatemetrics.MetricsStore + ResourceStorage storage.ResourceStorage +} + +func (c Config) GroupVersionKind() schema.GroupVersionKind { + return c.GroupVersionResource.GroupVersion().WithKind(c.Kind) +}