refactor: improve gc (#1292)

GC will run every day at 00:00.
GC will delete all VMOPs in completed state if their TTL has expired (24 hours).
GC will delete VMOPs in completed state even if their TTL has not expired, provided there are more than 10 such VMOPs, keeping only the latest 10.
---------

Signed-off-by: Yaroslav Borbat <[email protected]>
Signed-off-by: Yaroslav Borbat <[email protected]>
Co-authored-by: Ivan Mikheykin <[email protected]>

Author: 2 people

images/virtualization-artifact/pkg/controller/gc/cron_source.go

@@ -17,157 +17,90 @@ limitations under the License.
 package gc
 
 import (
-	"cmp"
 	"context"
 	"fmt"
-	"slices"
 	"time"
 
 	"github.com/robfig/cron/v3"
-	"k8s.io/apimachinery/pkg/api/meta"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/workqueue"
+	"k8s.io/utils/clock"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
-	"github.com/deckhouse/virtualization-controller/pkg/common/object"
 	"github.com/deckhouse/virtualization-controller/pkg/logger"
 )
 
-const sourceName = "CronSource"
-
-func NewCronSource(c client.Client,
-	standardSpec string,
-	objList client.ObjectList,
-	option CronSourceOption,
-	log *log.Logger,
-) *CronSource {
-	return &CronSource{
-		Client:       c,
-		standardSpec: standardSpec,
-		objList:      objList,
-		option:       option,
-		log:          log.With("WatchSource", sourceName),
-	}
-}
-
 var _ source.Source = &CronSource{}
 
-type CronSource struct {
-	client.Client
-	standardSpec string
-	objList      client.ObjectList
-	option       CronSourceOption
-	log          *log.Logger
-}
+const sourceName = "CronSource"
 
-type CronSourceOption struct {
-	GetOlder func(objList client.ObjectList) client.ObjectList
+type SourceGCManager interface {
+	ListForDelete(ctx context.Context, now time.Time) ([]client.Object, error)
 }
 
-func NewDefaultCronSourceOption(objs client.ObjectList, ttl time.Duration, log *log.Logger) CronSourceOption {
-	return CronSourceOption{
-		GetOlder: DefaultGetOlder(objs, ttl, 10, log),
+func NewCronSource(scheduleSpec string, mgr SourceGCManager, log *log.Logger) (*CronSource, error) {
+	schedule, err := cron.ParseStandard(scheduleSpec)
+	if err != nil {
+		return nil, fmt.Errorf("parsing standard spec %q: %w", scheduleSpec, err)
 	}
-}
 
-func DefaultGetOlder(objs client.ObjectList, ttl time.Duration, maxCount int, log *log.Logger) func(objList client.ObjectList) client.ObjectList {
-	return func(objList client.ObjectList) client.ObjectList {
-		var expiredItems []runtime.Object
-		var notExpiredItems []runtime.Object
-
-		if err := meta.EachListItem(objList, func(o runtime.Object) error {
-			obj, ok := o.(client.Object)
-			if !ok {
-				return nil
-			}
-			if object.GetAge(obj) > ttl {
-				expiredItems = append(expiredItems, o)
-			} else {
-				notExpiredItems = append(notExpiredItems, o)
-			}
-
-			return nil
-		}); err != nil {
-			log.Error("failed to populate list", logger.SlogErr(err))
-		}
-
-		if maxCount != 0 && len(notExpiredItems) > maxCount {
-			slices.SortFunc(notExpiredItems, func(a, b runtime.Object) int {
-				aObj, _ := a.(client.Object)
-				bObj, _ := b.(client.Object)
-
-				return cmp.Compare(object.GetAge(aObj), object.GetAge(bObj))
-			})
-			expiredItems = append(expiredItems, notExpiredItems[maxCount:]...)
-		}
+	return &CronSource{
+		schedule: schedule,
+		mgr:      mgr,
+		log:      log.With("WatchSource", sourceName),
+		clock:    &clock.RealClock{},
+	}, nil
+}
 
-		if err := meta.SetList(objs, expiredItems); err != nil {
-			log.Error("failed to set list", logger.SlogErr(err))
-		}
-		return objs
-	}
+type CronSource struct {
+	schedule cron.Schedule
+	mgr      SourceGCManager
+	log      *log.Logger
+	clock    clock.Clock
 }
 
 func (c *CronSource) Start(ctx context.Context, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) error {
-	schedule, err := cron.ParseStandard(c.standardSpec)
-	if err != nil {
-		return fmt.Errorf("parsing standard spec %q: %w", c.standardSpec, err)
-	}
-	work := func() {
-		if err = meta.SetList(c.objList, nil); err != nil {
-			c.log.Error("failed to reset resource list", logger.SlogErr(err))
-			return
-		}
-		if err = c.List(ctx, c.objList); err != nil {
-			c.log.Error("failed to listing resources", logger.SlogErr(err))
-			return
-		}
-		if meta.LenList(c.objList) == 0 {
-			c.log.Debug("no resources, skip")
-			return
-		}
-		if c.option.GetOlder != nil {
-			c.objList = c.option.GetOlder(c.objList)
-		}
-		if err = meta.EachListItem(c.objList, func(object runtime.Object) error {
-			obj, ok := object.(client.Object)
-			if !ok {
-				c.log.Error(fmt.Sprintf("%s's type isn't metav1.Object", object.GetObjectKind().GroupVersionKind().String()))
-				return nil
-			}
-
-			queue.Add(reconcile.Request{
-				NamespacedName: types.NamespacedName{
-					Namespace: obj.GetNamespace(),
-					Name:      obj.GetName(),
-				},
-			})
-			c.log.Debug(fmt.Sprintf("resource %s/%s enqueued", obj.GetNamespace(), obj.GetName()))
-			return nil
-		}); err != nil {
-			c.log.Error("failed to enqueueing resources", logger.SlogErr(err))
-			return
-		}
-	}
-	ta := nextScheduleTimeDuration(schedule, time.Now())
+	nextTime := nextScheduleTimeDuration(c.schedule, c.clock.Now())
 	go func() {
 		for {
 			select {
 			case <-ctx.Done():
 				return
-			case <-time.After(ta):
-				work()
-				ta = nextScheduleTimeDuration(schedule, time.Now())
+			case <-c.clock.After(nextTime):
+				c.addObjects(ctx, queue.Add)
+				nextTime = nextScheduleTimeDuration(c.schedule, c.clock.Now())
 			}
 		}
 	}()
 	return nil
 }
 
+func (c *CronSource) addObjects(ctx context.Context, addToQueue func(reconcile.Request)) {
+	objs, err := c.mgr.ListForDelete(ctx, c.clock.Now())
+	if err != nil {
+		c.log.Error("Failed to get ObjectList for delete", logger.SlogErr(err))
+		return
+	}
+
+	if len(objs) == 0 {
+		c.log.Debug("No resources, skip")
+		return
+	}
+
+	for _, obj := range objs {
+		addToQueue(reconcile.Request{
+			NamespacedName: types.NamespacedName{
+				Namespace: obj.GetNamespace(),
+				Name:      obj.GetName(),
+			},
+		})
+		c.log.Debug(fmt.Sprintf("Resource %s/%s enqueued", obj.GetNamespace(), obj.GetName()))
+	}
+}
+
 func nextScheduleTimeDuration(schedule cron.Schedule, now time.Time) time.Duration {
 	return schedule.Next(now).Sub(now)
 }

images/virtualization-artifact/pkg/controller/gc/cron_source_test.go

@@ -0,0 +1,137 @@
+/*
+Copyright 2025 Flant JSC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package gc
+
+import (
+	"context"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	clock "k8s.io/utils/clock/testing"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	dlog "github.com/deckhouse/deckhouse/pkg/log"
+	"github.com/deckhouse/virtualization-controller/pkg/common/testutil"
+)
+
+var _ = Describe("CronSource", func() {
+	const (
+		// Every day a 00:00
+		scheduleSpec = "0 0 * * *"
+	)
+
+	var (
+		log        *dlog.Logger
+		baseCtx    context.Context
+		fakeClient client.Client
+		mgr        *fakeGCManager
+		fakeClock  *clock.FakeClock
+	)
+
+	BeforeEach(func() {
+		log = testutil.NewNoOpLogger()
+		baseCtx = testutil.ToContext(context.Background(), log)
+
+		scheme := newScheme()
+		fakeClient = fake.NewClientBuilder().WithScheme(scheme).Build()
+
+		mgr = newFakeGCManager(fakeClient, 24*time.Hour, 10)
+
+		t := time.Date(2025, 1, 1, 12, 0, 0, 0, time.UTC)
+		fakeClock = clock.NewFakeClock(t)
+	})
+
+	newSource := func(scheduleSpec string) *CronSource {
+		source, err := NewCronSource(scheduleSpec, mgr, log)
+		Expect(err).NotTo(HaveOccurred())
+		source.clock = fakeClock
+		return source
+	}
+
+	Context("with spawned objects", func() {
+		var (
+			ctx    context.Context
+			cancel context.CancelFunc
+			source *CronSource
+			queue  *fakeQueue
+		)
+
+		BeforeEach(func() {
+			ctx, cancel = context.WithCancel(baseCtx)
+			source = newSource(scheduleSpec)
+			queue = newFakeQueue()
+		})
+
+		AfterEach(func() {
+			cancel()
+			queue.ShutDown()
+		})
+
+		It("should not enqueue objects because ttl is not expired", func() {
+			spawnFakeObjects(5, 10, fakeObjectPhaseCompleted, fakeClient, fakeClock)
+			spawnFakeObjects(10, 10, fakeObjectPhasePending, fakeClient, fakeClock)
+			spawnFakeObjects(15, 10, fakeObjectPhaseRunning, fakeClient, fakeClock)
+
+			Expect(source.Start(ctx, queue)).To(Succeed())
+			time.Sleep(1 * time.Second)
+
+			// Go to 2025-01-02 01:00.
+			// CronSource should be started but not enqueued any objects because ttl is not expired.
+			fakeClock.Step(13 * time.Hour)
+
+			Consistently(func() int {
+				return len(queue.Requests())
+			}).WithTimeout(10 * time.Second).Should(Equal(0))
+		})
+
+		It("should enqueue 10 objects because ttl is not expired but objects completed and them more that maxCount", func() {
+			spawnFakeObjects(10, 11, fakeObjectPhaseCompleted, fakeClient, fakeClock)
+			spawnFakeObjects(10, 11, fakeObjectPhasePending, fakeClient, fakeClock)
+			spawnFakeObjects(10, 11, fakeObjectPhaseRunning, fakeClient, fakeClock)
+
+			Expect(source.Start(ctx, queue)).To(Succeed())
+			time.Sleep(1 * time.Second)
+
+			// Go to 2025-01-02 01:00.
+			// CronSource should be started and enqueued 10 objects.
+			fakeClock.Step(13 * time.Hour)
+
+			Eventually(func() int {
+				return len(queue.Requests())
+			}).WithTimeout(10 * time.Second).Should(Equal(10))
+		})
+
+		It("should enqueue 100 objects in completed state", func() {
+			spawnFakeObjects(10, 10, fakeObjectPhaseCompleted, fakeClient, fakeClock)
+			spawnFakeObjects(10, 10, fakeObjectPhasePending, fakeClient, fakeClock)
+			spawnFakeObjects(10, 10, fakeObjectPhaseRunning, fakeClient, fakeClock)
+
+			Expect(source.Start(ctx, queue)).To(Succeed())
+			time.Sleep(1 * time.Second)
+
+			// Go to 2025-01-03 01:00.
+			// CronSource should be started and enqueued 100 objects.
+			fakeClock.Step(37 * time.Hour)
+
+			Eventually(func() int {
+				return len(queue.Requests())
+			}).WithTimeout(10 * time.Second).Should(Equal(100))
+		})
+	})
+})