Add integration test for shard lease

test/integration/shard/controller/controller_suite_test.go

@@ -35,7 +35,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
-	"github.com/timebertt/kubernetes-controller-sharding/pkg/utils/test"
 	. "github.com/timebertt/kubernetes-controller-sharding/pkg/utils/test/matchers"
 	. "github.com/timebertt/kubernetes-controller-sharding/test/integration/shard/controller"
 )
@@ -63,12 +62,7 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 	log = logf.Log.WithName(testID)
 
 	By("Start test environment")
-	testEnv := &envtest.Environment{
-		CRDInstallOptions: envtest.CRDInstallOptions{
-			Paths: []string{test.PathShardingCRDs()},
-		},
-		ErrorIfCRDPathMissing: true,
-	}
+	testEnv := &envtest.Environment{}
 
 	restConfig, err := testEnv.Start()
 	Expect(err).NotTo(HaveOccurred())

test/integration/shard/lease/lease_suite_test.go

@@ -0,0 +1,98 @@
+/*
+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 controller_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/client-go/rest"
+	"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"
+
+	. "github.com/timebertt/kubernetes-controller-sharding/pkg/utils/test/matchers"
+)
+
+func TestController(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Shard Lease Integration Test Suite")
+}
+
+const testID = "shard-lease-test"
+
+var (
+	log logr.Logger
+
+	restConfig *rest.Config
+
+	testClient client.Client
+
+	testRunID 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{}
+
+	var err error
+	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{})
+	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)
+
+	DeferCleanup(func(ctx SpecContext) {
+		By("Delete test Namespace")
+		Expect(testClient.Delete(ctx, testNamespace)).To(Or(Succeed(), BeNotFoundError()))
+	}, NodeTimeout(time.Minute))
+}, NodeTimeout(time.Minute))

test/integration/shard/lease/lease_test.go

@@ -0,0 +1,174 @@
+/*
+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 controller_test
+
+import (
+	"context"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	coordinationv1 "k8s.io/api/coordination/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+	. "sigs.k8s.io/controller-runtime/pkg/envtest/komega"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	shardlease "github.com/timebertt/kubernetes-controller-sharding/pkg/shard/lease"
+	"github.com/timebertt/kubernetes-controller-sharding/pkg/utils/test"
+)
+
+var _ = Describe("Shard lease", func() {
+	var (
+		mgrOptions manager.Options
+		mgrCancel  context.CancelFunc
+
+		leaseOptions shardlease.Options
+		lease        *coordinationv1.Lease
+	)
+
+	BeforeEach(func() {
+		mgrOptions = manager.Options{
+			LeaderElection:             true,
+			LeaderElectionResourceLock: "should-be-ignored", // -> should use provided lock instead
+			LeaderElectionNamespace:    "should-be-ignored", // -> should use provided namespace in lock instead
+			LeaderElectionID:           "should-be-ignored", // -> should be shard name instead
+
+			LeaseDuration: ptr.To(time.Second),
+			RenewDeadline: ptr.To(100 * time.Millisecond),
+			RetryPeriod:   ptr.To(50 * time.Millisecond),
+		}
+
+		leaseOptions = shardlease.Options{
+			ControllerRingName: "test-" + test.RandomSuffix(),
+			LeaseNamespace:     testRunID,
+			ShardName:          "test-" + test.RandomSuffix(),
+		}
+
+		lease = &coordinationv1.Lease{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: leaseOptions.LeaseNamespace,
+				Name:      leaseOptions.ShardName,
+			},
+		}
+	}, OncePerOrdered)
+
+	JustBeforeEach(func() {
+		shardLease, err := shardlease.NewResourceLock(restConfig, nil, leaseOptions)
+		Expect(err).NotTo(HaveOccurred())
+		mgrOptions.LeaderElectionResourceLockInterface = shardLease
+
+		By("Setup manager")
+		mgrOptions.Metrics.BindAddress = "0"
+		mgr, err := manager.New(restConfig, mgrOptions)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Start manager")
+		var mgrContext context.Context
+		mgrContext, mgrCancel = context.WithCancel(context.Background())
+
+		mgrDone := make(chan struct{})
+		go func() {
+			defer GinkgoRecover()
+			Expect(mgr.Start(mgrContext)).To(Succeed())
+			close(mgrDone)
+		}()
+
+		DeferCleanup(func(ctx SpecContext) {
+			By("Stop manager")
+			mgrCancel()
+
+			By("Wait for manager to stop")
+			Eventually(ctx, mgrDone).Should(BeClosed())
+		}, NodeTimeout(time.Minute))
+	}, OncePerOrdered)
+
+	Describe("should maintain the shard lease", Ordered, func() {
+		It("should create the shard lease according to the config", func(ctx SpecContext) {
+			Eventually(ctx, Object(lease)).Should(And(
+				HaveField("ObjectMeta.Labels", HaveKeyWithValue("alpha.sharding.timebertt.dev/controllerring", leaseOptions.ControllerRingName)),
+				HaveField("Spec.HolderIdentity", HaveValue(Equal(leaseOptions.ShardName))),
+				HaveField("Spec.LeaseDurationSeconds", HaveValue(BeEquivalentTo(1))),
+				HaveField("Spec.AcquireTime", Not(BeNil())),
+				HaveField("Spec.RenewTime", Not(BeNil())),
+			))
+		}, SpecTimeout(time.Minute))
+
+		It("should renew the shard lease", func(ctx SpecContext) {
+			Eventually(ctx, Object(lease)).Should(And(
+				HaveField("Spec.HolderIdentity", HaveValue(Equal(leaseOptions.ShardName))),
+				HaveField("Spec.LeaseDurationSeconds", HaveValue(BeEquivalentTo(1))),
+				HaveField("Spec.AcquireTime.Time", BeTemporally("~", lease.Spec.AcquireTime.Time)),
+				HaveField("Spec.RenewTime.Time", BeTemporally(">", lease.Spec.RenewTime.Time)),
+			))
+		}, SpecTimeout(time.Minute))
+	})
+
+	Describe("should renew shard lease that was acquired by sharder", Ordered, func() {
+		var sharderAcquireTime time.Time
+
+		BeforeAll(func(ctx SpecContext) {
+			sharderAcquireTime = time.Now()
+
+			lease.Spec.HolderIdentity = ptr.To("sharder")
+			lease.Spec.AcquireTime = ptr.To(metav1.NewMicroTime(sharderAcquireTime))
+			lease.Spec.RenewTime = ptr.To(metav1.NewMicroTime(sharderAcquireTime))
+			lease.Spec.LeaseDurationSeconds = ptr.To[int32](3)
+			lease.Spec.LeaseTransitions = ptr.To[int32](1)
+
+			Eventually(ctx, func() error {
+				return testClient.Create(ctx, lease)
+			}).Should(Succeed())
+		}, NodeTimeout(time.Minute))
+
+		It("should wait for lease to expire", func(ctx SpecContext) {
+			Consistently(ctx, Object(lease)).WithTimeout(2 * time.Second).ShouldNot(
+				HaveField("Spec.HolderIdentity", HaveValue(Equal(leaseOptions.ShardName))),
+			)
+		}, SpecTimeout(time.Minute))
+
+		It("should renew the shard lease", func(ctx SpecContext) {
+			Eventually(ctx, Object(lease)).Should(And(
+				HaveField("Spec.HolderIdentity", HaveValue(Equal(leaseOptions.ShardName))),
+				HaveField("Spec.LeaseDurationSeconds", HaveValue(BeEquivalentTo(1))),
+				HaveField("Spec.AcquireTime.Time", BeTemporally(">", sharderAcquireTime)),
+				HaveField("Spec.RenewTime.Time", BeTemporally(">", sharderAcquireTime)),
+				HaveField("Spec.LeaseTransitions", HaveValue(BeEquivalentTo(2))),
+			))
+		}, SpecTimeout(time.Minute))
+	})
+
+	When("LeaderElectionReleaseOnCancel is true", Ordered, func() {
+		BeforeAll(func() {
+			mgrOptions.LeaderElectionReleaseOnCancel = true
+		})
+
+		It("should create the shard lease", func(ctx SpecContext) {
+			Eventually(ctx, Object(lease)).Should(
+				HaveField("Spec.HolderIdentity", HaveValue(Equal(leaseOptions.ShardName))),
+			)
+		}, SpecTimeout(time.Minute))
+
+		It("should release the shard lease when canceling the manager", func(ctx SpecContext) {
+			mgrCancel()
+
+			Eventually(ctx, Object(lease)).Should(
+				HaveField("Spec.HolderIdentity", HaveValue(BeEmpty())),
+			)
+		}, SpecTimeout(time.Minute))
+	})
+})