Merge pull request #6612 from liaolecheng/fix/hookManagerSynced

Ensure hookManager waits for informer cache sync before reporting as synced in CustomizedInterpreter

Author: karmada-bot

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

@@ -17,13 +17,13 @@ limitations under the License.
 package configmanager
 
 import (
+	"errors"
 	"fmt"
 	"sort"
 	"sync/atomic"
 
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
@@ -48,6 +48,7 @@ type ConfigManager interface {
 // interpreterConfigManager collect the resource interpreter webhook configuration.
 type interpreterConfigManager struct {
 	configuration atomic.Value
+	informer      genericmanager.SingleClusterInformerManager
 	lister        cache.GenericLister
 	initialSynced atomic.Bool
 }
@@ -63,16 +64,12 @@ func (m *interpreterConfigManager) HasSynced() bool {
 		return true
 	}
 
-	if configuration, err := m.lister.List(labels.Everything()); err == nil && len(configuration) == 0 {
-		// the empty list we initially stored is valid to use.
-		// Setting initialSynced to true, so subsequent checks
-		// would be able to take the fast path on the atomic boolean in a
-		// cluster without any webhooks configured.
-		m.initialSynced.Store(true)
-		// the informer has synced, and we don't have any items
-		return true
+	err := m.updateConfiguration()
+	if err != nil {
+		klog.ErrorS(err, "error updating configuration")
+		return false
 	}
-	return false
+	return true
 }
 
 // NewExploreConfigManager return a new interpreterConfigManager with resourceinterpreterwebhookconfigurations handlers.
@@ -83,42 +80,53 @@ func NewExploreConfigManager(inform genericmanager.SingleClusterInformerManager)
 
 	manager.configuration.Store([]WebhookAccessor{})
 
+	manager.informer = inform
 	configHandlers := fedinformer.NewHandlerOnEvents(
-		func(_ interface{}) { manager.updateConfiguration() },
-		func(_, _ interface{}) { manager.updateConfiguration() },
-		func(_ interface{}) { manager.updateConfiguration() })
+		func(_ interface{}) { _ = manager.updateConfiguration() },
+		func(_, _ interface{}) { _ = manager.updateConfiguration() },
+		func(_ interface{}) { _ = manager.updateConfiguration() })
 	inform.ForResource(resourceExploringWebhookConfigurationsGVR, configHandlers)
 
 	return manager
 }
 
-func (m *interpreterConfigManager) updateConfiguration() {
+// updateConfiguration is used as the event handler for the ResourceInterpreterWebhookConfiguration resource.
+// Any changes (add, update, delete) to these resources will trigger this method, which loads all
+// ResourceInterpreterWebhookConfiguration resources and refreshes the internal cache accordingly.
+// Note: During startup, some events may be missed if the informer has not yet synced. If all events
+// are missed during startup, updateConfiguration will be called when HasSynced() is invoked for the
+// first time, ensuring the cache is updated on first use.
+func (m *interpreterConfigManager) updateConfiguration() error {
+	if m.informer == nil {
+		return errors.New("informer manager is not configured")
+	}
+	if !m.informer.IsInformerSynced(resourceExploringWebhookConfigurationsGVR) {
+		return errors.New("informer of ResourceInterpreterWebhookConfiguration not synced")
+	}
+
 	configurations, err := m.lister.List(labels.Everything())
 	if err != nil {
-		utilruntime.HandleError(fmt.Errorf("error updating configuration: %v", err))
-		return
+		return err
 	}
 
 	configs := make([]*configv1alpha1.ResourceInterpreterWebhookConfiguration, 0)
 	for _, c := range configurations {
 		unstructuredConfig, err := helper.ToUnstructured(c)
 		if err != nil {
-			klog.Errorf("Failed to transform ResourceInterpreterWebhookConfiguration: %v", err)
-			return
+			return err
 		}
 
 		config := &configv1alpha1.ResourceInterpreterWebhookConfiguration{}
 		err = helper.ConvertToTypedObject(unstructuredConfig, config)
 		if err != nil {
-			gvk := unstructuredConfig.GroupVersionKind().String()
-			klog.Errorf("Failed to convert object(%s), err: %v", gvk, err)
-			return
+			return err
 		}
 		configs = append(configs, config)
 	}
 
 	m.configuration.Store(mergeResourceExploreWebhookConfigurations(configs))
 	m.initialSynced.Store(true)
+	return nil
 }
 
 func mergeResourceExploreWebhookConfigurations(configurations []*configv1alpha1.ResourceInterpreterWebhookConfiguration) []WebhookAccessor {

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

@@ -17,14 +17,17 @@ limitations under the License.
 package configmanager
 
 import (
+	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/tools/cache"
 
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
@@ -62,6 +65,10 @@ func TestNewExploreConfigManager(t *testing.T) {
 
 			assert.NotNil(t, manager, "Manager should not be nil")
 			assert.NotNil(t, manager.HookAccessors(), "Accessors should be initialized")
+
+			internalManager, ok := manager.(*interpreterConfigManager)
+			assert.True(t, ok)
+			assert.Equal(t, informerManager, internalManager.informer)
 		})
 	}
 }
@@ -70,6 +77,7 @@ func TestHasSynced(t *testing.T) {
 	tests := []struct {
 		name           string
 		initialSynced  bool
+		informer       genericmanager.SingleClusterInformerManager
 		listErr        error
 		listResult     []runtime.Object
 		expectedSynced bool
@@ -80,24 +88,47 @@ func TestHasSynced(t *testing.T) {
 			expectedSynced: true,
 		},
 		{
-			name:           "not synced but empty list",
+			name:           "informer not configured",
 			initialSynced:  false,
+			informer:       nil,
+			expectedSynced: false,
+		},
+		{
+			name:          "informer not synced",
+			initialSynced: false,
+			informer: &mockSingleClusterInformerManager{
+				isSynced: false,
+			},
+			expectedSynced: false,
+		},
+		{
+			name:          "sync with empty list",
+			initialSynced: false,
+			informer: &mockSingleClusterInformerManager{
+				isSynced: true,
+			},
 			listResult:     []runtime.Object{},
 			expectedSynced: true,
 		},
 		{
-			name:          "not synced with items",
+			name:          "sync with items",
 			initialSynced: false,
+			informer: &mockSingleClusterInformerManager{
+				isSynced: true,
+			},
 			listResult: []runtime.Object{
 				&configv1alpha1.ResourceInterpreterWebhookConfiguration{
 					ObjectMeta: metav1.ObjectMeta{Name: "test"},
 				},
 			},
-			expectedSynced: false,
+			expectedSynced: true,
 		},
 		{
-			name:           "list error",
-			initialSynced:  false,
+			name:          "list error",
+			initialSynced: false,
+			informer: &mockSingleClusterInformerManager{
+				isSynced: true,
+			},
 			listErr:        fmt.Errorf("test error"),
 			expectedSynced: false,
 		},
@@ -106,6 +137,7 @@ func TestHasSynced(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			manager := &interpreterConfigManager{
+				informer: tt.informer,
 				lister: &mockLister{
 					err:   tt.listErr,
 					items: tt.listResult,
@@ -181,12 +213,30 @@ func TestUpdateConfiguration(t *testing.T) {
 		name          string
 		configs       []runtime.Object
 		listErr       error
+		informer      genericmanager.SingleClusterInformerManager
 		expectedCount int
 		wantSynced    bool
 	}{
 		{
-			name:          "empty configuration",
-			configs:       []runtime.Object{},
+			name:          "informer not configured",
+			informer:      nil,
+			expectedCount: 0,
+			wantSynced:    false,
+		},
+		{
+			name: "informer not synced",
+			informer: &mockSingleClusterInformerManager{
+				isSynced: false,
+			},
+			expectedCount: 0,
+			wantSynced:    false,
+		},
+		{
+			name:    "empty configuration",
+			configs: []runtime.Object{},
+			informer: &mockSingleClusterInformerManager{
+				isSynced: true,
+			},
 			expectedCount: 0,
 			wantSynced:    true,
 		},
@@ -204,13 +254,19 @@ func TestUpdateConfiguration(t *testing.T) {
 					},
 				},
 			},
+			informer: &mockSingleClusterInformerManager{
+				isSynced: true,
+			},
 			expectedCount: 1,
 			wantSynced:    true,
 		},
 		{
-			name:          "list error",
-			configs:       []runtime.Object{},
-			listErr:       fmt.Errorf("test error"),
+			name:    "list error",
+			configs: []runtime.Object{},
+			listErr: fmt.Errorf("test error"),
+			informer: &mockSingleClusterInformerManager{
+				isSynced: true,
+			},
 			expectedCount: 0,
 			wantSynced:    false,
 		},
@@ -223,21 +279,63 @@ func TestUpdateConfiguration(t *testing.T) {
 					items: tt.configs,
 					err:   tt.listErr,
 				},
+				informer: tt.informer,
 			}
 			manager.configuration.Store([]WebhookAccessor{})
 			manager.initialSynced.Store(false)
 
-			manager.updateConfiguration()
+			synced := manager.HasSynced()
+			assert.Equal(t, tt.wantSynced, synced)
 
 			accessors := manager.HookAccessors()
 			assert.Equal(t, tt.expectedCount, len(accessors))
-			assert.Equal(t, tt.wantSynced, manager.HasSynced())
 		})
 	}
 }
 
 // Mock Implementations
 
+type mockSingleClusterInformerManager struct {
+	isSynced bool
+}
+
+func (m *mockSingleClusterInformerManager) IsInformerSynced(_ schema.GroupVersionResource) bool {
+	return m.isSynced
+}
+
+func (m *mockSingleClusterInformerManager) Lister(_ schema.GroupVersionResource) cache.GenericLister {
+	return nil
+}
+
+func (m *mockSingleClusterInformerManager) ForResource(_ schema.GroupVersionResource, _ cache.ResourceEventHandler) {
+}
+
+func (m *mockSingleClusterInformerManager) Start() {
+}
+
+func (m *mockSingleClusterInformerManager) Stop() {
+}
+
+func (m *mockSingleClusterInformerManager) WaitForCacheSync() map[schema.GroupVersionResource]bool {
+	return nil
+}
+
+func (m *mockSingleClusterInformerManager) WaitForCacheSyncWithTimeout(_ time.Duration) map[schema.GroupVersionResource]bool {
+	return nil
+}
+
+func (m *mockSingleClusterInformerManager) Context() context.Context {
+	return context.Background()
+}
+
+func (m *mockSingleClusterInformerManager) GetClient() dynamic.Interface {
+	return nil
+}
+
+func (m *mockSingleClusterInformerManager) IsHandlerExist(_ schema.GroupVersionResource, _ cache.ResourceEventHandler) bool {
+	return false
+}
+
 type mockLister struct {
 	items []runtime.Object
 	err   error