ensure resource interpreter cache sync before starting controllers

Signed-off-by: changzhen <[email protected]>

Author: XiShanYongYe-Chang

cmd/agent/app/agent.go

@@ -260,10 +260,9 @@ func setupControllers(ctx context.Context, mgr controllerruntime.Manager, opts *
 	sharedFactory.WaitForCacheSync(ctx.Done())
 
 	resourceInterpreter := resourceinterpreter.NewResourceInterpreter(controlPlaneInformerManager, serviceLister)
-	if err := mgr.Add(resourceInterpreter); err != nil {
-		return fmt.Errorf("failed to setup custom resource interpreter: %w", err)
+	if err := resourceInterpreter.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start resource interpreter: %w", err)
 	}
-
 	rateLimiterGetter := util.GetClusterRateLimiterGetter().SetDefaultLimits(opts.ClusterAPIQPS, opts.ClusterAPIBurst)
 	clusterClientOption := &util.ClientOption{RateLimiterGetter: rateLimiterGetter.GetRateLimiter}
 

cmd/controller-manager/app/controllermanager.go

@@ -839,8 +839,8 @@ func setupControllers(ctx context.Context, mgr controllerruntime.Manager, opts *
 	sharedFactory.WaitForCacheSync(ctx.Done())
 
 	resourceInterpreter := resourceinterpreter.NewResourceInterpreter(controlPlaneInformerManager, serviceLister)
-	if err := mgr.Add(resourceInterpreter); err != nil {
-		klog.Fatalf("Failed to setup custom resource interpreter: %v", err)
+	if err := resourceInterpreter.Start(ctx); err != nil {
+		klog.Fatalf("Failed to start resource interpreter: %v", err)
 	}
 	rateLimiterGetter := util.GetClusterRateLimiterGetter().SetDefaultLimits(opts.ClusterAPIQPS, opts.ClusterAPIBurst)
 	clusterClientOption := &util.ClientOption{RateLimiterGetter: rateLimiterGetter.GetRateLimiter}

pkg/resourceinterpreter/customized/declarative/configmanager/manager.go

@@ -27,21 +27,21 @@ import (
 	"k8s.io/klog/v2"
 
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
+	"github.com/karmada-io/karmada/pkg/util"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
 	"github.com/karmada-io/karmada/pkg/util/helper"
 )
 
-var resourceInterpreterCustomizationsGVR = schema.GroupVersionResource{
-	Group:    configv1alpha1.GroupVersion.Group,
-	Version:  configv1alpha1.GroupVersion.Version,
-	Resource: "resourceinterpretercustomizations",
-}
-
 // ConfigManager can list custom resource interpreter.
 type ConfigManager interface {
 	CustomAccessors() map[schema.GroupVersionKind]CustomAccessor
 	HasSynced() bool
+	// LoadConfig is used to load ResourceInterpreterCustomization into the cache,
+	// it requires the provided customizations to be a full list of objects.
+	// It is recommended to be called during startup. After called, HasSynced() will always
+	// return true, and CustomAccessors() will return a map of CustomAccessor containing
+	// all ResourceInterpreterCustomization configurations.
 	LoadConfig(customizations []*configv1alpha1.ResourceInterpreterCustomization)
 }
 
@@ -81,12 +81,12 @@ func NewInterpreterConfigManager(informer genericmanager.SingleClusterInformerMa
 	// In interpret command, rules are not loaded from server, so we don't start informer for it.
 	if informer != nil {
 		manager.informer = informer
-		manager.lister = informer.Lister(resourceInterpreterCustomizationsGVR)
+		manager.lister = informer.Lister(util.ResourceInterpreterCustomizationsGVR)
 		configHandlers := fedinformer.NewHandlerOnEvents(
 			func(_ interface{}) { _ = manager.updateConfiguration() },
 			func(_, _ interface{}) { _ = manager.updateConfiguration() },
 			func(_ interface{}) { _ = manager.updateConfiguration() })
-		informer.ForResource(resourceInterpreterCustomizationsGVR, configHandlers)
+		informer.ForResource(util.ResourceInterpreterCustomizationsGVR, configHandlers)
 	}
 
 	return manager
@@ -102,7 +102,7 @@ func (configManager *interpreterConfigManager) updateConfiguration() error {
 	if configManager.informer == nil {
 		return errors.New("informer manager is not configured")
 	}
-	if !configManager.informer.IsInformerSynced(resourceInterpreterCustomizationsGVR) {
+	if !configManager.informer.IsInformerSynced(util.ResourceInterpreterCustomizationsGVR) {
 		return errors.New("informer of ResourceInterpreterCustomization not synced")
 	}
 

pkg/resourceinterpreter/customized/webhook/configmanager/manager.go

@@ -23,26 +23,26 @@ import (
 	"sync/atomic"
 
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
+	"github.com/karmada-io/karmada/pkg/util"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
 	"github.com/karmada-io/karmada/pkg/util/helper"
 )
 
-var resourceExploringWebhookConfigurationsGVR = schema.GroupVersionResource{
-	Group:    configv1alpha1.GroupVersion.Group,
-	Version:  configv1alpha1.GroupVersion.Version,
-	Resource: "resourceinterpreterwebhookconfigurations",
-}
-
 // ConfigManager can list dynamic webhooks.
 type ConfigManager interface {
 	HookAccessors() []WebhookAccessor
 	HasSynced() bool
+	// LoadConfig is used to load ResourceInterpreterWebhookConfiguration into the cache,
+	// it requires the provided webhookConfigurations to be a full list of objects.
+	// It is recommended to be called during startup. After called, HasSynced() will always
+	// return true, and HookAccessors() will return a list of WebhookAccessor containing
+	// all ResourceInterpreterWebhookConfiguration configurations.
+	LoadConfig(webhookConfigurations []*configv1alpha1.ResourceInterpreterWebhookConfiguration)
 }
 
 // interpreterConfigManager collect the resource interpreter webhook configuration.
@@ -75,7 +75,7 @@ func (m *interpreterConfigManager) HasSynced() bool {
 // NewExploreConfigManager return a new interpreterConfigManager with resourceinterpreterwebhookconfigurations handlers.
 func NewExploreConfigManager(inform genericmanager.SingleClusterInformerManager) ConfigManager {
 	manager := &interpreterConfigManager{
-		lister: inform.Lister(resourceExploringWebhookConfigurationsGVR),
+		lister: inform.Lister(util.ResourceInterpreterWebhookConfigurationsGVR),
 	}
 
 	manager.configuration.Store([]WebhookAccessor{})
@@ -85,7 +85,7 @@ func NewExploreConfigManager(inform genericmanager.SingleClusterInformerManager)
 		func(_ interface{}) { _ = manager.updateConfiguration() },
 		func(_, _ interface{}) { _ = manager.updateConfiguration() },
 		func(_ interface{}) { _ = manager.updateConfiguration() })
-	inform.ForResource(resourceExploringWebhookConfigurationsGVR, configHandlers)
+	inform.ForResource(util.ResourceInterpreterWebhookConfigurationsGVR, configHandlers)
 
 	return manager
 }
@@ -100,7 +100,7 @@ func (m *interpreterConfigManager) updateConfiguration() error {
 	if m.informer == nil {
 		return errors.New("informer manager is not configured")
 	}
-	if !m.informer.IsInformerSynced(resourceExploringWebhookConfigurationsGVR) {
+	if !m.informer.IsInformerSynced(util.ResourceInterpreterWebhookConfigurationsGVR) {
 		return errors.New("informer of ResourceInterpreterWebhookConfiguration not synced")
 	}
 
@@ -109,26 +109,26 @@ func (m *interpreterConfigManager) updateConfiguration() error {
 		return err
 	}
 
-	configs := make([]*configv1alpha1.ResourceInterpreterWebhookConfiguration, 0)
-	for _, c := range configurations {
-		unstructuredConfig, err := helper.ToUnstructured(c)
-		if err != nil {
-			return err
-		}
-
+	configs := make([]*configv1alpha1.ResourceInterpreterWebhookConfiguration, len(configurations))
+	for index, c := range configurations {
 		config := &configv1alpha1.ResourceInterpreterWebhookConfiguration{}
-		err = helper.ConvertToTypedObject(unstructuredConfig, config)
+		err = helper.ConvertToTypedObject(c, config)
 		if err != nil {
 			return err
 		}
-		configs = append(configs, config)
+		configs[index] = config
 	}
 
-	m.configuration.Store(mergeResourceExploreWebhookConfigurations(configs))
-	m.initialSynced.Store(true)
+	m.LoadConfig(configs)
 	return nil
 }
 
+// LoadConfig loads the webhook configurations and updates the initialSynced flag to true.
+func (m *interpreterConfigManager) LoadConfig(webhookConfigurations []*configv1alpha1.ResourceInterpreterWebhookConfiguration) {
+	m.configuration.Store(mergeResourceExploreWebhookConfigurations(webhookConfigurations))
+	m.initialSynced.Store(true)
+}
+
 func mergeResourceExploreWebhookConfigurations(configurations []*configv1alpha1.ResourceInterpreterWebhookConfiguration) []WebhookAccessor {
 	sort.SliceStable(configurations, func(i, j int) bool {
 		return configurations[i].Name < configurations[j].Name

pkg/resourceinterpreter/customized/webhook/configmanager/manager_test.go

@@ -32,6 +32,7 @@ import (
 
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
+	"github.com/karmada-io/karmada/pkg/util/helper"
 )
 
 func TestNewExploreConfigManager(t *testing.T) {
@@ -136,11 +137,23 @@ func TestHasSynced(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			// Convert typed objects to unstructured objects for proper testing
+			unstructuredItems := make([]runtime.Object, len(tt.listResult))
+			for i, config := range tt.listResult {
+				if config != nil {
+					unstructuredObj, err := helper.ToUnstructured(config)
+					assert.NoError(t, err)
+					unstructuredItems[i] = unstructuredObj
+				} else {
+					unstructuredItems[i] = config
+				}
+			}
+
 			manager := &interpreterConfigManager{
 				informer: tt.informer,
 				lister: &mockLister{
 					err:   tt.listErr,
-					items: tt.listResult,
+					items: unstructuredItems,
 				},
 			}
 			manager.initialSynced.Store(tt.initialSynced)
@@ -274,9 +287,21 @@ func TestUpdateConfiguration(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			// Convert typed objects to unstructured objects for proper testing
+			unstructuredItems := make([]runtime.Object, len(tt.configs))
+			for i, config := range tt.configs {
+				if config != nil {
+					unstructuredObj, err := helper.ToUnstructured(config)
+					assert.NoError(t, err)
+					unstructuredItems[i] = unstructuredObj
+				} else {
+					unstructuredItems[i] = config
+				}
+			}
+
 			manager := &interpreterConfigManager{
 				lister: &mockLister{
-					items: tt.configs,
+					items: unstructuredItems,
 					err:   tt.listErr,
 				},
 				informer: tt.informer,

pkg/resourceinterpreter/customized/webhook/customized.go

@@ -355,3 +355,8 @@ func (e *CustomizedInterpreter) InterpretHealth(ctx context.Context, attributes
 
 	return response.Healthy, matched, nil
 }
+
+// LoadConfig loads the webhook configurations.
+func (e *CustomizedInterpreter) LoadConfig(webhookConfigurations []*configv1alpha1.ResourceInterpreterWebhookConfiguration) {
+	e.hookManager.LoadConfig(webhookConfigurations)
+}

pkg/resourceinterpreter/customized/webhook/customized_test.go

@@ -1085,6 +1085,12 @@ func (m *mockConfigManager) HookAccessors() []configmanager.WebhookAccessor {
 	return m.hooks
 }
 
+func (m *mockConfigManager) LoadConfig(_ []*configv1alpha1.ResourceInterpreterWebhookConfiguration) {
+	// Mock implementation: in a real test, we might want to process the configurations
+	// and update the hooks accordingly. For now, this is a no-op implementation.
+	// This allows the mock to satisfy the ConfigManager interface.
+}
+
 // mockWebhookAccessor implements configmanager.WebhookAccessor interface for testing
 type mockWebhookAccessor struct {
 	uid             string

pkg/resourceinterpreter/interpreter.go

@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	corev1 "k8s.io/client-go/listers/core/v1"
@@ -32,12 +33,14 @@ import (
 	"github.com/karmada-io/karmada/pkg/resourceinterpreter/customized/webhook/request"
 	"github.com/karmada-io/karmada/pkg/resourceinterpreter/default/native"
 	"github.com/karmada-io/karmada/pkg/resourceinterpreter/default/thirdparty"
+	"github.com/karmada-io/karmada/pkg/util"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
+	"github.com/karmada-io/karmada/pkg/util/helper"
 )
 
 // ResourceInterpreter manages both default and customized webhooks to interpret custom resource structure.
 type ResourceInterpreter interface {
-	// Start starts running the component and will never stop running until the context is closed or an error occurs.
+	// Start initializes the resource interpreter and performs cache synchronization.
 	Start(ctx context.Context) (err error)
 
 	// HookEnabled tells if any hook exist for specific resource type and operation.
@@ -85,13 +88,16 @@ type customResourceInterpreterImpl struct {
 	defaultInterpreter      *native.DefaultInterpreter
 }
 
-// Start starts running the component and will never stop running until the context is closed or an error occurs.
-func (i *customResourceInterpreterImpl) Start(ctx context.Context) (err error) {
-	klog.Infof("Starting custom resource interpreter.")
+// Start initializes all interpreters and load all ResourceInterpreterCustomization and
+// ResourceInterpreterWebhookConfiguration configurations into the cache.
+// It is recommended to be called before all controllers. After called, the resource interpreter
+// will be ready to interpret custom resources.
+func (i *customResourceInterpreterImpl) Start(_ context.Context) (err error) {
+	klog.Infoln("Starting resource interpreter.")
 
 	i.customizedInterpreter, err = webhook.NewCustomizedInterpreter(i.informer, i.serviceLister)
 	if err != nil {
-		return
+		return err
 	}
 	i.configurableInterpreter = declarative.NewConfigurableInterpreter(i.informer)
 
@@ -100,8 +106,12 @@ func (i *customResourceInterpreterImpl) Start(ctx context.Context) (err error) {
 
 	i.informer.Start()
 	i.informer.WaitForCacheSync()
-	<-ctx.Done()
-	klog.Infof("Stopped as context canceled.")
+
+	if err = i.loadConfig(); err != nil {
+		return err
+	}
+
+	klog.Infoln("Resource interpreter started.")
 	return nil
 }
 
@@ -339,3 +349,46 @@ func (i *customResourceInterpreterImpl) InterpretHealth(object *unstructured.Uns
 	healthy, err = i.defaultInterpreter.InterpretHealth(object)
 	return
 }
+
+// loadConfig loads the full set of ResourceInterpreterCustomization and
+// ResourceInterpreterWebhookConfiguration configurations into the cache. It avoids resource interpreter
+// parsing errors when the resource interpreter starts and the cache is not synchronized.
+func (i *customResourceInterpreterImpl) loadConfig() error {
+	customizations, err := i.informer.Lister(util.ResourceInterpreterCustomizationsGVR).List(labels.Everything())
+	if err != nil {
+		klog.Errorf("Failed to list resourceinterpretercustomizations: %v", err)
+		return err
+	}
+	klog.V(5).Infof("Found %d resourceinterpretercustomizations", len(customizations))
+
+	declareConfigs := make([]*configv1alpha1.ResourceInterpreterCustomization, len(customizations))
+	for index, c := range customizations {
+		config := &configv1alpha1.ResourceInterpreterCustomization{}
+		if err = helper.ConvertToTypedObject(c, config); err != nil {
+			klog.Errorf("Failed to convert resourceinterpretercustomization: %v", err)
+			return err
+		}
+		declareConfigs[index] = config
+	}
+	i.configurableInterpreter.LoadConfig(declareConfigs)
+
+	webhooks, err := i.informer.Lister(util.ResourceInterpreterWebhookConfigurationsGVR).List(labels.Everything())
+	if err != nil {
+		klog.Errorf("Failed to list resourceinterpreterwebhookconfigurations: %v", err)
+		return err
+	}
+	klog.V(5).Infof("Found %d resourceinterpreterwebhookconfigurations", len(webhooks))
+
+	webhookConfigs := make([]*configv1alpha1.ResourceInterpreterWebhookConfiguration, len(webhooks))
+	for index, c := range webhooks {
+		config := &configv1alpha1.ResourceInterpreterWebhookConfiguration{}
+		if err = helper.ConvertToTypedObject(c, config); err != nil {
+			klog.Errorf("Failed to convert resourceinterpreterwebhookconfiguration: %v", err)
+			return err
+		}
+		webhookConfigs[index] = config
+	}
+	i.customizedInterpreter.LoadConfig(webhookConfigs)
+
+	return nil
+}

pkg/util/constants.go

@@ -21,6 +21,7 @@ import (
 
 	discoveryv1 "k8s.io/api/discovery/v1"
 
+	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
 	workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
 )
 
@@ -249,6 +250,15 @@ var (
 	EndpointSliceGVK = discoveryv1.SchemeGroupVersion.WithKind("EndpointSlice")
 )
 
+// Define resource group version resource.
+var (
+	// ResourceInterpreterCustomizationsGVR is the GroupVersionResource of ResourceInterpreterCustomizations.
+	ResourceInterpreterCustomizationsGVR = configv1alpha1.SchemeGroupVersion.WithResource("resourceinterpretercustomizations")
+
+	// ResourceInterpreterWebhookConfigurationsGVR is the GroupVersionResource of ResourceInterpreterWebhookConfigurations.
+	ResourceInterpreterWebhookConfigurationsGVR = configv1alpha1.SchemeGroupVersion.WithResource("resourceinterpreterwebhookconfigurations")
+)
+
 const (
 	// DefaultFilePerm default file perm
 	DefaultFilePerm = 0640