From ee923cd5f632945e200e1a2980447cfa190bb953 Mon Sep 17 00:00:00 2001 From: Wei Liu Date: Tue, 14 Mar 2023 17:01:12 +0800 Subject: [PATCH] appliedmanifestwork eviction (#190) Signed-off-by: Wei Liu --- deploy/spoke/appliedmanifestworks.crd.yaml | 4 + pkg/helper/helper_test.go | 3 +- .../manifestwork_finalize_controller.go | 8 +- .../manifestwork_finalize_controller_test.go | 10 +- ...nmanaged_appliedmanifestwork_controller.go | 188 +++++++------- ...ged_appliedmanifestwork_controller_test.go | 236 +++++++++--------- pkg/spoke/spokeagent.go | 24 +- .../integration/unmanaged_appliedwork_test.go | 178 +++++++++---- test/integration/work_test.go | 24 +- 9 files changed, 392 insertions(+), 283 deletions(-) diff --git a/deploy/spoke/appliedmanifestworks.crd.yaml b/deploy/spoke/appliedmanifestworks.crd.yaml index 07fe26ee8..b44731fca 100644 --- a/deploy/spoke/appliedmanifestworks.crd.yaml +++ b/deploy/spoke/appliedmanifestworks.crd.yaml @@ -72,6 +72,10 @@ spec: version: description: Version is the version of the Kubernetes resource. type: string + evictionStartTime: + description: 'EvictionStartTime represents the current appliedmanifestwork will be evicted after a grace period. An appliedmanifestwork will be evicted from the managed cluster in the following two scenarios: - the manifestwork of the current appliedmanifestwork is missing on the hub, or - the appliedmanifestwork hub hash does not match the current hub hash of the work agent.' + type: string + format: date-time served: true storage: true subresources: diff --git a/pkg/helper/helper_test.go b/pkg/helper/helper_test.go index e2fda5f91..8f4ad2e88 100644 --- a/pkg/helper/helper_test.go +++ b/pkg/helper/helper_test.go @@ -4,11 +4,12 @@ import ( "context" "encoding/json" "fmt" - "github.com/google/go-cmp/cmp" "reflect" "testing" "time" + "github.com/google/go-cmp/cmp" + "github.com/openshift/library-go/pkg/operator/events/eventstesting" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/equality" diff --git a/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller.go b/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller.go index 32c55bd81..d51e89918 100644 --- a/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller.go +++ b/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller.go @@ -68,13 +68,11 @@ func (m *ManifestWorkFinalizeController) sync(ctx context.Context, controllerCon manifestWork, err := m.manifestWorkLister.Get(manifestWorkName) - // Delete appliedmanifestwork if relating manfiestwork is not found or being deleted + // Delete appliedmanifestwork if relating manfiestwork is being deleted switch { case errors.IsNotFound(err): - err := m.deleteAppliedManifestWork(ctx, appliedManifestWorkName) - if err != nil { - return err - } + // the appliedmanifestwork will be evicted if the relating manfiestwork is not found + return nil case err != nil: return err case !manifestWork.DeletionTimestamp.IsZero(): diff --git a/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller_test.go b/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller_test.go index 817ea5f9a..d50a931b1 100644 --- a/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller_test.go +++ b/pkg/spoke/controllers/finalizercontroller/manifestwork_finalize_controller_test.go @@ -185,19 +185,13 @@ func TestSyncManifestWorkController(t *testing.T) { Name: fmt.Sprintf("%s-work", hubHash), }, }, - validateAppliedManifestWorkActions: func(t *testing.T, actions []clienttesting.Action) { - if len(actions) != 1 { - t.Errorf("Expect 2 actions on appliedmanifestwork, but have %d", len(actions)) - } - - spoketesting.AssertAction(t, actions[0], "delete") - }, + validateAppliedManifestWorkActions: noAction, validateManifestWorkActions: func(t *testing.T, actions []clienttesting.Action) { if len(actions) != 0 { t.Errorf("Suppose nothing done for manifestwork") } }, - expectedQueueLen: 1, + expectedQueueLen: 0, }, } diff --git a/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller.go b/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller.go index a28330cb8..ef1290a75 100644 --- a/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller.go +++ b/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller.go @@ -2,17 +2,20 @@ package finalizercontroller import ( "context" + "encoding/json" "fmt" + "strings" + "time" + jsonpatch "github.com/evanphx/json-patch" "github.com/openshift/library-go/pkg/controller/factory" "github.com/openshift/library-go/pkg/operator/events" "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" - utilerrors "k8s.io/apimachinery/pkg/util/errors" - utilruntime "k8s.io/apimachinery/pkg/util/runtime" - "k8s.io/client-go/tools/cache" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/workqueue" "k8s.io/klog/v2" workv1client "open-cluster-management.io/api/client/work/clientset/versioned/typed/work/v1" workinformer "open-cluster-management.io/api/client/work/informers/externalversions/work/v1" @@ -21,41 +24,42 @@ import ( "open-cluster-management.io/work/pkg/helper" ) -const byWorkNameAndAgentID = "UnManagedAppliedManifestWork-byWorkNameAndAgentID" - -// UnManagedAppliedWorkController deletes unmanaged applied works. -type UnManagedAppliedWorkController struct { - manifestWorkLister worklister.ManifestWorkNamespaceLister - appliedManifestWorkClient workv1client.AppliedManifestWorkInterface - appliedManifestWorkLister worklister.AppliedManifestWorkLister - appliedManifestWorkIndexer cache.Indexer - hubHash string - agentID string +type unmanagedAppliedWorkController struct { + manifestWorkLister worklister.ManifestWorkNamespaceLister + appliedManifestWorkClient workv1client.AppliedManifestWorkInterface + appliedManifestWorkLister worklister.AppliedManifestWorkLister + hubHash string + agentID string + evictionGracePeriod time.Duration + rateLimiter workqueue.RateLimiter } +// NewUnManagedAppliedWorkController returns a controller to evict the unmanaged appliedmanifestworks. +// +// An appliedmanifestwork will be considered unmanaged in the following scenarios: +// - the manifestwork of the current appliedmanifestwork is missing on the hub, or +// - the appliedmanifestwork hub hash does not match the current hub hash of the work agent. +// +// One unmanaged appliedmanifestwork will be evicted from the managed cluster after a grace period (by +// default, 10 minutes), after one appliedmanifestwork is evicted from the managed cluster, its owned +// resources will also be evicted from the managed cluster with Kubernetes garbage collection. func NewUnManagedAppliedWorkController( recorder events.Recorder, manifestWorkInformer workinformer.ManifestWorkInformer, manifestWorkLister worklister.ManifestWorkNamespaceLister, appliedManifestWorkClient workv1client.AppliedManifestWorkInterface, appliedManifestWorkInformer workinformer.AppliedManifestWorkInformer, + evictionGracePeriod time.Duration, hubHash, agentID string, ) factory.Controller { - - controller := &UnManagedAppliedWorkController{ - manifestWorkLister: manifestWorkLister, - appliedManifestWorkClient: appliedManifestWorkClient, - appliedManifestWorkLister: appliedManifestWorkInformer.Lister(), - appliedManifestWorkIndexer: appliedManifestWorkInformer.Informer().GetIndexer(), - hubHash: hubHash, - agentID: agentID, - } - - err := appliedManifestWorkInformer.Informer().AddIndexers(cache.Indexers{ - byWorkNameAndAgentID: indexByWorkNameAndAgentID, - }) - if err != nil { - utilruntime.HandleError(err) + controller := &unmanagedAppliedWorkController{ + manifestWorkLister: manifestWorkLister, + appliedManifestWorkClient: appliedManifestWorkClient, + appliedManifestWorkLister: appliedManifestWorkInformer.Lister(), + hubHash: hubHash, + agentID: agentID, + evictionGracePeriod: evictionGracePeriod, + rateLimiter: workqueue.NewItemExponentialFailureRateLimiter(1*time.Minute, evictionGracePeriod), } return factory.New(). @@ -63,92 +67,102 @@ func NewUnManagedAppliedWorkController( accessor, _ := meta.Accessor(obj) return fmt.Sprintf("%s-%s", hubHash, accessor.GetName()) }, manifestWorkInformer.Informer()). - WithFilteredEventsInformersQueueKeyFunc(func(obj runtime.Object) string { - accessor, _ := meta.Accessor(obj) - return accessor.GetName() - }, helper.AppliedManifestworkHubHashFilter(hubHash), appliedManifestWorkInformer.Informer()). + WithFilteredEventsInformersQueueKeyFunc( + func(obj runtime.Object) string { + accessor, _ := meta.Accessor(obj) + return accessor.GetName() + }, helper.AppliedManifestworkAgentIDFilter(agentID), appliedManifestWorkInformer.Informer()). WithSync(controller.sync).ToController("UnManagedAppliedManifestWork", recorder) } -func (m *UnManagedAppliedWorkController) sync(ctx context.Context, controllerContext factory.SyncContext) error { +func (m *unmanagedAppliedWorkController) sync(ctx context.Context, controllerContext factory.SyncContext) error { appliedManifestWorkName := controllerContext.QueueKey() - klog.V(4).Infof("Reconciling ManifestWork %q", appliedManifestWorkName) + klog.V(4).Infof("Reconciling AppliedManifestWork %q", appliedManifestWorkName) appliedManifestWork, err := m.appliedManifestWorkLister.Get(appliedManifestWorkName) if errors.IsNotFound(err) { - // work not found, could have been deleted, do nothing. + // appliedmanifestwork not found, could have been deleted, do nothing. return nil } if err != nil { return err } - // We delete the old AppliedManifestWork only when the related ManifestWork is applied with the new - // AppliedManifestWork as the new owner. This can avoid deleting the old AppliedManifestWork prematurely - // before the new AppliedManifestWork takes the ownership of the applied resources. - manifestWork, err := m.manifestWorkLister.Get(appliedManifestWork.Spec.ManifestWorkName) + _, err = m.manifestWorkLister.Get(appliedManifestWork.Spec.ManifestWorkName) if errors.IsNotFound(err) { - // work not found, could have been deleted, do nothing. - return nil + // evict the current appliedmanifestwork when its relating manifestwork is missing on the hub + return m.evictAppliedManifestWork(ctx, controllerContext, appliedManifestWork) } if err != nil { return err } - if !meta.IsStatusConditionTrue(manifestWork.Status.Conditions, workapiv1.WorkApplied) { - // the work is not applied, do nothing. + + // manifestwork exists but hub changed + if !strings.HasPrefix(appliedManifestWork.Name, m.hubHash) { + return m.evictAppliedManifestWork(ctx, controllerContext, appliedManifestWork) + } + + // stop to evict the current appliedmanifestwork when its relating manifestwork is recreated on the hub + return m.stopToEvictAppliedManifestWork(ctx, appliedManifestWork) +} + +func (m *unmanagedAppliedWorkController) evictAppliedManifestWork(ctx context.Context, + controllerContext factory.SyncContext, appliedManifestWork *workapiv1.AppliedManifestWork) error { + now := time.Now() + + evictionStartTime := appliedManifestWork.Status.EvictionStartTime + if evictionStartTime == nil { + return m.patchEvictionStartTime(ctx, appliedManifestWork, &metav1.Time{Time: now}) + } + + if now.Before(evictionStartTime.Add(m.evictionGracePeriod)) { + controllerContext.Queue().AddAfter(appliedManifestWork.Name, m.rateLimiter.When(appliedManifestWork.Name)) return nil } - unManagedAppliedWorks, err := m.getUnManagedAppliedManifestWorksByIndex(appliedManifestWork.Spec.ManifestWorkName, appliedManifestWork.Spec.AgentID) + klog.V(2).Infof("Delete appliedWork %s by agent %s after eviction grace periodby", appliedManifestWork.Name, m.agentID) + return m.appliedManifestWorkClient.Delete(ctx, appliedManifestWork.Name, metav1.DeleteOptions{}) +} + +func (m *unmanagedAppliedWorkController) stopToEvictAppliedManifestWork( + ctx context.Context, appliedManifestWork *workapiv1.AppliedManifestWork) error { + if appliedManifestWork.Status.EvictionStartTime == nil { + return nil + } + + m.rateLimiter.Forget(appliedManifestWork.Name) + return m.patchEvictionStartTime(ctx, appliedManifestWork, nil) +} + +func (m *unmanagedAppliedWorkController) patchEvictionStartTime(ctx context.Context, + appliedManifestWork *workapiv1.AppliedManifestWork, evictionStartTime *metav1.Time) error { + oldData, err := json.Marshal(workapiv1.AppliedManifestWork{ + Status: workapiv1.AppliedManifestWorkStatus{ + EvictionStartTime: appliedManifestWork.Status.EvictionStartTime, + }, + }) if err != nil { - return err + return fmt.Errorf("failed to Marshal old data for appliedmanifestwork status %s: %w", appliedManifestWork.Name, err) } - var errs []error - for _, appliedWork := range unManagedAppliedWorks { - klog.V(2).Infof("Delete appliedWork %s since it is not managed by agent %s anymore", appliedWork.Name, m.agentID) - err := m.appliedManifestWorkClient.Delete(ctx, appliedWork.Name, metav1.DeleteOptions{}) - if err != nil { - errs = append(errs, err) - } - } - - return utilerrors.NewAggregate(errs) -} - -// getUnManagedAppliedManifestWorksByIndex finds appliedmanifestwork with the same workname and agent ID but different hubhash. -// These appliedManifestWorks is considered to be not managed by this work agent anymore and should be deleted. -// The reason of marking them as unmanaged is because the only reason under this conditions is work agent is switched to connect -// to a recovered hub or a fresh new hub. Those appliedmanifestwork needs to be deleted to avoid conflict with the newly connected -// hub. -func (m *UnManagedAppliedWorkController) getUnManagedAppliedManifestWorksByIndex(workName, agentID string) ([]*workapiv1.AppliedManifestWork, error) { - index := agentIDWorkNameIndex(workName, agentID) - items, err := m.appliedManifestWorkIndexer.ByIndex(byWorkNameAndAgentID, index) + newData, err := json.Marshal(workapiv1.AppliedManifestWork{ + ObjectMeta: metav1.ObjectMeta{ + UID: appliedManifestWork.UID, + ResourceVersion: appliedManifestWork.ResourceVersion, + }, + Status: workapiv1.AppliedManifestWorkStatus{ + EvictionStartTime: evictionStartTime, + }, + }) if err != nil { - return nil, err + return fmt.Errorf("failed to Marshal new data for appliedmanifestwork status %s: %w", appliedManifestWork.Name, err) } - ret := make([]*workapiv1.AppliedManifestWork, 0, len(items)) - for _, item := range items { - appliedWork := item.(*workapiv1.AppliedManifestWork) - if appliedWork.Spec.HubHash == m.hubHash { - continue - } - ret = append(ret, item.(*workapiv1.AppliedManifestWork)) + patchBytes, err := jsonpatch.CreateMergePatch(oldData, newData) + if err != nil { + return fmt.Errorf("failed to create patch for cluster %s: %w", appliedManifestWork.Name, err) } - return ret, nil -} - -func indexByWorkNameAndAgentID(obj interface{}) ([]string, error) { - appliedWork, ok := obj.(*workapiv1.AppliedManifestWork) - if !ok { - return []string{}, fmt.Errorf("obj is supposed to be a AppliedManifestWork, but is %T", obj) - } - - return []string{agentIDWorkNameIndex(appliedWork.Spec.ManifestWorkName, appliedWork.Spec.AgentID)}, nil -} - -func agentIDWorkNameIndex(workName, agentID string) string { - return fmt.Sprintf("%s/%s", workName, agentID) + _, err = m.appliedManifestWorkClient.Patch(ctx, appliedManifestWork.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{}, "status") + return err } diff --git a/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller_test.go b/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller_test.go index 50bba806a..803146aeb 100644 --- a/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller_test.go +++ b/pkg/spoke/controllers/finalizercontroller/unmanaged_appliedmanifestwork_controller_test.go @@ -8,7 +8,8 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" clienttesting "k8s.io/client-go/testing" - "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" + fakeworkclient "open-cluster-management.io/api/client/work/clientset/versioned/fake" workinformers "open-cluster-management.io/api/client/work/informers/externalversions" workapiv1 "open-cluster-management.io/api/work/v1" @@ -18,32 +19,61 @@ import ( func TestSyncUnamanagedAppliedWork(t *testing.T) { cases := []struct { name string - workName string + appliedManifestWorkName string hubHash string agentID string + evictionGracePeriod time.Duration works []runtime.Object appliedWorks []runtime.Object + expectedQueueLen int validateAppliedManifestWorkActions func(t *testing.T, actions []clienttesting.Action) }{ { - name: "delete applied work if unmanaged", - workName: "test", - hubHash: "hubhash1", - agentID: "test-agent", + name: "appliedmanifestwork is not found", + appliedManifestWorkName: "hubhash-test", + hubHash: "hubhash", + agentID: "test-agent", + works: []runtime.Object{}, + appliedWorks: []runtime.Object{}, + validateAppliedManifestWorkActions: noAction, + }, + { + name: "evict appliedmanifestwork when its relating manifestwork is missing on the hub", + appliedManifestWorkName: "hubhash-test", + hubHash: "hubhash", + agentID: "test-agent", + works: []runtime.Object{}, + appliedWorks: []runtime.Object{ + &workapiv1.AppliedManifestWork{ + ObjectMeta: metav1.ObjectMeta{ + Name: "hubhash-test", + }, + Spec: workapiv1.AppliedManifestWorkSpec{ + ManifestWorkName: "test", + HubHash: "hubhash", + AgentID: "test-agent", + }, + }, + }, + validateAppliedManifestWorkActions: func(t *testing.T, actions []clienttesting.Action) { + if len(actions) != 1 { + t.Errorf("Expect 1 actions on appliedmanifestwork, but have %d", len(actions)) + } + + spoketesting.AssertAction(t, actions[0], "patch") + }, + }, + { + name: "evict appliedmanifestwork after the hub switched", + appliedManifestWorkName: "hubhash-test", + hubHash: "hubhash-new", + agentID: "test-agent", works: []runtime.Object{ &workapiv1.ManifestWork{ ObjectMeta: metav1.ObjectMeta{ Name: "test", Namespace: "test", }, - Status: workapiv1.ManifestWorkStatus{ - Conditions: []metav1.Condition{ - { - Type: workapiv1.WorkApplied, - Status: metav1.ConditionTrue, - }, - }, - }, }, }, appliedWorks: []runtime.Object{ @@ -57,15 +87,44 @@ func TestSyncUnamanagedAppliedWork(t *testing.T) { AgentID: "test-agent", }, }, + }, + validateAppliedManifestWorkActions: func(t *testing.T, actions []clienttesting.Action) { + if len(actions) != 1 { + t.Errorf("Expect 1 actions on appliedmanifestwork, but have %d", len(actions)) + } + + spoketesting.AssertAction(t, actions[0], "patch") + }, + }, + { + name: "delete appliedmanifestwork after eviction grace period ", + appliedManifestWorkName: "hubhash-test", + hubHash: "hubhash-new", + agentID: "test-agent", + evictionGracePeriod: 10 * time.Minute, + works: []runtime.Object{ + &workapiv1.ManifestWork{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test", + Namespace: "test", + }, + }, + }, + appliedWorks: []runtime.Object{ &workapiv1.AppliedManifestWork{ ObjectMeta: metav1.ObjectMeta{ - Name: "hubhash1-test", + Name: "hubhash-test", }, Spec: workapiv1.AppliedManifestWorkSpec{ ManifestWorkName: "test", - HubHash: "hubhash1", + HubHash: "hubhash", AgentID: "test-agent", }, + Status: workapiv1.AppliedManifestWorkStatus{ + EvictionStartTime: &metav1.Time{ + Time: time.Now().Add(-10 * time.Minute), + }, + }, }, }, validateAppliedManifestWorkActions: func(t *testing.T, actions []clienttesting.Action) { @@ -77,10 +136,10 @@ func TestSyncUnamanagedAppliedWork(t *testing.T) { }, }, { - name: "no action if the work is not applied", - workName: "test", - hubHash: "hubhash1", - agentID: "test-agent", + name: "stop to evicte appliedmanifestwork when its relating manifestwork is recreated on the hub", + appliedManifestWorkName: "hubhash-test", + hubHash: "hubhash", + agentID: "test-agent", works: []runtime.Object{ &workapiv1.ManifestWork{ ObjectMeta: metav1.ObjectMeta{ @@ -99,41 +158,28 @@ func TestSyncUnamanagedAppliedWork(t *testing.T) { HubHash: "hubhash", AgentID: "test-agent", }, - }, - &workapiv1.AppliedManifestWork{ - ObjectMeta: metav1.ObjectMeta{ - Name: "hubhash1-test", - }, - Spec: workapiv1.AppliedManifestWorkSpec{ - ManifestWorkName: "test", - HubHash: "hubhash1", - AgentID: "test-agent", - }, - }, - }, - validateAppliedManifestWorkActions: noAction, - }, - { - name: "no action for different AgentID", - workName: "test", - hubHash: "hubhash1", - agentID: "test-agent", - works: []runtime.Object{ - &workapiv1.ManifestWork{ - ObjectMeta: metav1.ObjectMeta{ - Name: "test", - Namespace: "test", - }, - Status: workapiv1.ManifestWorkStatus{ - Conditions: []metav1.Condition{ - { - Type: workapiv1.WorkApplied, - Status: metav1.ConditionTrue, - }, + Status: workapiv1.AppliedManifestWorkStatus{ + EvictionStartTime: &metav1.Time{ + Time: time.Now(), }, }, }, }, + validateAppliedManifestWorkActions: func(t *testing.T, actions []clienttesting.Action) { + if len(actions) != 1 { + t.Errorf("Expect 1 actions on appliedmanifestwork, but have %d", len(actions)) + } + + spoketesting.AssertAction(t, actions[0], "patch") + }, + }, + { + name: "requeue eviction appliedmanifestwork", + appliedManifestWorkName: "hubhash-test", + hubHash: "hubhash", + agentID: "test-agent", + evictionGracePeriod: 10 * time.Minute, + works: []runtime.Object{}, appliedWorks: []runtime.Object{ &workapiv1.AppliedManifestWork{ ObjectMeta: metav1.ObjectMeta{ @@ -142,65 +188,16 @@ func TestSyncUnamanagedAppliedWork(t *testing.T) { Spec: workapiv1.AppliedManifestWorkSpec{ ManifestWorkName: "test", HubHash: "hubhash", - AgentID: "test-agent1", - }, - }, - &workapiv1.AppliedManifestWork{ - ObjectMeta: metav1.ObjectMeta{ - Name: "hubhash1-test", - }, - Spec: workapiv1.AppliedManifestWorkSpec{ - ManifestWorkName: "test", - HubHash: "hubhash1", AgentID: "test-agent", }, - }, - }, - validateAppliedManifestWorkActions: noAction, - }, - { - name: "no action for different work", - workName: "test", - hubHash: "hubhash1", - agentID: "test-agent", - works: []runtime.Object{ - &workapiv1.ManifestWork{ - ObjectMeta: metav1.ObjectMeta{ - Name: "test", - Namespace: "test", - }, - Status: workapiv1.ManifestWorkStatus{ - Conditions: []metav1.Condition{ - { - Type: workapiv1.WorkApplied, - Status: metav1.ConditionTrue, - }, + Status: workapiv1.AppliedManifestWorkStatus{ + EvictionStartTime: &metav1.Time{ + Time: time.Now().Add(-5 * time.Minute), }, }, }, }, - appliedWorks: []runtime.Object{ - &workapiv1.AppliedManifestWork{ - ObjectMeta: metav1.ObjectMeta{ - Name: "hubhash-test1", - }, - Spec: workapiv1.AppliedManifestWorkSpec{ - ManifestWorkName: "test1", - HubHash: "hubhash", - AgentID: "test-agent", - }, - }, - &workapiv1.AppliedManifestWork{ - ObjectMeta: metav1.ObjectMeta{ - Name: "hubhash1-test", - }, - Spec: workapiv1.AppliedManifestWorkSpec{ - ManifestWorkName: "test", - HubHash: "hubhash1", - AgentID: "test-agent", - }, - }, - }, + expectedQueueLen: 1, validateAppliedManifestWorkActions: noAction, }, } @@ -209,12 +206,6 @@ func TestSyncUnamanagedAppliedWork(t *testing.T) { t.Run(c.name, func(t *testing.T) { fakeClient := fakeworkclient.NewSimpleClientset(c.appliedWorks...) informerFactory := workinformers.NewSharedInformerFactory(fakeClient, 5*time.Minute) - err := informerFactory.Work().V1().AppliedManifestWorks().Informer().AddIndexers(cache.Indexers{ - byWorkNameAndAgentID: indexByWorkNameAndAgentID, - }) - if err != nil { - t.Fatal(err) - } for _, work := range c.works { if err := informerFactory.Work().V1().ManifestWorks().Informer().GetStore().Add(work); err != nil { t.Fatal(err) @@ -226,23 +217,28 @@ func TestSyncUnamanagedAppliedWork(t *testing.T) { } } - controller := &UnManagedAppliedWorkController{ - manifestWorkLister: informerFactory.Work().V1().ManifestWorks().Lister().ManifestWorks("test"), - appliedManifestWorkClient: fakeClient.WorkV1().AppliedManifestWorks(), - appliedManifestWorkLister: informerFactory.Work().V1().AppliedManifestWorks().Lister(), - appliedManifestWorkIndexer: informerFactory.Work().V1().AppliedManifestWorks().Informer().GetIndexer(), - hubHash: c.hubHash, - agentID: c.agentID, + controller := &unmanagedAppliedWorkController{ + manifestWorkLister: informerFactory.Work().V1().ManifestWorks().Lister().ManifestWorks("test"), + appliedManifestWorkClient: fakeClient.WorkV1().AppliedManifestWorks(), + appliedManifestWorkLister: informerFactory.Work().V1().AppliedManifestWorks().Lister(), + hubHash: c.hubHash, + agentID: c.agentID, + evictionGracePeriod: c.evictionGracePeriod, + rateLimiter: workqueue.NewItemExponentialFailureRateLimiter(0, c.evictionGracePeriod), } - controllerContext := spoketesting.NewFakeSyncContext(t, c.hubHash+"-"+c.workName) - err = controller.sync(context.TODO(), controllerContext) - if err != nil { + controllerContext := spoketesting.NewFakeSyncContext(t, c.appliedManifestWorkName) + if err := controller.sync(context.TODO(), controllerContext); err != nil { t.Errorf("Expect no sync error, but got %v", err) } appliedWorkAction := fakeClient.Actions() c.validateAppliedManifestWorkActions(t, appliedWorkAction) + + queueLen := controllerContext.Queue().Len() + if queueLen != c.expectedQueueLen { + t.Errorf("expected %d, but %d", c.expectedQueueLen, queueLen) + } }) } } diff --git a/pkg/spoke/spokeagent.go b/pkg/spoke/spokeagent.go index 9c69c34f0..e5e1d71ec 100644 --- a/pkg/spoke/spokeagent.go +++ b/pkg/spoke/spokeagent.go @@ -28,21 +28,23 @@ import ( // WorkloadAgentOptions defines the flags for workload agent type WorkloadAgentOptions struct { - HubKubeconfigFile string - SpokeKubeconfigFile string - SpokeClusterName string - QPS float32 - Burst int - StatusSyncInterval time.Duration - AgentID string + HubKubeconfigFile string + SpokeKubeconfigFile string + SpokeClusterName string + AgentID string + Burst int + StatusSyncInterval time.Duration + AppliedManifestWorkEvictionGracePeriod time.Duration + QPS float32 } // NewWorkloadAgentOptions returns the flags with default value set func NewWorkloadAgentOptions() *WorkloadAgentOptions { return &WorkloadAgentOptions{ - QPS: 50, - Burst: 100, - StatusSyncInterval: 10 * time.Second, + QPS: 50, + Burst: 100, + StatusSyncInterval: 10 * time.Second, + AppliedManifestWorkEvictionGracePeriod: 10 * time.Minute, } } @@ -59,6 +61,7 @@ func (o *WorkloadAgentOptions) AddFlags(cmd *cobra.Command) { flags.Float32Var(&o.QPS, "spoke-kube-api-qps", o.QPS, "QPS to use while talking with apiserver on spoke cluster.") flags.IntVar(&o.Burst, "spoke-kube-api-burst", o.Burst, "Burst to use while talking with apiserver on spoke cluster.") flags.DurationVar(&o.StatusSyncInterval, "status-sync-interval", o.StatusSyncInterval, "Interval to sync resource status to hub.") + flags.DurationVar(&o.AppliedManifestWorkEvictionGracePeriod, "appliedmanifestwork-eviction-grace-period", o.AppliedManifestWorkEvictionGracePeriod, "Grace period for appliedmanifestwork eviction") } // RunWorkloadAgent starts the controllers on agent to process work from hub. @@ -164,6 +167,7 @@ func (o *WorkloadAgentOptions) RunWorkloadAgent(ctx context.Context, controllerC workInformerFactory.Work().V1().ManifestWorks().Lister().ManifestWorks(o.SpokeClusterName), spokeWorkClient.WorkV1().AppliedManifestWorks(), spokeWorkInformerFactory.Work().V1().AppliedManifestWorks(), + o.AppliedManifestWorkEvictionGracePeriod, hubhash, agentID, ) appliedManifestWorkController := appliedmanifestcontroller.NewAppliedManifestWorkController( diff --git a/test/integration/unmanaged_appliedwork_test.go b/test/integration/unmanaged_appliedwork_test.go index 48786fbae..f43347ca4 100644 --- a/test/integration/unmanaged_appliedwork_test.go +++ b/test/integration/unmanaged_appliedwork_test.go @@ -29,12 +29,7 @@ var _ = ginkgo.Describe("Unmanaged ApplieManifestWork", func() { var manifests []workapiv1.Manifest var appliedManifestWorkName string var err error - - var newHubKubeConfigFile string - var newKubeClient kubernetes.Interface - var newWorkClient workclientset.Interface - var newHub *envtest.Environment - var newHubTempDir string + var ns *corev1.Namespace ginkgo.BeforeEach(func() { o = spoke.NewWorkloadAgentOptions() @@ -42,8 +37,9 @@ var _ = ginkgo.Describe("Unmanaged ApplieManifestWork", func() { o.SpokeClusterName = utilrand.String(5) o.StatusSyncInterval = 3 * time.Second o.AgentID = utilrand.String(5) + o.AppliedManifestWorkEvictionGracePeriod = 10 * time.Second - ns := &corev1.Namespace{} + ns = &corev1.Namespace{} ns.Name = o.SpokeClusterName _, err := spokeKubeClient.CoreV1().Namespaces().Create(context.Background(), ns, metav1.CreateOptions{}) gomega.Expect(err).ToNot(gomega.HaveOccurred()) @@ -52,43 +48,15 @@ var _ = ginkgo.Describe("Unmanaged ApplieManifestWork", func() { ctx, cancel = context.WithCancel(context.Background()) go startWorkAgent(ctx, o) - // start another hub - newHub = &envtest.Environment{ - ErrorIfCRDPathMissing: true, - CRDDirectoryPaths: []string{ - filepath.Join(".", "deploy", "hub"), - filepath.Join(".", "deploy", "spoke"), - }, + manifests = []workapiv1.Manifest{ + util.ToManifest(util.NewConfigmap(o.SpokeClusterName, "cm1", map[string]string{"a": "b"}, nil)), } - newCfg, err := newHub.Start() - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - newHubTempDir, err = os.MkdirTemp("", "unmanaged_work_test") - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - newHubKubeConfigFile = path.Join(newHubTempDir, "kubeconfig") - err = util.CreateKubeconfigFile(newCfg, newHubKubeConfigFile) - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - newKubeClient, err = kubernetes.NewForConfig(newCfg) - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - newWorkClient, err = workclientset.NewForConfig(newCfg) - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - _, err = newKubeClient.CoreV1().Namespaces().Create(context.Background(), ns, metav1.CreateOptions{}) - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - - // reset manifests - manifests = nil - }) - - ginkgo.JustBeforeEach(func() { work = util.NewManifestWork(o.SpokeClusterName, "unmanaged-appliedwork", manifests) _, err = hubWorkClient.WorkV1().ManifestWorks(o.SpokeClusterName).Create(context.Background(), work, metav1.CreateOptions{}) - appliedManifestWorkName = fmt.Sprintf("%s-%s", hubHash, work.Name) gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + appliedManifestWorkName = fmt.Sprintf("%s-%s", hubHash, work.Name) }) ginkgo.AfterEach(func() { @@ -97,17 +65,53 @@ var _ = ginkgo.Describe("Unmanaged ApplieManifestWork", func() { } err := spokeKubeClient.CoreV1().Namespaces().Delete(context.Background(), o.SpokeClusterName, metav1.DeleteOptions{}) gomega.Expect(err).ToNot(gomega.HaveOccurred()) - err = newHub.Stop() - gomega.Expect(err).ToNot(gomega.HaveOccurred()) - if newHubTempDir != "" { - os.RemoveAll(newHubTempDir) - } }) - ginkgo.Context("Should delete unmanaged applied work", func() { + ginkgo.Context("Should delete unmanaged applied work when hub changed", func() { + var newHubKubeConfigFile string + var newKubeClient kubernetes.Interface + var newWorkClient workclientset.Interface + var newHub *envtest.Environment + var newHubTempDir string + ginkgo.BeforeEach(func() { - manifests = []workapiv1.Manifest{ - util.ToManifest(util.NewConfigmap(o.SpokeClusterName, "cm1", map[string]string{"a": "b"}, nil)), + // start another hub + newHub = &envtest.Environment{ + ErrorIfCRDPathMissing: true, + CRDDirectoryPaths: []string{ + filepath.Join(".", "deploy", "hub"), + filepath.Join(".", "deploy", "spoke"), + }, + } + + newCfg, err := newHub.Start() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + newHubTempDir, err = os.MkdirTemp("", "unmanaged_work_test") + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + newHubKubeConfigFile = path.Join(newHubTempDir, "kubeconfig") + err = util.CreateKubeconfigFile(newCfg, newHubKubeConfigFile) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + newKubeClient, err = kubernetes.NewForConfig(newCfg) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + newWorkClient, err = workclientset.NewForConfig(newCfg) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + _, err = newKubeClient.CoreV1().Namespaces().Create(context.Background(), ns, metav1.CreateOptions{}) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + }) + + ginkgo.AfterEach(func() { + if cancel != nil { + cancel() + } + err := newHub.Stop() + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + if newHubTempDir != "" { + os.RemoveAll(newHubTempDir) } }) @@ -128,6 +132,7 @@ var _ = ginkgo.Describe("Unmanaged ApplieManifestWork", func() { newOption.HubKubeconfigFile = newHubKubeConfigFile newOption.SpokeClusterName = o.SpokeClusterName newOption.AgentID = utilrand.String(5) + newOption.AppliedManifestWorkEvictionGracePeriod = 5 * time.Second var ctx context.Context ctx, cancel = context.WithCancel(context.Background()) @@ -172,6 +177,7 @@ var _ = ginkgo.Describe("Unmanaged ApplieManifestWork", func() { newOption.HubKubeconfigFile = newHubKubeConfigFile newOption.SpokeClusterName = o.SpokeClusterName newOption.AgentID = o.AgentID + newOption.AppliedManifestWorkEvictionGracePeriod = 5 * time.Second var ctx context.Context ctx, cancel = context.WithCancel(context.Background()) @@ -215,4 +221,82 @@ var _ = ginkgo.Describe("Unmanaged ApplieManifestWork", func() { }) }) + ginkgo.Context("Should evict applied work when its manifestwork is missing on the hub", func() { + ginkgo.BeforeEach(func() { + util.AssertExistenceOfConfigMaps(manifests, spokeKubeClient, eventuallyTimeout, eventuallyInterval) + util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkApplied), metav1.ConditionTrue, + []metav1.ConditionStatus{metav1.ConditionTrue}, eventuallyTimeout, eventuallyInterval) + util.AssertWorkCondition(work.Namespace, work.Name, hubWorkClient, string(workapiv1.WorkAvailable), metav1.ConditionTrue, + []metav1.ConditionStatus{metav1.ConditionTrue}, eventuallyTimeout, eventuallyInterval) + + // stop the agent + if cancel != nil { + cancel() + } + + // fore delete the work from hub + err := forceDeleteManifestWork(context.TODO(), work) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + }) + + ginkgo.It("should evict appliemanifestwork after the work agent reconnected to the hub", func() { + // restart the work agent + var ctx context.Context + ctx, cancel = context.WithCancel(context.Background()) + go startWorkAgent(ctx, o) + + // ensure the manifestwork is removed. + gomega.Eventually(func() error { + _, err := spokeWorkClient.WorkV1().AppliedManifestWorks().Get(context.TODO(), appliedManifestWorkName, metav1.GetOptions{}) + if errors.IsNotFound(err) { + return nil + } + if err != nil { + return err + } + return fmt.Errorf("appliedmanifestwork %s still exists", appliedManifestWorkName) + }, eventuallyTimeout, eventuallyInterval).ShouldNot(gomega.HaveOccurred()) + }) + + ginkgo.It("should keep appliemanifestwork if its manifestwork is recreated on new hub", func() { + // restart the work agent + var ctx context.Context + ctx, cancel = context.WithCancel(context.Background()) + go startWorkAgent(ctx, o) + + // recreate the work on the hub + _, err = hubWorkClient.WorkV1().ManifestWorks(o.SpokeClusterName).Create(context.Background(), work, metav1.CreateOptions{}) + gomega.Expect(err).ToNot(gomega.HaveOccurred()) + + // ensure the appliemanifestwork eviction is stopped + gomega.Eventually(func() error { + appliedWork, err := spokeWorkClient.WorkV1().AppliedManifestWorks().Get(context.TODO(), appliedManifestWorkName, metav1.GetOptions{}) + if err != nil { + return err + } + + if appliedWork.Status.EvictionStartTime != nil { + return fmt.Errorf("the appliedmanifestwork %s evication is not stopped", appliedManifestWorkName) + } + return nil + }, eventuallyTimeout, eventuallyInterval).ShouldNot(gomega.HaveOccurred()) + + }) + }) }) + +func forceDeleteManifestWork(ctx context.Context, work *workapiv1.ManifestWork) error { + err := hubWorkClient.WorkV1().ManifestWorks(work.Namespace).Delete(ctx, work.Name, metav1.DeleteOptions{}) + if err != nil { + return err + } + + deleted, err := hubWorkClient.WorkV1().ManifestWorks(work.Namespace).Get(ctx, work.Name, metav1.GetOptions{}) + if errors.IsNotFound(err) { + return nil + } + + deleted.Finalizers = []string{} + _, err = hubWorkClient.WorkV1().ManifestWorks(work.Namespace).Update(ctx, deleted, metav1.UpdateOptions{}) + return err +} diff --git a/test/integration/work_test.go b/test/integration/work_test.go index 9a16de798..9400c2980 100644 --- a/test/integration/work_test.go +++ b/test/integration/work_test.go @@ -46,6 +46,7 @@ var _ = ginkgo.Describe("ManifestWork", func() { o.HubKubeconfigFile = hubKubeconfigFileName o.SpokeClusterName = utilrand.String(5) o.StatusSyncInterval = 3 * time.Second + o.AppliedManifestWorkEvictionGracePeriod = 5 * time.Second ns := &corev1.Namespace{} ns.Name = o.SpokeClusterName @@ -396,11 +397,23 @@ var _ = ginkgo.Describe("ManifestWork", func() { return errors.IsNotFound(err) }, eventuallyTimeout, eventuallyInterval).Should(gomega.BeTrue()) - // Once manifest work is deleted, all CRs/CRD should have already been deleted too - for i := range gvrs { - _, err := util.GetResource(namespaces[i], names[i], gvrs[i], spokeDynamicClient) - gomega.Expect(errors.IsNotFound(err)).To(gomega.BeTrue()) - } + // Once manifest work is not found, its relating appliedmanifestwork will be evicted, and finally, + // all CRs/CRD should been deleted too + gomega.Eventually(func() error { + for i := range gvrs { + _, err := util.GetResource(namespaces[i], names[i], gvrs[i], spokeDynamicClient) + if errors.IsNotFound(err) { + continue + } + if err != nil { + return err + } + + return fmt.Errorf("the resource %s/%s still exists", namespaces[i], names[i]) + } + + return nil + }, eventuallyTimeout, eventuallyInterval).ShouldNot(gomega.HaveOccurred()) }) }) @@ -673,6 +686,7 @@ var _ = ginkgo.Describe("ManifestWork", func() { }, Spec: workapiv1.AppliedManifestWorkSpec{ HubHash: hubHash, + AgentID: hubHash, ManifestWorkName: "fakework", }, }