events

package
v0.2.0-alpha.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 5, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

README

pkg/events

Generic, domain-agnostic event bus with pub/sub and scatter-gather coordination. The controller pulls this in; everything else in the tree stays clean of it (only pkg/controller/events knows about domain event types).

Module path: gitlab.com/haproxy-haptic/haptic. Source is authoritative (go doc ./pkg/events); this README is a short orientation.

Minimal Usage

import "gitlab.com/haproxy-haptic/haptic/pkg/events"

bus := events.NewEventBus(100)          // pre-start buffer capacity
ch := bus.Subscribe("my-component", 50) // per-subscriber buffer

bus.Start()   // call once all subscribers are registered; releases buffered events

bus.Publish(MyEvent{...})

for event := range ch {
    switch e := event.(type) {
    case MyEvent:
        // ...
    }
}

Every event type implements Event:

type Event interface {
    EventType() string
    Timestamp() time.Time
}

The CoalescibleEvent interface marks events that pkg/controller/coalesce can collapse when bursts arrive faster than a component can process them.

Startup Buffering

Subscribers must register before Start() or they miss the replay of events published during initialisation. The controller's reinitialisation loop does this in a strict order inside iteration.go; bespoke users of this library should follow the same pattern:

  1. NewEventBus(n) — n is the buffer that holds pre-start publishes.
  2. Construct every component. Each calls Subscribe(...) inside its constructor.
  3. bus.Start() — flushes the pre-start buffer to each subscriber's channel.
  4. Start the component goroutines.

Forgetting step 3 is a common footgun: publishes succeed silently (buffered) but nothing ever fires.

Typed Subscriptions

Filter pattern on top of the base Subscribe:

  • SubscribeTypes(name, bufferSize, types...) — filters at the bus, only delivers events whose EventType() is in the list. Cheapest when you only care about a handful of types. Variants: SubscribeTypesLeaderOnly (for components that subscribe after leader election) and SubscribeTypesLossy (silent drops, for observability consumers).

The commentator and anything logging "everything" should use plain Subscribe — filtering there would just hide events.

Scatter-Gather (Request/Response)

Used when multiple independent responders all need to approve something — the canonical caller is pkg/controller/configchange.ConfigChangeHandler, which fans ConfigValidationRequest out to BasicValidator, TemplateValidator, and JSONPathValidator and aggregates the responses. Note that the admission webhook does not use scatter-gather; production calls dryrunvalidator.ValidateDirect synchronously to keep the admission path tight.

req := NewMyRequest(payload)
result, err := bus.Request(ctx, req, events.RequestOptions{
    Timeout:            10 * time.Second,
    ExpectedResponders: []string{"basic", "template", "jsonpath"},
})
if err != nil {
    // timeout or context cancelled
}
for _, resp := range result.Responses {
    // resp.Responder() identifies who sent it, resp.RequestID() ties it back
}

Responders listen on their own subscription, match on request ID, and Publish a Response — there's no direct wiring, they just need to call bus.Publish(resp) with the request ID matching.

Don't nest Request() calls on the same path without spawning a goroutine for the outer call — a responder that blocks on another Request while its own pending request waits can deadlock.

Back-Pressure

Publish is non-blocking. If a subscriber's buffer is full, the event is dropped for that subscriber (others still receive it). The drop accounting splits into two paths (see recordDrop in bus.go):

  • Non-lossy ("critical") subscribers — the default Subscribe / SubscribeTypes. Drops bump DroppedEventsCritical and fire the registered DropCallback (used by pkg/controller/metrics to emit haptic_events_dropped_critical_total and haptic_events_dropped_by_subscriber_total).
  • Lossy subscribers — opt-in via SubscribeLossy / SubscribeTypesLossy. Drops bump DroppedEventsObservability only; the DropCallback is not invoked. Use this for observability subscribers (commentator, debug ring buffer) where occasional drops on a burst are acceptable and shouldn't trip the per-subscriber alert metric.

Slow consumers are your problem — hand work off to a goroutine or raise the subscriber buffer.

Buffer Sizing Rule of Thumb

pkg/events/defaults.go exposes five named constants — prefer them over raw integers so the intent is readable at the call site:

Tier Constant Value For
Low LowVolumeSubscriberBuffer 10 Components that see at most a handful of events per second (leadership transitions, config reloads).
Standard StandardSubscriberBuffer 50 Default for most controller components.
High HighVolumeSubscriberBuffer 100 Reconciliation-path consumers that fan in from many resource types.
Publishing PublishingSubscriberBuffer 200 Components that publish downstream work in response to every event.
Debug DebugSubscriberBuffer 1000 Observability consumers (commentator, debug ring buffer) that must catch everything.

Pre-start buffer on NewEventBus: roughly the number of events you expect during initialisation; 100 is fine for the main controller.

Testing

go test ./pkg/events/...           # unit tests
go test ./pkg/events/... -race     # race detector
go test ./pkg/events/... -bench=.  # pub/sub benchmarks

Tests define ad-hoc Event types inline — the infrastructure never needs domain types.

See Also

  • pkg/events/CLAUDE.md — design rationale, pitfall catalogue (blocking handlers, buffer sizing, Request deadlocks), extension points
  • pkg/events/ringbuffer — generic thread-safe ring buffer used by commentator and debug event history
  • pkg/controller/events — domain event catalogue (~50 types across lifecycle, config, resources, reconciliation, deployment, leader election)
  • pkg/controller/commentator — subscribes to every event for domain-aware logging
  • pkg/controller/coalesce — collapses CoalescibleEvent bursts

License

Apache-2.0 — see root LICENSE.

Documentation

Overview

Package events provides an event bus for component coordination in the HAPTIC controller.

The event bus supports two communication patterns: 1. Async pub/sub: Fire-and-forget event publishing for observability and loose coupling 2. Sync request-response: Scatter-gather pattern for coordinated validation and queries

Index

Constants

View Source 
const (
	// LowVolumeSubscriberBuffer suits components that receive at most a
	// handful of events per reconciliation cycle (for example, config
	// validators that respond to a single request event).
	LowVolumeSubscriberBuffer = 10

	// StandardSubscriberBuffer is the default tier for most controller
	// components - roughly one event per second under normal load.
	StandardSubscriberBuffer = 50

	// HighVolumeSubscriberBuffer covers components that fan in many
	// resource-change or reconciliation-triggered events (the reconciler
	// debouncer, pod discovery).
	HighVolumeSubscriberBuffer = 100

	// PublishingSubscriberBuffer absorbs bursts from publishing paths that
	// batch many small events back-to-back before draining.
	PublishingSubscriberBuffer = 200

	// DebugSubscriberBuffer is used for debug/introspection subscriptions
	// that tap every event flowing through the bus.
	DebugSubscriberBuffer = 1000
)

Buffer size tiers for event subscriptions. Subscribers pick the tier that matches their expected inbound rate so tuning stays centralized and each component avoids defining its own magic number.

When in doubt, start with StandardSubscriberBuffer. Move to HighVolumeSubscriberBuffer once the component is shown to fan in events from many resource types, and to PublishingSubscriberBuffer only for components that briefly buffer bursts of outbound-adjacent events.

View Source 
const (
	// MaxPreStartBufferSize is the maximum number of events that can be buffered
	// before EventBus.Start() is called. This prevents unbounded memory growth
	// during startup if many events are published before subscribers are ready.
	// Events exceeding this limit are dropped with a warning.
	MaxPreStartBufferSize = 1000
)

Variables

This section is empty.

Functions

This section is empty.

Types

type CoalescibleEvent 

type CoalescibleEvent interface {
	Event
	// Coalescible returns true if this event can be safely skipped when a newer
	// event of the same type is available. The emitter sets this based on context:
	// - true: "state update" events where only the latest matters
	// - false: "command" events that must be processed individually
	Coalescible() bool
}

CoalescibleEvent is an optional interface for events that support coalescing. Events implementing this interface can be safely skipped when a newer event of the same type is available in the queue.

This interface follows the Interface Segregation Principle - only events that need coalescing implement it, keeping the base Event interface minimal.

The Coalescible() method is set by the event emitter (not derived from event fields), following the Single Responsibility Principle - the emitter knows the context and decides whether this specific event instance can be coalesced.

type DropCallback 

type DropCallback func(info DropInfo)

DropCallback is called when an event is dropped due to a full subscriber buffer. This callback pattern keeps the EventBus domain-agnostic while allowing the controller layer to handle drops with appropriate logging and metrics.

type DropInfo 

type DropInfo struct {
	EventType      string // The type of event that was dropped
	SubscriberName string // Name of the subscriber whose buffer was full
	BufferSize     int    // Total buffer size
	EventTypes     string // For typed subscriptions, the event types being filtered (comma-separated)
}

DropInfo contains information about a dropped event for debugging.

type Event 

type Event interface {
	// EventType returns a unique identifier for this event type.
	// Convention: use dot-notation like "config.parsed" or "deployment.completed"
	EventType() string

	// Timestamp returns when this event occurred.
	// Used for event correlation and temporal analysis.
	Timestamp() time.Time
}

Event is the base interface for all events in the system. Events are used for asynchronous pub/sub communication between components.

type EventBus 

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

EventBus provides centralized pub/sub coordination for all controller components.

The EventBus supports two patterns: - Publish() for async fire-and-forget events (observability, notifications) - Request() for sync scatter-gather pattern (validation, queries)

EventBus is thread-safe and can be used concurrently from multiple goroutines.

Startup Coordination: Events published before Start() is called are buffered and replayed after Start(). This prevents race conditions during component initialization.

Typed Subscriptions: In addition to universal subscriptions (Subscribe), the EventBus supports typed subscriptions (SubscribeTypes) that filter events at the bus level for efficiency.

Lossy Subscriptions: For observability components where occasional drops are acceptable, use SubscribeLossy(). These subscriptions silently drop events without triggering the onDrop callback, and drops are counted separately in DroppedEventsObservability().

Event Drop Monitoring: When subscriber buffers are full, events are dropped to prevent blocking. Use SetDropCallback() to receive notifications when critical drops occur. DroppedEventsCritical() returns drops from business-critical subscribers. DroppedEventsObservability() returns expected drops from lossy subscribers.

func NewEventBus 

func NewEventBus(capacity int) *EventBus

NewEventBus creates a new EventBus.

The bus starts in buffering mode - events published before Start() is called will be buffered and replayed when Start() is invoked. This ensures no events are lost during component initialization.

The capacity parameter sets the initial buffer size for pre-start events. Recommended: 100 for most applications.

func (*EventBus) DroppedEventsCritical 

func (b *EventBus) DroppedEventsCritical() uint64

DroppedEventsCritical returns the number of events dropped from business-critical (non-lossy) subscribers.

Non-zero values indicate a problem that needs attention - critical subscribers are not keeping up with event volume.

func (*EventBus) DroppedEventsObservability 

func (b *EventBus) DroppedEventsObservability() uint64

DroppedEventsObservability returns the number of events dropped from lossy subscribers (observability components like commentator, debug/events).

Non-zero values are expected and acceptable during high load. These drops don't affect controller operation - they just mean some log entries or debug info was skipped.

func (*EventBus) Pause 

func (b *EventBus) Pause()

Pause temporarily suspends event delivery, buffering events for later replay.

This reuses the existing preStartBuffer infrastructure used during startup. Events published while paused are buffered and will be replayed when Start() is called again.

Use cases:

  • Leadership transition (pause while starting leader-only components)
  • Hot reload scenarios
  • Testing

This method is idempotent - calling it when already paused has no effect. Thread-safe and can be called concurrently with Publish() and Subscribe().

Example: 

// During leadership transition
bus.Pause()                                    // Buffer events
bus.Publish(BecameLeaderEvent{})               // Buffered
startLeaderOnlyComponents()                    // Components subscribe
bus.Start()                                    // Replay buffered events

func (*EventBus) Publish 

func (b *EventBus) Publish(event Event) int

Publish sends an event to all subscribers.

If Start() has not been called yet, the event is buffered and will be replayed when Start() is invoked. This prevents events from being lost during component initialization.

After Start() is called, this is a non-blocking operation. If a subscriber's channel is full, the event is dropped for that subscriber to prevent slow consumers from blocking the entire system.

Returns the number of subscribers that successfully received the event. Returns 0 if event was buffered (before Start()).

func (*EventBus) Request 

func (b *EventBus) Request(ctx context.Context, request Request, opts RequestOptions) (*RequestResult, error)

Request sends a request event and waits for responses using the scatter-gather pattern.

This is a synchronous operation that: 1. Publishes the request event to all subscribers (scatter phase) 2. Collects response events matching the request ID (gather phase) 3. Returns when all expected responders have replied or timeout occurs

The request must implement the Request interface to provide a unique RequestID for correlating responses.

Use this method when you need coordinated responses from multiple components, such as multi-phase validation or distributed queries.

Example: 

req := NewConfigValidationRequest(config, version)
result, err := bus.Request(ctx, req, RequestOptions{
    Timeout: 10 * time.Second,
    ExpectedResponders: []string{"basic", "template", "jsonpath"},
})

func (*EventBus) SetDropCallback 

func (b *EventBus) SetDropCallback(cb DropCallback)

SetDropCallback sets a callback to be invoked when events are dropped from CRITICAL (non-lossy) subscriber buffers. The callback receives the event type string.

This callback is NOT called for lossy subscribers (created via SubscribeLossy()). Lossy drops are expected and silently counted in DroppedEventsObservability().

This callback pattern keeps the EventBus domain-agnostic while allowing the controller layer to handle drops with appropriate logging and metrics.

Set to nil to disable drop notifications.

Example: 

bus.SetDropCallback(func(info DropInfo) {
    slog.Warn("Event dropped from critical subscriber",
        "event_type", info.EventType,
        "subscriber", info.SubscriberName,
        "buffer_size", info.BufferSize)
    metrics.EventsDroppedCritical.Inc()
})

func (*EventBus) Start 

func (b *EventBus) Start()

Start releases all buffered events and switches the bus to normal operation mode.

This method should be called after all components have subscribed to the bus during application startup. It ensures that no events are lost during the initialization phase.

Behavior:

  1. Marks the bus as started
  2. Replays all buffered events to subscribers in order
  3. Clears the buffer
  4. All subsequent Publish() calls go directly to subscribers

This method is idempotent - calling it multiple times has no additional effect. Thread-safe and can be called concurrently with Publish() and Subscribe().

Example: 

bus := NewEventBus(100)

// Components subscribe during setup
commentator := NewEventCommentator(bus, logger, 1000)
validator := NewValidator(bus)
// ... more subscribers ...

// Release buffered events
bus.Start()

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe(name string, bufferSize int) <-chan Event

Subscribe creates a new subscription to the event bus.

The returned channel will receive all events published to the bus. The bufferSize parameter controls the channel buffer size - larger buffers reduce the chance of dropped events for slow consumers.

Subscribers must continuously read from the channel to avoid dropped events. A bufferSize of 100 is recommended for most use cases.

The returned channel is read-only and will never be closed.

IMPORTANT: For all-replica components, call this method BEFORE EventBus.Start() to ensure buffered events are received. Subscribing after Start() will trigger a warning as it may indicate a bug. For leader-only components that intentionally subscribe late (after leader election), use SubscribeTypesLeaderOnly() instead.

Parameters:

  • name: Subscriber name for debugging (e.g., "commentator", "reconciler")
  • bufferSize: Size of the channel buffer

func (*EventBus) SubscribeLossy 

func (b *EventBus) SubscribeLossy(name string, bufferSize int) <-chan Event

SubscribeLossy creates a subscription that silently drops events when buffer is full.

Use this for observability components (like commentator, debug/events) where occasional event drops are acceptable and expected during high load. Drops from lossy subscribers:

  • Are counted in DroppedEventsObservability() (for metrics)
  • Do NOT trigger the onDrop callback (no WARN logs)

This prevents log spam from expected drops in observability components while still allowing monitoring via metrics.

The returned channel is read-only and will never be closed.

Parameters:

  • name: Subscriber name for debugging (e.g., "commentator")
  • bufferSize: Size of the channel buffer

func (*EventBus) SubscribeTypes 

func (b *EventBus) SubscribeTypes(name string, bufferSize int, eventTypes ...string) <-chan Event

SubscribeTypes creates a subscription that only receives events of the specified types.

This is more efficient than universal Subscribe() when a component only cares about specific event types, as filtering happens at the EventBus level rather than in each subscriber's event loop.

Parameters:

  • name: Subscriber name for debugging (e.g., "reconciler", "renderer")
  • bufferSize: Size of the output channel buffer
  • eventTypes: Event type strings to filter for (from Event.EventType())

Returns a channel that receives only events matching the specified types. The channel is read-only and will never be closed.

To stop receiving events and prevent memory leaks, call UnsubscribeTyped() with the returned channel.

Example: 

eventChan := bus.SubscribeTypes("executor", 100,
    "reconciliation.triggered",
    "template.rendered",
    "validation.completed")
defer bus.UnsubscribeTyped(eventChan) // Clean up when done
for event := range eventChan {
    // Only receives the specified event types
}

func (*EventBus) SubscribeTypesLeaderOnly 

func (b *EventBus) SubscribeTypesLeaderOnly(name string, bufferSize int, eventTypes ...string) <-chan Event

SubscribeTypesLeaderOnly creates a typed subscription for leader-only components.

This method is identical to SubscribeTypes() but is semantically named to indicate it's intended for leader-only components that subscribe after leader election.

Leader-only components rely on the state replay mechanism: all-replica components re-publish their cached state when BecameLeaderEvent is received, ensuring leader-only components don't miss critical state even though they subscribe late.

Parameters:

  • name: Subscriber name for debugging (e.g., "deployer", "scheduler")
  • bufferSize: Size of the output channel buffer
  • eventTypes: Event type strings to filter for (from Event.EventType())

Returns a channel that receives only events matching the specified types. The channel is read-only and will never be closed.

To stop receiving events and prevent memory leaks, call UnsubscribeTyped() with the returned channel.

Example: 

// In a leader-only component's constructor (after BecameLeaderEvent)
eventChan := bus.SubscribeTypesLeaderOnly("scheduler", 50,
    events.EventTypeTemplateRendered,
    events.EventTypeValidationCompleted,
    events.EventTypeLostLeadership)
defer bus.UnsubscribeTyped(eventChan)

func (*EventBus) SubscriberCount 

func (b *EventBus) SubscriberCount() int

SubscriberCount returns the number of active subscriptions (universal + typed). All subscriptions are created before Start(), so this is safe to call from any goroutine.

func (*EventBus) UnsubscribeTyped 

func (b *EventBus) UnsubscribeTyped(ch <-chan Event)

UnsubscribeTyped removes a typed subscription from the event bus.

This method should be called when a subscriber no longer needs to receive events from a typed subscription (created via SubscribeTypes), to prevent memory leaks.

Note: The channel is not closed by this method. The subscriber is responsible for draining any remaining events from the channel if needed.

This method is safe to call multiple times for the same channel.

type Request 

type Request interface {
	Event
	// RequestID returns a unique identifier for this request.
	// Responses must include this ID to be correlated correctly.
	RequestID() string
}

Request is the interface for request events in the scatter-gather pattern.

Requests are broadcast to all subscribers (scatter phase), and responses are correlated by RequestID (gather phase).

type RequestOptions 

type RequestOptions struct {
	// Timeout is the maximum time to wait for responses.
	// If zero, defaults to 10 seconds.
	Timeout time.Duration

	// ExpectedResponders lists the names of components expected to respond.
	// If empty, the request will wait indefinitely for responses.
	ExpectedResponders []string
}

RequestOptions configures the behavior of a scatter-gather request.

type RequestResult 

type RequestResult struct {
	// Responses contains all responses received before timeout or completion.
	Responses []Response

	// Errors contains error messages for responders that didn't respond
	// or timed out. Empty if all expected responders replied.
	Errors []string
}

RequestResult contains the aggregated results from a scatter-gather request.

type Response 

type Response interface {
	Event
	// RequestID returns the ID of the request this response belongs to.
	RequestID() string
	// Responder returns the name of the component that sent this response.
	Responder() string
}

Response is the interface for response events in the scatter-gather pattern.

Each response includes the original RequestID and the name of the responder for tracking and debugging purposes.

Source Files

View all Source files

Directories

Path Synopsis
Package ringbuffer provides a thread-safe generic ring buffer implementation.
 Package ringbuffer provides a thread-safe generic ring buffer implementation.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.