diff --git a/pkg/resourcekeeper/dispatch.go b/pkg/resourcekeeper/dispatch.go index ba48dd37d..28604e072 100644 --- a/pkg/resourcekeeper/dispatch.go +++ b/pkg/resourcekeeper/dispatch.go @@ -161,6 +161,9 @@ func (h *resourceKeeper) dispatch(ctx context.Context, manifests []*unstructured if err != nil { return errors.Wrapf(err, "failed to apply once policy for application %s,%s", h.app.Name, err.Error()) } + if manifest == nil { + return nil + } return h.applicator.Apply(applyCtx, manifest, ao...) }, velaslices.Parallelism(MaxDispatchConcurrent)) return velaerrors.AggregateErrors(errs) diff --git a/pkg/resourcekeeper/dispatch_and_delete_test.go b/pkg/resourcekeeper/dispatch_and_delete_test.go index ef928ce31..b65bd5dec 100644 --- a/pkg/resourcekeeper/dispatch_and_delete_test.go +++ b/pkg/resourcekeeper/dispatch_and_delete_test.go @@ -18,8 +18,10 @@ package resourcekeeper import ( "context" + "fmt" "testing" + "github.com/crossplane/crossplane-runtime/pkg/test" "github.com/stretchr/testify/require" v1 "k8s.io/api/core/v1" v12 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -103,3 +105,80 @@ func TestResourceKeeperAdmissionDispatchAndDelete(t *testing.T) { r.NotNil(err) r.Contains(err.Error(), "forbidden") } + +// TestApplyStrategiesNilReturnOnStateKeep verifies that ApplyStrategies returns nil +// when called with ApplyOnceStrategyOnAppStateKeep and the resource is not found. +// This is the precondition for the nil-guard in the dispatch path being correct: +// the dispatch path uses ApplyOnceStrategyOnAppUpdate, which never produces nil, +// so the guard there is purely defensive. +func TestApplyStrategiesNilReturnOnStateKeep(t *testing.T) { + r := require.New(t) + cli := fake.NewClientBuilder().WithScheme(common.Scheme).Build() + + app := &v1beta1.Application{ObjectMeta: v12.ObjectMeta{Name: "app", Namespace: "default"}} + rk := &resourceKeeper{ + Client: cli, + app: app, + applyOncePolicy: &v1alpha1.ApplyOncePolicySpec{ + Enable: true, + Rules: []v1alpha1.ApplyOncePolicyRule{{ + Selector: v1alpha1.ResourcePolicyRuleSelector{ + CompNames: []string{"my-comp"}, + }, + Strategy: &v1alpha1.ApplyOnceStrategy{Path: []string{"*"}}, + }}, + }, + } + + manifest := &unstructured.Unstructured{} + manifest.SetGroupVersionKind(v1.SchemeGroupVersion.WithKind("ConfigMap")) + manifest.SetName("nonexistent-cm") + manifest.SetNamespace("default") + manifest.SetLabels(map[string]string{oam.LabelAppComponent: "my-comp"}) + + // For ApplyOnceStrategyOnAppStateKeep, a missing resource returns nil. + result, err := ApplyStrategies(context.Background(), rk, manifest, v1alpha1.ApplyOnceStrategyOnAppStateKeep) + r.NoError(err) + r.Nil(result) + + // For ApplyOnceStrategyOnAppUpdate, a missing resource returns the original manifest (not nil). + // This means the nil-guard in dispatch.go is defensive and cannot be triggered today. + result, err = ApplyStrategies(context.Background(), rk, manifest, v1alpha1.ApplyOnceStrategyOnAppUpdate) + r.NoError(err) + r.NotNil(result) +} + +// TestCleanupStaleEntriesUpdateError verifies that cleanupStaleEntries propagates +// errors from the underlying client Update call. +func TestCleanupStaleEntriesUpdateError(t *testing.T) { + r := require.New(t) + updateErr := fmt.Errorf("simulated update failure") + cli := &test.MockClient{ + MockUpdate: test.NewMockUpdateFn(updateErr), + } + + app := &v1beta1.Application{ObjectMeta: v12.ObjectMeta{Name: "app", Namespace: "default"}} + rk := &resourceKeeper{ + Client: cli, + app: app, + } + + rt := &v1beta1.ResourceTracker{ + ObjectMeta: v12.ObjectMeta{Name: "test-rt", UID: "test-uid"}, + } + cm := &unstructured.Unstructured{} + cm.SetGroupVersionKind(v1.SchemeGroupVersion.WithKind("ConfigMap")) + cm.SetName("stale-cm") + cm.SetNamespace("default") + + mr := v1beta1.ManagedResource{} + mr.APIVersion = v1.SchemeGroupVersion.String() + mr.Kind = "ConfigMap" + mr.Name = "stale-cm" + mr.Namespace = "default" + + entries := []staleEntry{{mr: mr, rt: rt}} + err := rk.cleanupStaleEntries(context.Background(), entries) + r.Error(err) + r.Contains(err.Error(), "failed to remove stale entries from resourcetracker test-rt") +} diff --git a/pkg/resourcekeeper/statekeep.go b/pkg/resourcekeeper/statekeep.go index b97131d9a..9a6905e2f 100644 --- a/pkg/resourcekeeper/statekeep.go +++ b/pkg/resourcekeeper/statekeep.go @@ -18,6 +18,7 @@ package resourcekeeper import ( "context" + "sync" "github.com/crossplane/crossplane-runtime/pkg/fieldpath" "github.com/kubevela/pkg/util/maps" @@ -53,6 +54,8 @@ func (h *resourceKeeper) StateKeep(ctx context.Context) error { } } } + var stalesMu sync.Mutex + var staleEntries []staleEntry errs := slices.ParMap(maps.Values(mrs), func(mr v1beta1.ManagedResource) error { rt := belongs[mr.ResourceKey()] entry := h.cache.get(ctx, mr) @@ -81,6 +84,9 @@ func (h *resourceKeeper) StateKeep(ctx context.Context) error { return errors.Wrapf(err, "failed to apply once resource %s from resourcetracker %s", mr.ResourceKey(), rt.Name) } if manifest == nil { + stalesMu.Lock() + staleEntries = append(staleEntries, staleEntry{mr: mr, rt: rt}) + stalesMu.Unlock() return nil } ao := []apply.ApplyOption{apply.MustBeControlledByApp(h.app)} @@ -102,6 +108,45 @@ func (h *resourceKeeper) StateKeep(ctx context.Context) error { } return nil }, slices.Parallelism(MaxDispatchConcurrent)) + if err := h.cleanupStaleEntries(ctx, staleEntries); err != nil { + errs = append(errs, err) + } + return velaerrors.AggregateErrors(errs) +} + +// staleEntry records a managed resource whose backing object no longer exists +// and should be removed from its ResourceTracker. +type staleEntry struct { + mr v1beta1.ManagedResource + rt *v1beta1.ResourceTracker +} + +// cleanupStaleEntries removes managed resources from their ResourceTrackers +// when the underlying resource no longer exists (e.g. externally deleted +// apply-once resources). This avoids unnecessary API server GETs on future +// StateKeep cycles. +func (h *resourceKeeper) cleanupStaleEntries(ctx context.Context, entries []staleEntry) error { + if len(entries) == 0 { + return nil + } + // Group by ResourceTracker to batch updates + rtUpdates := make(map[string]*v1beta1.ResourceTracker) + rtRemovals := make(map[string][]v1beta1.ManagedResource) + for _, e := range entries { + key := string(e.rt.UID) + rtUpdates[key] = e.rt + rtRemovals[key] = append(rtRemovals[key], e.mr) + } + var errs []error + for key, rt := range rtUpdates { + for _, mr := range rtRemovals[key] { + obj := mr.ToUnstructured() + rt.DeleteManagedResource(obj, true) + } + if err := h.Client.Update(multicluster.ContextInLocalCluster(ctx), rt); err != nil { + errs = append(errs, errors.Wrapf(err, "failed to remove stale entries from resourcetracker %s", rt.Name)) + } + } return velaerrors.AggregateErrors(errs) } diff --git a/pkg/resourcekeeper/statekeep_suite_test.go b/pkg/resourcekeeper/statekeep_suite_test.go index 59f163461..17b0ce536 100644 --- a/pkg/resourcekeeper/statekeep_suite_test.go +++ b/pkg/resourcekeeper/statekeep_suite_test.go @@ -171,7 +171,7 @@ var _ = Describe("Test ResourceKeeper StateKeep", func() { Expect(err.Error()).Should(ContainSubstring("failed to re-apply")) }) - It("Test StateKeep apply-once does not re-create externally deleted resource", func() { + It("Test StateKeep apply-once does not re-create externally deleted resource and cleans up stale RT entry", func() { cli := testClient // Resource tracked in the resource tracker but not present in the cluster, @@ -202,8 +202,19 @@ var _ = Describe("Test ResourceKeeper StateKeep", func() { }}, }, } - h._currentRT = &v1beta1.ResourceTracker{ + + // Create the ResourceTracker in the API server so the stale-entry + // cleanup can persist its update. + rt := &v1beta1.ResourceTracker{ + ObjectMeta: metav1.ObjectMeta{ + Name: "app-apply-once-gc-rt", + Labels: map[string]string{ + oam.LabelAppName: "app-apply-once-gc", + oam.LabelAppNamespace: "default", + }, + }, Spec: v1beta1.ResourceTrackerSpec{ + Type: v1beta1.ResourceTrackerTypeVersioned, ManagedResources: []v1beta1.ManagedResource{{ ClusterObjectReference: createConfigMapClusterObjectReference("cm-apply-once-gc"), OAMObjectReference: common.NewOAMObjectReferenceFromObject(cm), @@ -211,6 +222,8 @@ var _ = Describe("Test ResourceKeeper StateKeep", func() { }}, }, } + Expect(cli.Create(context.Background(), rt)).Should(Succeed()) + h._currentRT = rt Expect(h.StateKeep(context.Background())).Should(Succeed()) @@ -220,6 +233,11 @@ var _ = Describe("Test ResourceKeeper StateKeep", func() { err = cli.Get(context.Background(), client.ObjectKeyFromObject(cm), got) Expect(err).Should(HaveOccurred()) Expect(kerrors.IsNotFound(err)).Should(BeTrue()) + + // Verify the stale entry was removed from the ResourceTracker + updatedRT := &v1beta1.ResourceTracker{} + Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(rt), updatedRT)).Should(Succeed()) + Expect(updatedRT.Spec.ManagedResources).Should(BeEmpty()) }) It("Test StateKeep for shared resources", func() {