diff --git a/main.go b/main.go index 5d9043b14..3277bad42 100644 --- a/main.go +++ b/main.go @@ -110,6 +110,11 @@ func initialize(services *service.Services) { var jobServiceAccount string // get prometheus endpoint from environment var prometheusServiceEndpoint string + // HA mode for Active/Standby cross-cluster deployments. See docs/design + // for the full HA proposal (LFX #305). Default "disabled" preserves + // existing single-cluster behaviour; "active" and "standby" install a + // manager-backed leader gate so only the elected pod can mutate state. + var haMode string flag.StringVar(&rbacResourcePrefix, "rbac-resource-prefix", service.RbacResourcePrefix, "RBAC resource prefix") flag.StringVar(&projectNameSpacePrefixFromCustomer, "project-namespace-prefix", service.ProjectNamespacePrefix, fmt.Sprintf("Overrides the default %s kubeslice namespace", service.ProjectNamespacePrefix)) @@ -137,9 +142,33 @@ func initialize(services *service.Services) { flag.BoolVar(&enableLeaderElection, "leader-elect", false, "Enable leader election for controller manager. "+ "Enabling this will ensure there is only one active controller manager.") + flag.StringVar(&haMode, "ha-mode", "disabled", + "Active/Standby HA mode: 'disabled' (default, single-cluster behaviour), "+ + "'active' or 'standby' (cross-cluster HA — installs a leader-gate "+ + "so only the elected pod mutates state). Modes other than 'disabled' "+ + "require --leader-elect=true.") flag.Parse() + // Validate --ha-mode early; bail at startup rather than silently + // running mis-fenced. Accepted values are stable contract. + switch haMode { + case "disabled", "active", "standby": + // ok + default: + setupLog.Error(nil, "invalid --ha-mode value; must be one of disabled|active|standby", + "value", haMode) + os.Exit(1) + } + // HA mode without leader election would let both clusters mutate at + // once. Fail loud at startup; the operator can either flip --leader-elect + // on or drop back to --ha-mode=disabled. + if haMode != "disabled" && !enableLeaderElection { + setupLog.Error(nil, "--ha-mode requires --leader-elect=true", + "ha-mode", haMode, "leader-elect", enableLeaderElection) + os.Exit(1) + } + // initialize logger if logLevel == "" { logLevel = "info" @@ -269,6 +298,21 @@ func initialize(services *service.Services) { setupLog.Error(err, "unable to start manager") os.Exit(1) } + + // Install the process-wide LeaderGate. In "disabled" mode this is a + // no-op gate (preserves pre-HA behaviour); in "active"/"standby" mode + // every util.Create/Update/Delete/UpdateStatus and finalizer-add/remove + // is fenced behind mgr.Elected(), so a non-leader pod cannot mutate + // cluster state even if a goroutine escapes controller-runtime's + // reconciler gate. Must be called BEFORE any reconciler runs. + if haMode == "disabled" { + util.SetDefaultLeaderGate(util.NoOpLeaderGate{}) + setupLog.Info("HA leader-gate installed", "mode", haMode, "gate", "no-op") + } else { + util.SetDefaultLeaderGate(util.NewManagerLeaderGate(mgr)) + setupLog.Info("HA leader-gate installed", "mode", haMode, "gate", "manager-elected") + } + //setting up the event recorder eventRecorder := events.NewEventRecorder(mgr.GetClient(), mgr.GetScheme(), ossEvents.EventsMap, events.EventRecorderOptions{ Version: "v1alpha1", diff --git a/util/cleanup_utility.go b/util/cleanup_utility.go index 603c1cd28..62962f63a 100644 --- a/util/cleanup_utility.go +++ b/util/cleanup_utility.go @@ -18,6 +18,7 @@ package util import ( "context" + "fmt" "go.uber.org/zap" "sigs.k8s.io/controller-runtime/pkg/client" @@ -28,6 +29,10 @@ func CleanupUpdateResource(ctx context.Context, object client.Object) error { logger := CtxLogger(ctx) logger.Debugf("Updating %s %s in namespace %s", GetObjectKind(object), object.GetName(), object.GetNamespace()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused CleanupUpdateResource of %s/%s: %v", object.GetNamespace(), object.GetName(), err) + return fmt.Errorf("cleanup update %s %s/%s: %w", GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) err := kubeSliceCtx.Update(ctx, object) if err != nil { @@ -44,6 +49,10 @@ func CleanupUpdateStatus(ctx context.Context, object client.Object) error { logger := CtxLogger(ctx) logger.Debugf("Updating status of %s with name %s in namespace %s", GetObjectKind(object), object.GetName(), object.GetNamespace()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused CleanupUpdateStatus of %s/%s: %v", object.GetNamespace(), object.GetName(), err) + return fmt.Errorf("cleanup update status %s %s/%s: %w", GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) err := kubeSliceCtx.Status().Update(ctx, object) if err != nil { @@ -67,6 +76,10 @@ func CleanupDeleteResource(ctx context.Context, object client.Object) error { logger := CtxLogger(ctx) logger.Debugf("Deleting %s %s in namespace %s", GetObjectKind(object), object.GetName(), object.GetNamespace()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused CleanupDeleteResource of %s/%s: %v", object.GetNamespace(), object.GetName(), err) + return fmt.Errorf("cleanup delete %s %s/%s: %w", GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) err := kubeSliceCtx.Delete(ctx, object) if err != nil { diff --git a/util/kube_slice_hub_context.go b/util/kube_slice_hub_context.go index 510394f54..ce7884902 100644 --- a/util/kube_slice_hub_context.go +++ b/util/kube_slice_hub_context.go @@ -32,12 +32,21 @@ var LoglevelString string type kubeSliceControllerContextKey struct { } -// kubeSliceControllerRequestContext is a schema for request context +// kubeSliceControllerRequestContext is a schema for request context. +// +// leaderGate is the per-request HA fence (see util/leader_gate.go). It is +// populated in PrepareKubeSliceControllersRequestContext from the +// process-wide DefaultLeaderGate so existing call sites need no changes; +// tests may swap it for a mock by inserting their own request context. +// A nil value is treated as "fall back to package default" by +// util.requireLeader, so any older caller that constructs this struct +// directly still gets correct behaviour. type kubeSliceControllerRequestContext struct { client.Client Scheme *runtime.Scheme Log *zap.SugaredLogger eventRecorder *events.EventRecorder + leaderGate LeaderGate } // kubeSliceControllerContext is instance of kubeSliceControllerContextKey @@ -64,6 +73,7 @@ func PrepareKubeSliceControllersRequestContext(ctx context.Context, client clien Scheme: scheme, Log: log, eventRecorder: er, + leaderGate: DefaultLeaderGate(), } newCtx := context.WithValue(ctx, kubeSliceControllerContext, ctxVal) return newCtx @@ -98,3 +108,19 @@ func CtxEventRecorder(ctx context.Context) events.EventRecorder { recorder := GetKubeSliceControllerRequestContext(ctx).eventRecorder return *recorder } + +// CtxLeaderGate returns the per-request LeaderGate, falling back to the +// process-wide DefaultLeaderGate when the request context does not +// carry one (e.g., contexts produced before HA wiring landed, or by +// callers that bypass PrepareKubeSliceControllersRequestContext). +// +// Reconciler code paths rarely need this directly; util's mutation +// helpers consult it internally via requireLeader. +func CtxLeaderGate(ctx context.Context) LeaderGate { + if ctx != nil { + if rc := GetKubeSliceControllerRequestContext(ctx); rc != nil && rc.leaderGate != nil { + return rc.leaderGate + } + } + return DefaultLeaderGate() +} diff --git a/util/leader_gate.go b/util/leader_gate.go new file mode 100644 index 000000000..8784e57c2 --- /dev/null +++ b/util/leader_gate.go @@ -0,0 +1,173 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. # # SPDX-License-Identifier: Apache-2.0 + * + * 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 util — leader_gate.go provides the runtime fence that backs +// kubeslice-controller's Active/Standby HA contract. +// +// Every mutating Kubernetes operation in this controller funnels through +// CreateResource / UpdateResource / UpdateStatus / DeleteResource (and their +// cleanup-binary counterparts). Those helpers consult a LeaderGate before +// issuing the API call so that, when HA mode is enabled, a Standby instance +// (or any code path that escapes controller-runtime's leader-election gate) +// cannot accidentally mutate cluster state. +// +// The default gate is a no-op so existing single-cluster deployments behave +// identically to the pre-HA codebase. main.go installs a manager-backed +// gate at startup when --ha-mode is set to "active" or "standby"; in those +// modes, only the controller-runtime-elected leader is permitted to write. +// +// Design notes (see LFX #305 plan, PoC PR #2): +// - Default behaviour preserved: zero call-site changes; nil gate or +// missing context falls back to NoOpLeaderGate. +// - Defence in depth: this gate is the SECOND fence. The FIRST is +// controller-runtime's manager-level leader election, which suppresses +// reconciler dispatch entirely on non-leaders. The gate catches any +// mutation path that escapes the manager — goroutines started outside +// mgr.Add, webhook handlers, the standalone cleanup binary, future code +// that forgets to wire through the manager. +// - Per-request override: tests and future advanced callers may stash a +// gate on the per-request context via PrepareKubeSliceControllersRequestContext +// to inject mocks without touching package state. +package util + +import ( + "context" + "errors" + "sync/atomic" + + ctrl "sigs.k8s.io/controller-runtime" +) + +// ErrNotLeader is returned by LeaderGate.RequireLeader when this instance +// is not the elected leader and is therefore forbidden from mutating +// cluster state. Callers should detect it with errors.Is and treat it as +// a transient condition: requeue rather than escalating to an event/alert. +var ErrNotLeader = errors.New("kubeslice-controller: instance is not the active leader; mutation refused by leader-gate") + +// LeaderGate decides whether the calling instance is permitted to perform +// mutating Kubernetes API operations. +// +// Implementations MUST be safe for concurrent use and MUST keep RequireLeader +// cheap (sub-microsecond): the gate is consulted on every API write and on +// every finalizer add/remove. +type LeaderGate interface { + // RequireLeader returns nil if the caller is permitted to mutate, or a + // non-nil error wrapping ErrNotLeader otherwise. + RequireLeader() error +} + +// NoOpLeaderGate always permits mutations. It is the default gate for +// non-HA deployments and for code paths whose request context does not +// carry a gate (e.g., the standalone cleanup binary, unit tests that +// don't exercise gating semantics). +type NoOpLeaderGate struct{} + +// RequireLeader always returns nil. +func (NoOpLeaderGate) RequireLeader() error { return nil } + +// managerElectedGate gates on controller-runtime's leader-election signal. +// mgr.Elected() returns a channel that is closed once this manager has +// won (or will never win, e.g., when leader election is disabled — in +// that case the channel is closed at startup, so the gate permits all +// mutations). See sigs.k8s.io/controller-runtime/pkg/manager.Manager. +type managerElectedGate struct { + elected <-chan struct{} +} + +// NewManagerLeaderGate returns a LeaderGate that permits mutations only +// after the supplied manager has won leader election. It is safe to wire +// unconditionally: a manager with LeaderElection disabled closes Elected() +// at startup, so the gate behaves like NoOpLeaderGate in that mode. +// +// Callers should pass the SAME manager that owns the reconcilers; mixing +// managers (e.g., one for webhooks, one for reconcilers) defeats the gate. +func NewManagerLeaderGate(mgr ctrl.Manager) LeaderGate { + if mgr == nil { + // Defensive: a nil manager would panic on Elected(). Treat as + // permissive so tests / misconfigured setups fail loudly elsewhere + // rather than at every mutation. + return NoOpLeaderGate{} + } + return &managerElectedGate{elected: mgr.Elected()} +} + +// RequireLeader returns nil if this manager has been elected, or +// ErrNotLeader otherwise. The non-blocking select keeps the check at +// nanosecond-cost on the hot path: once Elected() is closed, the receive +// case is always immediately ready. +func (g *managerElectedGate) RequireLeader() error { + select { + case <-g.elected: + return nil + default: + return ErrNotLeader + } +} + +// defaultLeaderGate holds the process-wide gate. Stored as an +// atomic.Pointer so SetDefaultLeaderGate and DefaultLeaderGate can be +// called concurrently without locks. +// +// Initialised in init() to a NoOpLeaderGate so any package init or test +// that touches mutation paths before main.go has wired the gate sees +// identical behaviour to the pre-HA codebase. +var defaultLeaderGate atomic.Pointer[LeaderGate] + +func init() { + var initial LeaderGate = NoOpLeaderGate{} + defaultLeaderGate.Store(&initial) +} + +// SetDefaultLeaderGate replaces the process-wide gate. Passing nil resets +// to NoOpLeaderGate. main.go calls this once at startup based on the +// --ha-mode flag; tests may call it to inject mock gates. +// +// The call is safe for concurrent use, but callers should treat the gate +// as effectively immutable after startup: there is no synchronisation +// between an in-flight mutation observing the old gate and a concurrent +// re-set. In practice this is a non-issue because main.go wires the gate +// before any reconciler runs. +func SetDefaultLeaderGate(g LeaderGate) { + if g == nil { + g = NoOpLeaderGate{} + } + defaultLeaderGate.Store(&g) +} + +// DefaultLeaderGate returns the current process-wide gate. Never nil. +func DefaultLeaderGate() LeaderGate { + return *defaultLeaderGate.Load() +} + +// requireLeader is the internal hot-path helper consulted by every +// mutating util function. It prefers a gate embedded in the request +// context (via PrepareKubeSliceControllersRequestContext) over the +// process-wide default, letting tests inject context-scoped gates +// without touching package state. +// +// Resolution order: +// 1. ctx != nil AND ctx carries a kubeslice request context AND that +// context's leaderGate field is non-nil → use it. +// 2. Otherwise fall back to DefaultLeaderGate(), which is itself +// NoOpLeaderGate by default. +func requireLeader(ctx context.Context) error { + if ctx != nil { + if rc := GetKubeSliceControllerRequestContext(ctx); rc != nil && rc.leaderGate != nil { + return rc.leaderGate.RequireLeader() + } + } + return DefaultLeaderGate().RequireLeader() +} diff --git a/util/leader_gate_test.go b/util/leader_gate_test.go new file mode 100644 index 000000000..ddc4a9b7e --- /dev/null +++ b/util/leader_gate_test.go @@ -0,0 +1,243 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. # # SPDX-License-Identifier: Apache-2.0 + * + * 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 util + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +// newGateForTest constructs a managerElectedGate from a channel we control. +// The package-private constructor lets tests exercise the same code path +// the production NewManagerLeaderGate uses, without spinning up a real +// controller-runtime manager (which would drag in API server scaffolding). +func newGateForTest(elected <-chan struct{}) LeaderGate { + return &managerElectedGate{elected: elected} +} + +func TestNoOpLeaderGate_AlwaysPermits(t *testing.T) { + gate := NoOpLeaderGate{} + require.NoError(t, gate.RequireLeader()) + // Idempotency: many calls, same result. + for i := 0; i < 1000; i++ { + require.NoError(t, gate.RequireLeader()) + } +} + +func TestManagerElectedGate_BlocksUntilElected(t *testing.T) { + elected := make(chan struct{}) + gate := newGateForTest(elected) + + err := gate.RequireLeader() + require.Error(t, err, "should block before election") + require.ErrorIs(t, err, ErrNotLeader, "must wrap ErrNotLeader so callers can detect it with errors.Is") + + close(elected) + require.NoError(t, gate.RequireLeader(), "must permit after election") + + // Once permitted, stays permitted (the channel cannot re-open). + for i := 0; i < 100; i++ { + require.NoError(t, gate.RequireLeader()) + } +} + +func TestManagerElectedGate_AlreadyClosed(t *testing.T) { + // Real controller-runtime behaviour: when LeaderElection is disabled, + // mgr.Elected() returns a channel that is already closed. The gate + // must permit immediately in that case so it's safe to wire + // unconditionally. + elected := make(chan struct{}) + close(elected) + gate := newGateForTest(elected) + require.NoError(t, gate.RequireLeader()) +} + +func TestNewManagerLeaderGate_NilManagerIsDefensive(t *testing.T) { + // A nil manager would panic on Elected(); the constructor should + // fall back to NoOpLeaderGate so a misconfigured wiring fails + // loudly elsewhere instead of at every mutation. + gate := NewManagerLeaderGate(nil) + require.IsType(t, NoOpLeaderGate{}, gate) + require.NoError(t, gate.RequireLeader()) +} + +func TestSetDefaultLeaderGate_RoundTrip(t *testing.T) { + // Preserve original default so this test doesn't leak state into + // other tests that read DefaultLeaderGate(). + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + elected := make(chan struct{}) + gate := newGateForTest(elected) + SetDefaultLeaderGate(gate) + require.Same(t, gate, DefaultLeaderGate()) + + require.ErrorIs(t, DefaultLeaderGate().RequireLeader(), ErrNotLeader) + close(elected) + require.NoError(t, DefaultLeaderGate().RequireLeader()) +} + +func TestSetDefaultLeaderGate_NilResetsToNoOp(t *testing.T) { + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + SetDefaultLeaderGate(nil) + require.IsType(t, NoOpLeaderGate{}, DefaultLeaderGate(), + "passing nil must reset the package default to NoOpLeaderGate, not store a typed nil") + require.NoError(t, DefaultLeaderGate().RequireLeader()) +} + +func TestPackageDefault_StartsAsNoOp(t *testing.T) { + // Note: ordering with other tests matters. We do not assume nobody + // has called SetDefaultLeaderGate before us; we assert only that the + // default — whatever it currently is — permits mutations by default, + // which is the actual contract callers depend on. + require.NoError(t, DefaultLeaderGate().RequireLeader(), + "the package default must permit mutations so pre-HA call sites behave identically") +} + +func TestSetDefaultLeaderGate_ConcurrentSafe(t *testing.T) { + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + // Race: writers set the gate, readers consult it. With the + // atomic.Pointer backing, neither side blocks the other. + var wg sync.WaitGroup + stop := make(chan struct{}) + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + SetDefaultLeaderGate(NoOpLeaderGate{}) + } + } + }() + } + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 10000; j++ { + _ = DefaultLeaderGate().RequireLeader() + } + }() + } + close(stop) + wg.Wait() +} + +func TestRequireLeader_PrefersContextOverDefault(t *testing.T) { + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + // Default permits. + SetDefaultLeaderGate(NoOpLeaderGate{}) + + // Per-request context carries a gate that does NOT permit. + rc := &kubeSliceControllerRequestContext{ + leaderGate: newGateForTest(make(chan struct{})), // open channel = not leader + } + ctx := context.WithValue(context.Background(), kubeSliceControllerContext, rc) + + err := requireLeader(ctx) + require.Error(t, err, "context gate must override permissive default") + require.ErrorIs(t, err, ErrNotLeader) +} + +func TestRequireLeader_FallsBackToDefaultWhenContextHasNoGate(t *testing.T) { + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + // Default refuses. + SetDefaultLeaderGate(newGateForTest(make(chan struct{}))) + + // Context has request-context but no gate (mimics legacy callers + // who construct the struct directly without calling Prepare-). + rc := &kubeSliceControllerRequestContext{leaderGate: nil} + ctx := context.WithValue(context.Background(), kubeSliceControllerContext, rc) + + err := requireLeader(ctx) + require.Error(t, err, "nil context gate must fall through to package default") + require.ErrorIs(t, err, ErrNotLeader) +} + +func TestRequireLeader_NilContext(t *testing.T) { + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + // Default permits. + SetDefaultLeaderGate(NoOpLeaderGate{}) + //nolint:staticcheck // intentionally passing nil to assert defensive behaviour + require.NoError(t, requireLeader(nil)) + + // Default refuses. + SetDefaultLeaderGate(newGateForTest(make(chan struct{}))) + //nolint:staticcheck // intentionally passing nil to assert defensive behaviour + err := requireLeader(nil) + require.Error(t, err) + require.ErrorIs(t, err, ErrNotLeader) +} + +func TestRequireLeader_NoRequestContext(t *testing.T) { + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + // A bare context.Background() has no request context at all; must + // not panic and must consult the package default. + SetDefaultLeaderGate(NoOpLeaderGate{}) + require.NoError(t, requireLeader(context.Background())) + + SetDefaultLeaderGate(newGateForTest(make(chan struct{}))) + require.ErrorIs(t, requireLeader(context.Background()), ErrNotLeader) +} + +func TestErrNotLeader_IsSentinel(t *testing.T) { + // Callers (reconcilers) detect this via errors.Is to decide whether + // to requeue vs alert. Verify the sentinel survives the typical + // fmt.Errorf wrap chain used by the mutation helpers. + wrapped := errors.Join(ErrNotLeader, errors.New("higher-level context")) + require.ErrorIs(t, wrapped, ErrNotLeader) +} + +func TestCtxLeaderGate_ReadsContextThenFallsBack(t *testing.T) { + original := DefaultLeaderGate() + t.Cleanup(func() { SetDefaultLeaderGate(original) }) + + // Empty context → package default. + SetDefaultLeaderGate(NoOpLeaderGate{}) + gate := CtxLeaderGate(context.Background()) + require.IsType(t, NoOpLeaderGate{}, gate) + + // Context with explicit gate wins. + customGate := NoOpLeaderGate{} + rc := &kubeSliceControllerRequestContext{leaderGate: customGate} + ctx := context.WithValue(context.Background(), kubeSliceControllerContext, rc) + require.Equal(t, customGate, CtxLeaderGate(ctx)) + + // Nil context defended. + //nolint:staticcheck // intentionally passing nil to assert defensive behaviour + require.IsType(t, NoOpLeaderGate{}, CtxLeaderGate(nil)) +} diff --git a/util/reconciliation_utility.go b/util/reconciliation_utility.go index d399ceae5..62b1d9b06 100644 --- a/util/reconciliation_utility.go +++ b/util/reconciliation_utility.go @@ -83,6 +83,10 @@ func CreateResource(ctx context.Context, object client.Object) error { logger := CtxLogger(ctx) logger.Debugf("Creating object kind %s with name %s in namespace %s", GetObjectKind(object), object.GetName(), object.GetNamespace()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused Create of %s/%s: %v", object.GetNamespace(), object.GetName(), err) + return fmt.Errorf("create %s %s/%s: %w", GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) err := kubeSliceCtx.Create(ctx, object) if err != nil { @@ -99,6 +103,10 @@ func UpdateResource(ctx context.Context, object client.Object) error { logger := CtxLogger(ctx) logger.Debugf("Updating object kind %s with name %s in namespace %s", GetObjectKind(object), object.GetName(), object.GetNamespace()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused Update of %s/%s: %v", object.GetNamespace(), object.GetName(), err) + return fmt.Errorf("update %s %s/%s: %w", GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) err := kubeSliceCtx.Update(ctx, object) if err != nil { @@ -115,6 +123,10 @@ func UpdateStatus(ctx context.Context, object client.Object) error { logger := CtxLogger(ctx) logger.Debugf("Updating object status %s with name %s in namespace %s", GetObjectKind(object), object.GetName(), object.GetNamespace()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused UpdateStatus of %s/%s: %v", object.GetNamespace(), object.GetName(), err) + return fmt.Errorf("update status %s %s/%s: %w", GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) err := kubeSliceCtx.Status().Update(ctx, object) if err != nil { @@ -138,6 +150,10 @@ func DeleteResource(ctx context.Context, object client.Object) error { logger := CtxLogger(ctx) logger.Debugf("Deleting object kind %s with name %s in namespace %s", GetObjectKind(object), object.GetName(), object.GetNamespace()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused Delete of %s/%s: %v", object.GetNamespace(), object.GetName(), err) + return fmt.Errorf("delete %s %s/%s: %w", GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) err := kubeSliceCtx.Delete(ctx, object) if err != nil { @@ -164,6 +180,10 @@ func AddFinalizer(ctx context.Context, object client.Object, finalizerName strin kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) logger := CtxLogger(ctx) logger.Debugf("Adding finalizer %s to %s", finalizerName, object.GetName()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused AddFinalizer of %s on %s/%s: %v", finalizerName, object.GetNamespace(), object.GetName(), err) + return ctrl.Result{}, fmt.Errorf("add finalizer %s on %s %s/%s: %w", finalizerName, GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } controllerutil.AddFinalizer(object, finalizerName) if err := kubeSliceCtx.Update(ctx, object); err != nil { logger.With(zap.Error(err)).Errorf("Failed to add finalizer") @@ -185,6 +205,10 @@ func RemoveFinalizer(ctx context.Context, object client.Object, finalizerName st kubeSliceCtx := GetKubeSliceControllerRequestContext(ctx) logger := CtxLogger(ctx) logger.Debugf("Removing finalizer %s from %s", finalizerName, object.GetName()) + if err := requireLeader(ctx); err != nil { + logger.Debugf("leader-gate refused RemoveFinalizer of %s on %s/%s: %v", finalizerName, object.GetNamespace(), object.GetName(), err) + return ctrl.Result{}, fmt.Errorf("remove finalizer %s on %s %s/%s: %w", finalizerName, GetObjectKind(object), object.GetNamespace(), object.GetName(), err) + } controllerutil.RemoveFinalizer(object, finalizerName) if err := kubeSliceCtx.Update(ctx, object); err != nil { logger.With(zap.Error(err)).Errorf("Failed to remove finalizer %s", finalizerName)