README
¶
pkg/controller/leaderelection
Event adapter around pkg/k8s/leaderelection. The pure elector runs the Kubernetes Lease-based election; this adapter forwards every state change to the controller's EventBus so the commentator, metrics adapter, and leader-only components can react.
Minimal Usage
import (
"context"
"time"
busevents "github.com/haproxy-haptic/HAPTIC/pkg/events"
"github.com/haproxy-haptic/HAPTIC/pkg/controller/leaderelection"
k8sle "github.com/haproxy-haptic/HAPTIC/pkg/k8s/leaderelection"
)
cfg := &k8sle.Config{
Enabled: true,
Identity: podName,
LeaseName: "haptic",
LeaseNamespace: namespace,
LeaseDuration: 15 * time.Second,
RenewDeadline: 10 * time.Second,
RetryPeriod: 2 * time.Second,
ReleaseOnCancel: true,
}
callbacks := k8sle.Callbacks{
OnStartedLeading: func(ctx context.Context) { /* start leader-only work */ },
OnStoppedLeading: func() { /* stop leader-only work */ },
OnNewLeader: func(identity string) { /* optional */ },
}
comp, err := leaderelection.New(cfg, clientset, eventBus, callbacks, logger)
if err != nil {
return err
}
go comp.Start(ctx) // blocks inside until ctx is cancelled
Start(ctx) is the entry point — the README previously used Run(ctx), which doesn't exist. Leadership transitions are observed via the published events (BecameLeaderEvent, LostLeadershipEvent, NewLeaderObservedEvent); there are no synchronous state accessors.
Events Published
Published before the matching user callback runs, so listeners see the transition before side-effects kick in:
| Event | When |
|---|---|
LeaderElectionStartedEvent |
At the top of Start |
BecameLeaderEvent |
Inside the wrapper for OnStartedLeading, while the bus is paused (see below) |
LostLeadershipEvent |
Just before OnStoppedLeading |
NewLeaderObservedEvent |
Every time OnNewLeader fires (including when this replica loses) |
The publishing happens in the callback wrappers — user callbacks don't need to emit events themselves, and wouldn't be able to guarantee the "before the effect" ordering if they tried.
Pause/Start contract on becoming leader
The OnStartedLeading wrapper runs EventBus.Pause() → publish BecameLeaderEvent → user callback → EventBus.Start(). This is deliberate: the user callback is where leader-only components are constructed and call Subscribe(), and the bus must replay the buffered BecameLeaderEvent to them after they've subscribed. Without the pause, leader-only components miss the very signal meant to wake them up. See "Late Subscriber Problem" in pkg/controller/CLAUDE.md for the full rationale.
Why an Adapter At All
The pure elector in pkg/k8s/leaderelection has no dependency on the event bus — that keeps it reusable outside this controller. All bus-coupled observability (logging via commentator, metrics via pkg/controller/metrics, leader-only gating via pkg/controller/leadership) is concentrated here.
Start delegates directly to the pure elector's Start. If you find yourself adding business logic to this adapter, it probably belongs in the pure elector or a dedicated controller component instead.
See Also
pkg/k8s/leaderelection— the pure elector and itsConfig/Callbacksshapespkg/controller/leadership— gating utilities for leader-only componentspkg/controller/events— event type definitionsdocs/site/docs/operations/high-availability.md— operator-facing configuration reference
License
Apache-2.0 — see root LICENSE.
Documentation
¶
Index ¶
Constants ¶
const EpochAnnotation = "haptic.io/leader-epoch"
EpochAnnotation carries the fencing epoch on the leader Lease.
Variables ¶
var ErrForeignLeader = errors.New("the lease is held at a newer epoch")
ErrForeignLeader reports that the Lease names another holder, or an epoch this controller never claimed: a newer leader owns the fleet.
Functions ¶
This section is empty.
Types ¶
type Component ¶
type Component struct {
// contains filtered or unexported fields
}
Component is an event adapter that wraps the pure leader election component and publishes events for observability.
This is the coordination layer that connects the pure k8s/leaderelection package to the controller's event bus.
func New ¶
func New( config *k8sleaderelection.Config, clientset kubernetes.Interface, eventBus *busevents.EventBus, callbacks k8sleaderelection.Callbacks, epoch *Term, logger *slog.Logger, ) (*Component, error)
New creates a new leader election component.
This function wraps the pure leader election elector and adds event publishing for observability. The callbacks provided by the caller are wrapped to also publish events before/after the callback executes. epoch may be nil, in which case no fencing epoch is claimed; the deployer then dispatches at epoch zero, which is correct only for a single writer.
func (*Component) KeepLeaseOnStop ¶
func (c *Component) KeepLeaseOnStop()
KeepLeaseOnStop makes the loop's stop leave the Lease held, for a successor iteration on this replica to resume without any other replica seeing a vacancy.
func (*Component) ReleaseLease ¶
func (c *Component) ReleaseLease()
ReleaseLease vacates a Lease kept for a successor that will not run.
type LeaseEpoch ¶
type LeaseEpoch struct {
// contains filtered or unexported fields
}
LeaseEpoch is the fencing epoch of this controller's leadership terms: a counter on the leader Lease that each term increments before it dispatches anything. Every apply carries it, and a pod that has accepted a higher epoch refuses lower ones — so a controller that lost the lease but has not noticed yet cannot write over its successor.
The Lease's own fields are not usable for this: the identity is the pod name, so client-go carries AcquireTime and LeaderTransitions across a same-identity re-acquire.
func NewLeaseEpoch ¶
func NewLeaseEpoch(client kubernetes.Interface, namespace, name, identity string, logger *slog.Logger) *LeaseEpoch
NewLeaseEpoch builds the epoch source for one Lease.
func (*LeaseEpoch) Bump ¶
func (e *LeaseEpoch) Bump(ctx context.Context) error
Bump claims the next epoch by incrementing the Lease annotation, retrying the read-modify-write while another writer wins the race. It must complete before this term dispatches: an unclaimed epoch is refused by every pod that has seen a higher one, which stalls deployments rather than corrupting them.
func (*LeaseEpoch) Identity ¶
func (e *LeaseEpoch) Identity() string
Identity is this controller's leader-election identity.
func (*LeaseEpoch) LeaderEpoch ¶
func (e *LeaseEpoch) LeaderEpoch() uint64
LeaderEpoch is the epoch this controller last claimed. Zero means it has claimed none, which every pod's fence outranks.
func (*LeaseEpoch) Reclaim ¶
Reclaim lifts this controller's epoch past one a pod has already accepted. A pod outranking the leader means the counter regressed — a Lease that was deleted, recreated or restored from a backup loses the annotation — so the Lease is re-read: while it still names this holder at the epoch this term claimed, no rival exists and the annotation is raised to floor+1. It reports ErrForeignLeader when the Lease proves otherwise, which is the one case where the pod is right and this controller must stop writing.
type Term ¶
type Term struct {
*LeaseEpoch
// contains filtered or unexported fields
}
Term is one leadership term as its writers see it: the identity and epoch every apply carries, and the way to hand the term back when something proves this controller is not the fleet's writer any more.
func NewTerm ¶
func NewTerm(epoch *LeaseEpoch, standDown func(reason string)) *Term
NewTerm binds the term's epoch to the way its holder gives leadership up. standDown must release the Lease — only a fresh acquisition claims a fresh epoch, so anything short of it leaves this replica elected and refused.
Source Files