Documentation
¶
Overview ¶
Package topology provides cluster topology monitoring for drain mode support.
Helix uses a NATS Key-Value store to broadcast topology constraints to all connected clients. This enables operations teams to gracefully drain traffic from a cluster before maintenance (patching, scaling, upgrades) without causing client-side errors.
Overview ¶
The topology package provides implementations of the helix.TopologyWatcher and helix.TopologyOperator interfaces:
- helix.TopologyWatcher: Monitors external signals and emits helix.TopologyUpdate events when cluster availability changes.
- helix.TopologyOperator: Allows setting drain states programmatically.
NATS Topology ¶
NATS watches a NATS KV bucket for drain mode configuration:
nc, _ := nats.Connect("nats://localhost:4222")
js, _ := jetstream.New(nc)
kv, _ := js.KeyValue(ctx, "helix-config")
watcher, _ := topology.NewNATS(kv,
topology.WithKey("topology.drain"), // custom key
)
client, _ := helix.NewCQLClient(sessionA, sessionB,
helix.WithTopologyWatcher(watcher),
)
Drain Configuration Format ¶
The NATS KV value is a JSON object specifying which clusters to drain:
{
"drain": ["B"],
"reason": "OS Patching"
}
Valid drain values are "A", "B", or both. When a cluster is in the drain list, Helix clients will:
- Stop sending writes to the drained cluster (enqueue for replay instead)
- Force read preference away from the drained cluster
- Disable failover attempts to the drained cluster
Lifecycle ¶
Drain mode requires explicit operator actions:
- Start maintenance: PUT the drain configuration to NATS KV
- End maintenance: DELETE the key (or PUT with empty drain list)
There is no automatic expiry. This is intentional to prevent race conditions where clients resume traffic while maintenance is still in progress.
Local Topology ¶
Local provides an in-memory implementation for testing. It implements both helix.TopologyWatcher and helix.TopologyOperator:
local := topology.NewLocal() _ = local.SetDrain(ctx, helix.ClusterB, true, "maintenance") // Simulate drain // Later... _ = local.SetDrain(ctx, helix.ClusterB, false, "") // Clear drain mode
Index ¶
- type DrainConfig
- type Local
- func (l *Local) Close() error
- func (l *Local) GetDrainReason() string
- func (l *Local) IsDraining(cluster types.ClusterID) bool
- func (l *Local) SetDrain(_ context.Context, cluster types.ClusterID, draining bool, reason string) error
- func (l *Local) Watch(ctx context.Context) <-chan helix.TopologyUpdate
- type NATS
- type WatcherConfig
- type WatcherOption
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DrainConfig ¶
type DrainConfig struct {
// Drain lists the clusters currently being drained.
// Valid values: "A", "B", or both.
Drain []types.ClusterID `json:"drain"`
// Reason is a human-readable explanation for the drain.
// Example: "OS Patching", "Scaling", "Upgrade to v4.1"
Reason string `json:"reason,omitempty"`
}
DrainConfig represents the drain mode configuration stored in NATS KV.
This is the JSON structure that operations teams PUT to the KV store to signal cluster maintenance.
func (*DrainConfig) ContainsCluster ¶
func (d *DrainConfig) ContainsCluster(cluster types.ClusterID) bool
ContainsCluster returns true if the given cluster is in the drain list.
Parameters:
- cluster: The cluster ID to check
Returns:
- bool: true if the cluster is being drained
type Local ¶
type Local struct {
// contains filtered or unexported fields
}
Local provides an in-memory topology watcher and operator for testing.
Unlike NATS, this implementation allows programmatic control of drain states, making it ideal for unit tests and demos. It implements both TopologyWatcher (for observing) and TopologyOperator (for controlling drain states).
func NewLocal ¶
func NewLocal() *Local
NewLocal creates a new in-memory topology watcher/operator.
Returns:
- *Local: A new local topology instance
func (*Local) GetDrainReason ¶
GetDrainReason returns the current drain reason, if any.
Returns:
- string: The drain reason, or empty string if not draining
func (*Local) IsDraining ¶
IsDraining returns whether the specified cluster is currently in drain mode.
Parameters:
- cluster: The cluster to check
Returns:
- bool: true if the cluster is being drained
func (*Local) SetDrain ¶
func (l *Local) SetDrain(_ context.Context, cluster types.ClusterID, draining bool, reason string) error
SetDrain sets the drain state for a cluster.
This method emits a TopologyUpdate if the state changes.
Parameters:
- ctx: Context for cancellation. For the local in-memory implementation, this parameter is accepted for interface compliance but not used.
- cluster: The cluster to update
- draining: true to enable drain mode, false to disable
- reason: Human-readable reason for the drain (only used when draining=true)
Returns:
- error: Always nil for local implementation
func (*Local) Watch ¶
func (l *Local) Watch(ctx context.Context) <-chan helix.TopologyUpdate
Watch returns a channel that receives topology updates.
Updates are emitted when SetDrain is called. The channel is closed when Close() is called or the context is cancelled.
Multiple calls to Watch return the same channel; only the first call's context controls the watch lifecycle.
Parameters:
- ctx: Context for cancellation (only used on first call)
Returns:
- <-chan helix.TopologyUpdate: Channel of topology changes
type NATS ¶
type NATS struct {
// contains filtered or unexported fields
}
NATS monitors a NATS KV bucket for drain mode configuration.
It watches a configurable key and emits TopologyUpdate events when the drain status of any cluster changes. This enables operations teams to gracefully drain traffic before cluster maintenance.
Watch() should be called once per instance. Subsequent calls return the same channel. The channel is closed when Close() is called or the context is cancelled.
func NewNATS ¶
func NewNATS(kv jetstream.KeyValue, opts ...WatcherOption) (*NATS, error)
NewNATS creates a new NATS KV topology watcher.
The watcher will begin monitoring the KV bucket for drain configuration when Watch() is called.
Parameters:
- kv: A NATS JetStream KeyValue store
- opts: Optional configuration options
Returns:
- *NATS: A new watcher instance
- error: Error if kv is nil
Example:
nc, _ := nats.Connect("nats://localhost:4222")
js, _ := jetstream.New(nc)
kv, _ := js.KeyValue(ctx, "helix-config")
watcher, _ := topology.NewNATS(kv,
topology.WithKey("topology.drain"),
topology.WithPollInterval(10*time.Second),
)
func (*NATS) Close ¶
Close stops the watcher and releases resources.
This method is safe to call multiple times.
func (*NATS) Config ¶
func (n *NATS) Config() WatcherConfig
Config returns the watcher configuration.
This method is primarily useful for testing to verify configuration options.
Returns:
- WatcherConfig: The current watcher configuration
func (*NATS) GetDrainReason ¶
GetDrainReason returns the current drain reason, if any.
This returns the cached reason from the last processed KV entry. It does not perform a live KV fetch.
Returns:
- string: The drain reason, or empty if not draining
func (*NATS) IsDraining ¶
IsDraining returns whether the specified cluster is currently in drain mode.
This provides a synchronous way to check drain status without waiting for channel updates.
Parameters:
- cluster: The cluster to check
Returns:
- bool: true if the cluster is being drained
func (*NATS) Watch ¶
func (n *NATS) Watch(ctx context.Context) <-chan helix.TopologyUpdate
Watch returns a channel that receives topology updates.
The watcher spawns a background goroutine that monitors the NATS KV key. When the drain configuration changes, it emits TopologyUpdate events for each affected cluster.
The channel is closed when Close() is called or the context is cancelled. Multiple calls to Watch return the same channel; only the first call's context controls the watch lifecycle.
Parameters:
- ctx: Context for cancellation (only used on first call)
Returns:
- <-chan helix.TopologyUpdate: Channel of topology changes
type WatcherConfig ¶
type WatcherConfig struct {
// Key is the NATS KV key to watch for drain configuration.
// Default: "helix.topology.drain"
Key string
// PollInterval is the fallback polling interval if watch fails.
// Default: 5 seconds
PollInterval time.Duration
// InitialFetchTimeout is the timeout for the initial KV fetch.
// Default: 10 seconds
InitialFetchTimeout time.Duration
// Logger receives diagnostic messages about watcher lifecycle events,
// such as falling back to polling when the underlying watch mechanism
// fails. Default: a no-op logger that discards all messages.
Logger types.Logger
}
WatcherConfig holds configuration for topology watchers.
func DefaultWatcherConfig ¶
func DefaultWatcherConfig() WatcherConfig
DefaultWatcherConfig returns a WatcherConfig with sensible defaults.
Returns:
- WatcherConfig: Default configuration
type WatcherOption ¶
type WatcherOption func(*WatcherConfig)
WatcherOption configures a topology watcher.
func WithInitialFetchTimeout ¶
func WithInitialFetchTimeout(d time.Duration) WatcherOption
WithInitialFetchTimeout sets the timeout for the initial KV fetch.
Parameters:
- d: Timeout duration
Returns:
- WatcherOption: Configuration option
func WithKey ¶
func WithKey(key string) WatcherOption
WithKey sets the NATS KV key to watch.
Parameters:
- key: The key name (e.g., "storage.topology.maintenance")
Returns:
- WatcherOption: Configuration option
func WithLogger ¶ added in v1.5.3
func WithLogger(logger types.Logger) WatcherOption
WithLogger sets the logger used for watcher diagnostic messages.
If not set (or set to nil, including a typed nil such as a nil `*myLogger` assigned to the types.Logger parameter), a no-op logger is used that discards all messages. Without normalizing the typed-nil case here, a caller-supplied nil concrete pointer would survive as a non-nil types.Logger interface value and panic the first time a background watch/fetch/parse goroutine invokes it.
Parameters:
- logger: The logger implementation
Returns:
- WatcherOption: Configuration option
func WithPollInterval ¶
func WithPollInterval(d time.Duration) WatcherOption
WithPollInterval sets the fallback polling interval.
If the NATS watch fails or disconnects, the watcher falls back to polling at this interval.
Parameters:
- d: Polling interval duration
Returns:
- WatcherOption: Configuration option
Source Files