Add integration test for shardlease controller (#488)

* Update `controllerring` integration test

* Add integration test for `shardlease` controller

Author: timebertt

pkg/controller/shardlease/add.go

@@ -18,13 +18,15 @@ package shardlease
 
 import (
 	coordinationv1 "k8s.io/api/coordination/v1"
+	"k8s.io/client-go/util/workqueue"
 	"k8s.io/utils/clock"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	shardingv1alpha1 "github.com/timebertt/kubernetes-controller-sharding/pkg/apis/sharding/v1alpha1"
 	shardinghandler "github.com/timebertt/kubernetes-controller-sharding/pkg/sharding/handler"
@@ -54,6 +56,12 @@ func (r *Reconciler) AddToManager(mgr manager.Manager) error {
 		).
 		WithOptions(controller.Options{
 			MaxConcurrentReconciles: 5,
+			NewQueue: func(controllerName string, rateLimiter workqueue.TypedRateLimiter[reconcile.Request]) workqueue.TypedRateLimitingInterface[reconcile.Request] {
+				return workqueue.NewTypedRateLimitingQueueWithConfig(rateLimiter, workqueue.TypedRateLimitingQueueConfig[reconcile.Request]{
+					Name:  controllerName,
+					Clock: r.Clock,
+				})
+			},
 		}).
 		Complete(r)
 }

pkg/controller/shardlease/add_test.go

@@ -44,11 +44,11 @@ var _ = Describe("Reconciler", func() {
 			p           predicate.Predicate
 			obj, objOld *coordinationv1.Lease
 
-			fakeClock *testing.FakePassiveClock
+			fakeClock *testing.FakeClock
 		)
 
 		BeforeEach(func() {
-			fakeClock = testing.NewFakePassiveClock(time.Now())
+			fakeClock = testing.NewFakeClock(time.Now())
 			r.Clock = fakeClock
 
 			p = r.LeasePredicate()

pkg/controller/shardlease/reconciler.go

@@ -39,7 +39,7 @@ import (
 // Reconciler reconciles shard leases.
 type Reconciler struct {
 	Client client.Client
-	Clock  clock.PassiveClock
+	Clock  clock.WithTicker
 }
 
 // Reconcile reconciles a Lease object.

test/integration/sharder/controller/controllerring/controllerring_suite_test.go

@@ -24,7 +24,6 @@ import (
 	"github.com/go-logr/logr"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	coordinationv1 "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -51,7 +50,7 @@ func TestControllerRing(t *testing.T) {
 	RunSpecs(t, "Sharder ControllerRing Controller Integration Test Suite")
 }
 
-const testID = "controllerring-test"
+const testID = "controllerring-controller-test"
 
 var (
 	log logr.Logger
@@ -60,7 +59,6 @@ var (
 
 	clock *testclock.FakePassiveClock
 
-	testNamespace *corev1.Namespace
 	testRunID     string
 	testRunLabels map[string]string
 )
@@ -97,7 +95,7 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 	DeferCleanup(clientCancel)
 
 	By("Create test Namespace")
-	testNamespace = &corev1.Namespace{
+	testNamespace := &corev1.Namespace{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: testID + "-",
 		},
@@ -123,9 +121,6 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 				&shardingv1alpha1.ControllerRing{}: {
 					Label: labels.SelectorFromSet(testRunLabels),
 				},
-				&coordinationv1.Lease{}: {
-					Label: labels.SelectorFromSet(testRunLabels),
-				},
 			},
 		},
 	})

test/integration/sharder/controller/controllerring/controllerring_test.go

@@ -102,38 +102,38 @@ var _ = Describe("ControllerRing controller", func() {
 	Describe("should reflect the shard leases in the status", Ordered, func() {
 		var lease *coordinationv1.Lease
 
-		It("Create available shard lease", func(ctx SpecContext) {
+		It("create available shard lease", func(ctx SpecContext) {
 			lease = newLease(controllerRing.Name)
 			Expect(testClient.Create(ctx, lease)).To(Succeed())
 
 			Eventually(ctx, Object(controllerRing)).Should(haveStatusShards(1, 1))
 		}, SpecTimeout(time.Minute))
 
-		It("Create orphaned shard lease", func(ctx SpecContext) {
+		It("create orphaned shard lease", func(ctx SpecContext) {
 			lease = newLease(controllerRing.Name)
 			lease.Spec.HolderIdentity = nil
 			Expect(testClient.Create(ctx, lease)).To(Succeed())
 
 			Eventually(ctx, Object(controllerRing)).Should(haveStatusShards(1, 2))
 		}, SpecTimeout(time.Minute))
 
-		It("Make lease healthy", func(ctx SpecContext) {
+		It("make lease healthy", func(ctx SpecContext) {
 			Eventually(ctx, Update(lease, func() {
 				lease.Spec.HolderIdentity = ptr.To(lease.Name)
 			})).Should(Succeed())
 
 			Eventually(ctx, Object(controllerRing)).Should(haveStatusShards(2, 2))
 		}, SpecTimeout(time.Minute))
 
-		It("Make lease unhealthy", func(ctx SpecContext) {
+		It("make lease unhealthy", func(ctx SpecContext) {
 			Eventually(ctx, Update(lease, func() {
 				lease.Spec.HolderIdentity = nil
 			})).Should(Succeed())
 
 			Eventually(ctx, Object(controllerRing)).Should(haveStatusShards(1, 2))
 		}, SpecTimeout(time.Minute))
 
-		It("Delete unhealthy lease", func(ctx SpecContext) {
+		It("delete unhealthy lease", func(ctx SpecContext) {
 			Expect(testClient.Delete(ctx, lease)).To(Succeed())
 
 			Eventually(ctx, Object(controllerRing)).Should(haveStatusShards(1, 1))
@@ -144,11 +144,13 @@ var _ = Describe("ControllerRing controller", func() {
 func newLease(controllerRingName string) *coordinationv1.Lease {
 	name := testRunID + "-" + test.RandomSuffix()
 
-	lease := &coordinationv1.Lease{
+	return &coordinationv1.Lease{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: testRunID,
-			Labels:    maps.Clone(testRunLabels),
+			Labels: map[string]string{
+				shardingv1alpha1.LabelControllerRing: controllerRingName,
+			},
 		},
 		Spec: coordinationv1.LeaseSpec{
 			HolderIdentity:       ptr.To(name),
@@ -157,9 +159,6 @@ func newLease(controllerRingName string) *coordinationv1.Lease {
 			RenewTime:            ptr.To(metav1.NewMicroTime(clock.Now().Add(-2 * time.Second))),
 		},
 	}
-	metav1.SetMetaDataLabel(&lease.ObjectMeta, shardingv1alpha1.LabelControllerRing, controllerRingName)
-
-	return lease
 }
 
 func haveStatusShards(availableShards, shards int32) gomegatypes.GomegaMatcher {

test/integration/sharder/controller/shardlease/shardlease_suite_test.go

@@ -0,0 +1,151 @@
+/*
+Copyright 2025 Tim Ebert.
+
+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 shardlease_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/go-logr/logr"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	testclock "k8s.io/utils/clock/testing"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/envtest/komega"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
+	configv1alpha1 "github.com/timebertt/kubernetes-controller-sharding/pkg/apis/config/v1alpha1"
+	shardingv1alpha1 "github.com/timebertt/kubernetes-controller-sharding/pkg/apis/sharding/v1alpha1"
+	"github.com/timebertt/kubernetes-controller-sharding/pkg/controller/shardlease"
+	utilclient "github.com/timebertt/kubernetes-controller-sharding/pkg/utils/client"
+	"github.com/timebertt/kubernetes-controller-sharding/pkg/utils/test"
+	. "github.com/timebertt/kubernetes-controller-sharding/pkg/utils/test/matchers"
+)
+
+func TestShardLease(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Sharder Shard Lease Controller Integration Test Suite")
+}
+
+const testID = "shardlease-controller-test"
+
+var (
+	log logr.Logger
+
+	testClient client.Client
+
+	clock *testclock.FakeClock
+
+	testRunID     string
+	testRunLabels map[string]string
+)
+
+var _ = BeforeSuite(func(ctx SpecContext) {
+	logf.SetLogger(zap.New(zap.UseDevMode(true), zap.WriteTo(GinkgoWriter)))
+	log = logf.Log.WithName(testID)
+
+	By("Start test environment")
+	testEnv := &envtest.Environment{
+		CRDInstallOptions: envtest.CRDInstallOptions{
+			Paths: []string{test.PathShardingCRDs()},
+		},
+		ErrorIfCRDPathMissing: true,
+	}
+
+	restConfig, err := testEnv.Start()
+	Expect(err).NotTo(HaveOccurred())
+	Expect(restConfig).NotTo(BeNil())
+
+	DeferCleanup(func() {
+		By("Stop test environment")
+		Expect(testEnv.Stop()).To(Succeed())
+	})
+
+	By("Create test clients")
+	testClient, err = client.New(restConfig, client.Options{Scheme: utilclient.SharderScheme})
+	Expect(err).NotTo(HaveOccurred())
+	Expect(testClient).NotTo(BeNil())
+
+	clientContext, clientCancel := context.WithCancel(context.Background())
+	komega.SetClient(testClient)
+	komega.SetContext(clientContext)
+	DeferCleanup(clientCancel)
+
+	By("Create test Namespace")
+	testNamespace := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: testID + "-",
+		},
+	}
+	Expect(testClient.Create(ctx, testNamespace)).To(Succeed())
+	log.Info("Created Namespace for test", "namespaceName", testNamespace.Name)
+	testRunID = testNamespace.Name
+	log = log.WithValues("testRunID", testRunID)
+	testRunLabels = map[string]string{testID: testRunID}
+
+	DeferCleanup(func(ctx SpecContext) {
+		By("Delete test Namespace")
+		Expect(testClient.Delete(ctx, testNamespace)).To(Or(Succeed(), BeNotFoundError()))
+	}, NodeTimeout(time.Minute))
+
+	By("Setup manager")
+	mgr, err := manager.New(restConfig, manager.Options{
+		Scheme:  utilclient.SharderScheme,
+		Metrics: metricsserver.Options{BindAddress: "0"},
+		Cache: cache.Options{
+			DefaultNamespaces: map[string]cache.Config{testNamespace.Name: {}},
+			ByObject: map[client.Object]cache.ByObject{
+				&shardingv1alpha1.ControllerRing{}: {
+					Label: labels.SelectorFromSet(testRunLabels),
+				},
+			},
+		},
+	})
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Register controller")
+	config := &configv1alpha1.SharderConfig{}
+	mgr.GetScheme().Default(config)
+
+	clock = testclock.NewFakeClock(time.Now())
+
+	Expect((&shardlease.Reconciler{
+		Clock: clock,
+	}).AddToManager(mgr)).To(Succeed())
+
+	By("Start manager")
+	mgrContext, mgrCancel := context.WithCancel(context.Background())
+
+	go func() {
+		defer GinkgoRecover()
+		Expect(mgr.Start(mgrContext)).To(Succeed())
+	}()
+
+	DeferCleanup(func() {
+		By("Stop manager")
+		mgrCancel()
+	})
+}, NodeTimeout(time.Minute))