appliedmanifestwork eviction (#190)

Signed-off-by: Wei Liu <liuweixa@redhat.com>

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:

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"

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():

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,
 		},
 	}
 

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
 }

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)
+			}
 		})
 	}
 }

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(

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
+}

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",
 				},
 			}