lifecycle

type Component ¶

type Component interface {
	// Name returns a unique identifier for this component.
	// This is used for logging and status tracking.
	Name() string

	// Start begins the component's operation.
	// This method should block until ctx is cancelled or an error occurs.
	// Returning nil indicates graceful shutdown.
	Start(ctx context.Context) error
}

Component is the minimal interface for components managed by the Registry.

Components must provide a unique name and a Start method. The Start method should block until the context is cancelled or an error occurs.

type ComponentInfo ¶

type ComponentInfo struct {
	// Name is the component's unique identifier.
	Name string `json:"name"`

	// Status is the current lifecycle status.
	Status Status `json:"status"`

	// LeaderOnly indicates if this component runs only on the leader.
	LeaderOnly bool `json:"leader_only,omitempty"`

	// Error contains the last error message if Status is Failed.
	Error string `json:"error,omitempty"`

	// Healthy indicates the result of the last health check (if supported).
	// nil means health check not supported, true means healthy, false means unhealthy.
	Healthy *bool `json:"healthy,omitempty"`
}

ComponentInfo provides information about a registered component.

type HealthChecker ¶

type HealthChecker interface {
	// HealthCheck returns nil if the component is healthy, or an error describing
	// the health issue.
	HealthCheck() error
}

HealthChecker is an optional interface for components that support health checks.

Components implementing this interface will have their health status periodically checked and exposed via the registry's status endpoint.

type HealthTracker ¶

type HealthTracker struct {
	// contains filtered or unexported fields
}

HealthTracker provides stall detection for controller components.

It supports two tracking modes that can be used independently or together:

Activity-based tracking (for timer-based components like DriftMonitor):

• Call RecordActivity() whenever the timer fires
• CheckActivity() returns error if no activity for > timeout
• Use ActivityStallTimeout() to calculate timeout from interval

Processing-based tracking (for event-driven components like ConfigPublisher):

• Call StartProcessing() before handling an event
• Call EndProcessing() when done (including on error paths!)
• CheckProcessing() returns error if processing takes > timeout
• Idle state (no active processing) is always healthy

Components should call Check() which combines both checks.

func NewActivityTracker(componentName string, timeout time.Duration) *HealthTracker

func NewProcessingTracker(componentName string, timeout time.Duration) *HealthTracker

func (t *HealthTracker) Check() error

func (t *HealthTracker) EndProcessing()

func (t *HealthTracker) RecordActivity()

func (t *HealthTracker) StartProcessing()

type Registry ¶

type Registry struct {
	// contains filtered or unexported fields
}

Registry manages component lifecycles.

The Registry provides:

• Component registration with options (leader-only)
• Concurrent component startup
• Status tracking and health checks
• Leader-only component management

Example:

registry := lifecycle.NewRegistry()
registry.Register(reconciler.New(bus, logger), false)
registry.Register(deployer.New(bus, logger), true)

// StartAll requires isLeader so leader-only components can be skipped
// on follower replicas (they're started later via
// StartLeaderOnlyComponentsAsync on the elected leader).
err := registry.StartAll(ctx, isLeader)

func NewRegistry() *Registry

func (r *Registry) Count() int

func (r *Registry) Register(c Component, leaderOnly bool)

registry.Register(reconciler.New(bus, logger), false)
registry.Register(deployer.New(bus, logger), true)

func (r *Registry) StartAll(ctx context.Context, isLeader bool) error

err := registry.StartAll(ctx, isLeader)
if err != nil {
    return fmt.Errorf("starting components: %w", err)
}

func (r *Registry) StartLeaderOnlyComponentsAsync(ctx context.Context) (<-chan error, error)

func (h *leadershipHandler) onBecameLeader(ctx context.Context) error {
    errCh, err := h.registry.StartLeaderOnlyComponentsAsync(ctx)
    if err != nil {
        return err
    }
    // Components are now subscribed, safe to call eventBus.Start()
    // Track errors asynchronously
    go func() {
        if err := <-errCh; err != nil {
            log.Error("Leader component failed", "error", err)
        }
    }()
    return nil
}

func (r *Registry) Status() map[string]ComponentInfo

func (r *Registry) WithLogger(logger *slog.Logger) *Registry

type Status ¶

type Status string

Status represents the current lifecycle state of a component.

const (
	// StatusPending indicates the component has been registered but not yet started.
	StatusPending Status = "pending"

	// StatusStarting indicates the component is in the process of starting.
	StatusStarting Status = "starting"

	// StatusRunning indicates the component is running normally.
	StatusRunning Status = "running"

	// StatusFailed indicates the component failed to start or encountered a fatal error.
	StatusFailed Status = "failed"

	// StatusStopped indicates the component has been gracefully stopped.
	StatusStopped Status = "stopped"

	// StatusStandby indicates the component is intentionally not active.
	// This is used for leader-only components on non-leader pods that are waiting
	// for potential leadership acquisition. Unlike StatusPending (which implies
	// "about to start"), StatusStandby means "waiting for conditions to be met".
	StatusStandby Status = "standby"
)

type SubscriptionReadySignaler ¶

type SubscriptionReadySignaler interface {
	// SubscriptionReady returns a channel that is closed when the component has
	// completed its event subscription and is ready to receive events.
	// The registry will wait for this channel before considering the component
	// ready.
	SubscriptionReady() <-chan struct{}
}

SubscriptionReadySignaler is an optional interface for components that subscribe to events during Start() rather than during construction.

Leader-only components typically subscribe in Start() because they're created during construction but only started when leadership is acquired. This interface allows such components to signal when subscription is complete, ensuring the registry waits for subscription before considering the component "ready".

Without this interface, there's a race condition where:

1. Component's Start() goroutine is launched
2. Registry marks component as "ready"
3. EventBus replays buffered events
4. But the component hasn't subscribed yet (Start() hasn't executed enough)
5. Component misses critical events

By implementing this interface, the component can signal after subscription:

func (c *Component) Start(ctx context.Context) error {
    c.eventChan = c.eventBus.Subscribe(ComponentName, 100)  // Subscribe (name, buffer)
    close(c.subscriptionReady)                                // Signal ready
    // ... event loop
}

func (c *Component) SubscriptionReady() <-chan struct{} {
    return c.subscriptionReady
}