diff --git a/pkg/yurthub/autonomymanager/autonomy_manager.go b/pkg/yurthub/autonomymanager/autonomy_manager.go index b6a3355dfee..993f0f04a9e 100644 --- a/pkg/yurthub/autonomymanager/autonomy_manager.go +++ b/pkg/yurthub/autonomymanager/autonomy_manager.go @@ -67,29 +67,26 @@ func NewAutonomyManager(restConfigMgr *hubrest.RestConfigManager, cacheMgr cache // utilize queue and wait func (am *AutonomyManager) Run(ctx context.Context) { // 1.Consistency Validate every 20 seconds - go func() { - checkTicker := time.NewTicker(time.Second * 20) - defer checkTicker.Stop() - for { - select { - case <-checkTicker.C: - config := am.restConfigMgr.GetRestConfig(true) - if config == nil { - return - } - client, err := kubernetes.NewForConfig(config) - if err != nil { - return - } - am.ConsistencyValidate(client) - case <-ctx.Done(): - return - } + go wait.JitterUntil(func() { + if am.restConfigMgr == nil { + return } - }() + config := am.restConfigMgr.GetRestConfig(true) + if config == nil { + return + } + client, err := kubernetes.NewForConfig(config) + if err != nil { + return + } + am.ConsistencyValidate(client) + }, 20*time.Second, 2, true, ctx.Done()) // 2. Sync - go wait.Until(func() { + go wait.JitterUntil(func() { + if am.restConfigMgr == nil { + return + } config := am.restConfigMgr.GetRestConfig(true) if config == nil { return @@ -99,7 +96,7 @@ func (am *AutonomyManager) Run(ctx context.Context) { return } am.Sync(dynamicClient) - }, 20*time.Second, ctx.Done()) + }, 20*time.Second, 3, true, ctx.Done()) // 3. Update errorKeys go func() { @@ -205,7 +202,6 @@ func EnsureAutonomyCondition(client kubernetes.Interface, node *v1.Node, oldCond // Sync fetch object from api server and write it to cache manager func (am *AutonomyManager) Sync(dynamicClient dynamic.Interface) { - errorKey, quit := am.queue.Get() if quit { return diff --git a/pkg/yurthub/autonomymanager/autonomy_manager_test.go b/pkg/yurthub/autonomymanager/autonomy_manager_test.go index ec911eb3fd1..9ebc245883a 100644 --- a/pkg/yurthub/autonomymanager/autonomy_manager_test.go +++ b/pkg/yurthub/autonomymanager/autonomy_manager_test.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "testing" + "time" "github.com/stretchr/testify/assert" v1 "k8s.io/api/core/v1" @@ -278,3 +279,36 @@ func TestAutonomyManager_Sync(t *testing.T) { }) } } + +func TestAutonomyManager_Run(t *testing.T) { + testcases := []struct { + name string + }{ + { + name: "test1", + }, + } + for _, tc := range testcases { + fakeClient := clientsetfake.NewSimpleClientset() + fakeSharedInformerFactory := informers.NewSharedInformerFactory(fakeClient, 0) + dStorage, err := disk.NewDiskStorage("/tmp/cache-manager") + if err != nil { + t.Errorf("failed to create disk storage, %v", err) + } + restRESTMapperMgr, err := hubmeta.NewRESTMapperManager("/tmp/cache-manager") + if err != nil { + t.Errorf("failed to create RESTMapper manager, %v", err) + } + sWrapper := cachemanager.NewStorageWrapper(dStorage) + serializerM := serializer.NewSerializerManager() + mgr := cachemanager.NewCacheManager(sWrapper, serializerM, restRESTMapperMgr, fakeSharedInformerFactory) + am, err := NewAutonomyManager(nil, mgr, tc.name) + if err != nil { + t.Errorf("failed to create autonomy manager, %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + am.Run(ctx) + time.Sleep(20 * time.Second) + cancel() + } +} diff --git a/pkg/yurtmanager/controller/autonomy/autonomy_controller_test.go b/pkg/yurtmanager/controller/autonomy/autonomy_controller_test.go index c1a5d8c5d82..6beb0b25383 100644 --- a/pkg/yurtmanager/controller/autonomy/autonomy_controller_test.go +++ b/pkg/yurtmanager/controller/autonomy/autonomy_controller_test.go @@ -31,6 +31,10 @@ import ( "github.com/openyurtio/openyurt/pkg/projectinfo" ) +func TestFormat(t *testing.T) { + assert.Equal(t, "autonomy-controller: autonomy-controller add controller Node", Format("autonomy-controller add controller %s", "Node")) +} + func TestReconcileAutonomy_Reconcile(t *testing.T) { testcases := []struct { name string diff --git a/test/e2e/yurt/yurtstaticset.go b/test/e2e/yurt/yurtstaticset.go index 3fe7803978c..3eacc14c830 100644 --- a/test/e2e/yurt/yurtstaticset.go +++ b/test/e2e/yurt/yurtstaticset.go @@ -40,7 +40,7 @@ const ( var _ = Describe("yurtStaticSet Test", Ordered, func() { ctx := context.Background() - timeout := 60 * time.Second + timeout := 90 * time.Second k8sClient := ycfg.YurtE2eCfg.RuntimeClient nodeToImageMap := make(map[string]string)