Add migration package for gating runnables behind pre-start migrations

migration/migration.go

@@ -0,0 +1,140 @@
+// SPDX-FileCopyrightText: 2026 SAP SE or an SAP affiliate company and IronCore contributors
+// SPDX-License-Identifier: Apache-2.0
+
+// Package migration provides a Migrator that runs a set of migrations concurrently
+// before allowing controller-runtime runnables to start. Use WrapManager to gate
+// runnables behind migration completion: each runnable added via the wrapped manager
+// will block until all migrations succeed, and will return an error without starting
+// if any migration fails.
+package migration
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+// Migration is a single migration that can be run before controllers start.
+type Migration interface {
+	Migrate(ctx context.Context) error
+}
+
+// Migrator manages a set of migrations and runs them concurrently when Start is called.
+// The Done channel is closed after all migrations complete (whether or not they succeed).
+type Migrator struct {
+	mu      sync.Mutex
+	started bool
+
+	done       chan struct{}
+	migrateErr error
+
+	migrations []Migration
+}
+
+// NewMigrator creates a new Migrator with no registered migrations.
+func NewMigrator() *Migrator {
+	return &Migrator{
+		done: make(chan struct{}),
+	}
+}
+
+// Add registers a migration. It returns an error if the Migrator has already been started.
+func (m *Migrator) Add(migration Migration) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.started {
+		return fmt.Errorf("cannot add migrations once started")
+	}
+	m.migrations = append(m.migrations, migration)
+	return nil
+}
+
+// Start runs all registered migrations concurrently. It blocks until every migration
+// has completed and returns a joined error if any migration failed. Start may only be
+// called once; subsequent calls return an error immediately.
+func (m *Migrator) Start(ctx context.Context) error {
+	m.mu.Lock()
+
+	if m.started {
+		m.mu.Unlock()
+		return fmt.Errorf("migrator already started")
+	}
+
+	m.started = true
+	m.mu.Unlock()
+
+	defer close(m.done)
+
+	var (
+		wg      sync.WaitGroup
+		errChan = make(chan error)
+	)
+	for _, migration := range m.migrations {
+		wg.Go(func() {
+			if err := migration.Migrate(ctx); err != nil {
+				errChan <- err
+			}
+		})
+	}
+	go func() {
+		defer close(errChan)
+		wg.Wait()
+	}()
+
+	var errs []error
+	for err := range errChan {
+		errs = append(errs, err)
+	}
+
+	m.migrateErr = errors.Join(errs...)
+	return m.migrateErr
+}
+
+// Done returns a channel that is closed once Start has finished running all migrations.
+func (m *Migrator) Done() <-chan struct{} {
+	return m.done
+}
+
+// Err returns the error produced by Start or nil if Done is not yet closed.
+func (m *Migrator) Err() error {
+	select {
+	case <-m.Done():
+		return m.migrateErr
+	default:
+		return nil
+	}
+}
+
+type migrationAwareManager struct {
+	migrator *Migrator
+	manager.Manager
+}
+
+// WrapManager returns a manager.Manager that gates all runnables added via Add behind
+// the given Migrator's completion. If the migration succeeds, the runnables start normally.
+// If the migration fails, each runnable returns the migration error without starting.
+// If the context is canceled before migration completes, each runnable exits with nil.
+func WrapManager(migrator *Migrator, mgr manager.Manager) manager.Manager {
+	return &migrationAwareManager{
+		migrator: migrator,
+		Manager:  mgr,
+	}
+}
+
+func (m *migrationAwareManager) Add(fn manager.Runnable) error {
+	return m.Manager.Add(manager.RunnableFunc(func(ctx context.Context) error {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-m.migrator.Done():
+			if err := m.migrator.Err(); err != nil {
+				return fmt.Errorf("migration error, won't start (%w)", err)
+			}
+			return fn.Start(ctx)
+		}
+	}))
+}

migration/migration_test.go

@@ -0,0 +1,245 @@
+// SPDX-FileCopyrightText: 2026 SAP SE or an SAP affiliate company and IronCore contributors
+// SPDX-License-Identifier: Apache-2.0
+
+package migration
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+func TestMigration(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Migration Suite")
+}
+
+type fakeMigration struct {
+	err     error
+	called  atomic.Bool
+	started chan struct{}
+	block   chan struct{}
+}
+
+func newFakeMigration(err error) *fakeMigration {
+	return &fakeMigration{
+		err:     err,
+		started: make(chan struct{}),
+		block:   make(chan struct{}),
+	}
+}
+
+func newUnblockedMigration(err error) *fakeMigration {
+	fm := newFakeMigration(err)
+	fm.unblock()
+	return fm
+}
+
+func (f *fakeMigration) Migrate(_ context.Context) error {
+	f.called.Store(true)
+	close(f.started)
+	<-f.block
+	return f.err
+}
+
+func (f *fakeMigration) unblock() {
+	close(f.block)
+}
+
+// captureManager is a minimal fake that satisfies the manager.Manager interface.
+// Only the Add method is overridden to capture the runnable; all other methods
+// delegate to the embedded nil interface and will panic if called.
+type captureManager struct {
+	captured manager.Runnable
+	manager.Manager
+}
+
+func (c *captureManager) Add(r manager.Runnable) error {
+	c.captured = r
+	return nil
+}
+
+var _ = Describe("Migrator", func() {
+	It("should create a new migrator with open Done channel and nil Err", func() {
+		m := NewMigrator()
+		Expect(m).NotTo(BeNil())
+		Expect(m.Done()).NotTo(BeClosed())
+		Expect(m.Err()).NotTo(HaveOccurred())
+	})
+
+	It("should allow adding migrations before start", func() {
+		m := NewMigrator()
+		Expect(m.Add(newUnblockedMigration(nil))).To(Succeed())
+	})
+
+	It("should reject adding migrations after start", func(ctx SpecContext) {
+		m := NewMigrator()
+		Expect(m.Start(ctx)).To(Succeed())
+
+		Expect(m.Add(newUnblockedMigration(nil))).To(MatchError(ContainSubstring("cannot add migrations once started")))
+	})
+
+	It("should run all registered migrations", func(ctx SpecContext) {
+		m := NewMigrator()
+		fm1 := newUnblockedMigration(nil)
+		fm2 := newUnblockedMigration(nil)
+
+		Expect(m.Add(fm1)).To(Succeed())
+		Expect(m.Add(fm2)).To(Succeed())
+		Expect(m.Start(ctx)).To(Succeed())
+
+		Expect(fm1.called.Load()).To(BeTrue())
+		Expect(fm2.called.Load()).To(BeTrue())
+	})
+
+	It("should succeed with no migrations", func(ctx SpecContext) {
+		m := NewMigrator()
+		Expect(m.Start(ctx)).To(Succeed())
+		Expect(m.Done()).To(BeClosed())
+	})
+
+	It("should collect errors from failing migrations", func(ctx SpecContext) {
+		m := NewMigrator()
+		Expect(m.Add(newUnblockedMigration(fmt.Errorf("migration-1 failed")))).To(Succeed())
+		Expect(m.Add(newUnblockedMigration(fmt.Errorf("migration-2 failed")))).To(Succeed())
+		Expect(m.Add(newUnblockedMigration(nil))).To(Succeed())
+
+		err := m.Start(ctx)
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("migration-1 failed"))
+		Expect(err.Error()).To(ContainSubstring("migration-2 failed"))
+		Expect(m.Err()).To(Equal(err))
+	})
+
+	It("should reject starting twice", func(ctx SpecContext) {
+		m := NewMigrator()
+		Expect(m.Start(ctx)).To(Succeed())
+		Expect(m.Start(ctx)).To(MatchError(ContainSubstring("already started")))
+	})
+
+	It("should close Done only after migrations complete", func(ctx SpecContext) {
+		m := NewMigrator()
+		fm := newFakeMigration(nil)
+		Expect(m.Add(fm)).To(Succeed())
+
+		go func() {
+			defer GinkgoRecover()
+			_ = m.Start(ctx)
+		}()
+
+		Eventually(fm.started).Should(BeClosed())
+		Expect(m.Done()).NotTo(BeClosed())
+
+		fm.unblock()
+		Eventually(m.Done()).Should(BeClosed())
+	})
+
+	It("should run migrations concurrently", func(ctx SpecContext) {
+		m := NewMigrator()
+		fm1 := newFakeMigration(nil)
+		fm2 := newFakeMigration(nil)
+		Expect(m.Add(fm1)).To(Succeed())
+		Expect(m.Add(fm2)).To(Succeed())
+
+		go func() {
+			defer GinkgoRecover()
+			_ = m.Start(ctx)
+		}()
+
+		Eventually(fm1.started).Should(BeClosed())
+		Eventually(fm2.started).Should(BeClosed())
+
+		fm1.unblock()
+		fm2.unblock()
+		Eventually(m.Done()).Should(BeClosed())
+	})
+})
+
+var _ = Describe("WrapManager", func() {
+	It("should block runnables until migration completes, then run them", func(ctx SpecContext) {
+		m := NewMigrator()
+		fm := newFakeMigration(nil)
+		Expect(m.Add(fm)).To(Succeed())
+
+		mgr := &captureManager{}
+		wrapped := WrapManager(m, mgr)
+
+		runnableStarted := make(chan struct{})
+		Expect(wrapped.Add(manager.RunnableFunc(func(ctx context.Context) error {
+			close(runnableStarted)
+			return nil
+		}))).To(Succeed())
+
+		runnableDone := make(chan error, 1)
+		go func() {
+			runnableDone <- mgr.captured.Start(ctx)
+		}()
+
+		Consistently(runnableStarted, 50*time.Millisecond).ShouldNot(BeClosed())
+
+		fm.unblock()
+		go func() {
+			defer GinkgoRecover()
+			_ = m.Start(ctx)
+		}()
+
+		Eventually(runnableStarted).Should(BeClosed())
+		Eventually(runnableDone).Should(Receive(Succeed()))
+	})
+
+	It("should prevent runnables from starting when migration fails", func(ctx SpecContext) {
+		m := NewMigrator()
+		Expect(m.Add(newUnblockedMigration(fmt.Errorf("boom")))).To(Succeed())
+		_ = m.Start(ctx)
+
+		mgr := &captureManager{}
+		wrapped := WrapManager(m, mgr)
+
+		called := false
+		Expect(wrapped.Add(manager.RunnableFunc(func(ctx context.Context) error {
+			called = true
+			return nil
+		}))).To(Succeed())
+
+		err := mgr.captured.Start(ctx)
+		Expect(err).To(MatchError(ContainSubstring("boom")))
+		Expect(called).To(BeFalse())
+	})
+
+	It("should exit cleanly when context is canceled before migration finishes", func(ctx SpecContext) {
+		m := NewMigrator()
+		fm := newFakeMigration(nil)
+		Expect(m.Add(fm)).To(Succeed())
+
+		mgr := &captureManager{}
+		wrapped := WrapManager(m, mgr)
+
+		called := false
+		Expect(wrapped.Add(manager.RunnableFunc(func(ctx context.Context) error {
+			called = true
+			return nil
+		}))).To(Succeed())
+
+		runnableCtx, cancel := context.WithCancel(ctx)
+
+		runnableDone := make(chan error, 1)
+		go func() {
+			runnableDone <- mgr.captured.Start(runnableCtx)
+		}()
+
+		cancel()
+
+		Eventually(runnableDone).Should(Receive(Succeed()))
+		Expect(called).To(BeFalse())
+
+		fm.unblock()
+		go func() { _ = m.Start(ctx) }()
+		Eventually(m.Done()).Should(BeClosed())
+	})
+})