types

package
v1.6.0 Latest Latest
Warning

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

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

Documentation

Overview

Package types provides shared types and error definitions for the helix library.

This is a leaf package with zero helix imports to prevent import cycles. All packages in helix can safely import this package.

Types

ClusterID identifies which cluster is being referenced: 

const (
    ClusterA ClusterID = 0
    ClusterB ClusterID = 1
)

Consistency levels mirror gocql consistency levels for database operations: 

const (
    Any         Consistency = 0x00
    One         Consistency = 0x01
    Two         Consistency = 0x02
    Three       Consistency = 0x03
    Quorum      Consistency = 0x04
    All         Consistency = 0x05
    LocalQuorum Consistency = 0x06
    EachQuorum  Consistency = 0x07
    LocalOne    Consistency = 0x0A
)

Errors

Sentinel errors are provided for common failure scenarios:

  • ErrBothClustersFailed: Both clusters failed during a dual-write operation
  • ErrNilSession: A nil session was provided
  • ErrNotConnected: Client is not connected to any cluster
  • ErrReplayQueueFull: Replay queue has reached capacity
  • ErrNoHealthyCluster: No healthy cluster available for the operation

ReplayPayload

ReplayPayload carries failed write operations for asynchronous reconciliation: 

type ReplayPayload struct {
    TargetCluster ClusterID
    Query         string
    Args          []any
    Priority      PriorityLevel
    Timestamp     time.Time
}

Package types provides shared types and errors for the Helix library.

This is a "leaf" package with no imports from other helix packages, allowing it to be imported by any package without causing import cycles.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrBothClustersFailed indicates that a write failed on both clusters.
	// This is returned to the caller as a hard failure.
	ErrBothClustersFailed = errors.New("helix: write failed on both clusters")

	// ErrBothClustersDraining indicates both clusters are in drain mode.
	// No writes can be performed until at least one cluster exits drain mode.
	ErrBothClustersDraining = errors.New("helix: both clusters are draining")

	// ErrSessionClosed indicates an operation was attempted on a closed session.
	ErrSessionClosed = errors.New("helix: session is closed")

	// ErrReplayQueueFull indicates the in-memory replay queue is at capacity.
	// The failed write could not be enqueued for later reconciliation.
	ErrReplayQueueFull = errors.New("helix: replay queue is full")

	// ErrNilSession indicates that a nil session was provided.
	ErrNilSession = errors.New("helix: session cannot be nil")

	// ErrWriteAsync indicates a write was sent asynchronously (fire-and-forget).
	// This is returned by AdaptiveDualWrite when a cluster is degraded.
	// The write is still attempted in the background, but the caller should
	// not wait for it. Replay system handles reconciliation if it fails.
	ErrWriteAsync = errors.New("helix: write sent asynchronously to degraded cluster")

	// ErrWriteDropped indicates a fire-and-forget write was dropped due to
	// concurrency limit. This protects the application from resource exhaustion
	// when a degraded cluster is slow. The replay system handles reconciliation.
	ErrWriteDropped = errors.New("helix: write dropped due to fire-and-forget concurrency limit")

	// ErrNoValidClusters indicates that the allowed-clusters override and
	// drain state conflict, leaving no cluster available for reads.
	// The operator must resolve the conflict (adjust the override or clear drain).
	ErrNoValidClusters = errors.New("helix: no valid clusters for read — override and drain state conflict")

	// ErrInvalidClusterOverride indicates that the AllowedClustersFunc returned
	// only unknown ClusterIDs, or targeted an unconfigured cluster in
	// single-cluster mode. This is a fail-closed condition.
	ErrInvalidClusterOverride = errors.New("helix: invalid cluster override — no recognized clusters in returned list")

	// ErrClusterOverridePanic indicates that the AllowedClustersFunc panicked.
	// This is a fail-closed condition; the panic is recovered and the read fails.
	ErrClusterOverridePanic = errors.New("helix: cluster override function panicked")

	// ErrInvalidCluster indicates an operation referenced a cluster the client
	// is not configured for (e.g., SwapSession(ClusterB, …) on a single-cluster
	// client) or used an unknown ClusterID value.
	//
	// Distinct from ErrInvalidClusterOverride, which is read-override-specific.
	ErrInvalidCluster = errors.New("helix: invalid cluster for this client")

	// ErrNoSessionRefresher indicates RefreshSession was called but no
	// SessionRefresher was registered via WithSessionRefresher. The caller
	// must either register one at construction or use the lower-level
	// SwapSession to provide a freshly-built session directly.
	ErrNoSessionRefresher = errors.New("helix: no session refresher configured")

	// ErrClusterDegraded indicates a Strict() write was skipped because the
	// cluster is currently flagged degraded by AdaptiveDualWrite. The write
	// was not sent to that cluster and was not enqueued for replay.
	ErrClusterDegraded = errors.New("helix: cluster is degraded; strict write skipped")

	// ErrClusterDraining indicates a Strict() write was skipped because the
	// cluster is currently in topology drain mode. The write was not sent to
	// that cluster and was not enqueued for replay.
	ErrClusterDraining = errors.New("helix: cluster is draining; strict write skipped")

	// ErrStrictUnsupported indicates the configured WriteStrategy does not
	// implement StrictWriter and a Strict() statement was attempted. Switch to
	// ConcurrentDualWrite, SyncDualWrite, or AdaptiveDualWrite to use Strict().
	ErrStrictUnsupported = errors.New("helix: configured WriteStrategy does not support Strict() writes")

	// ErrStrictMirrorUnsupported indicates Strict() and Mirror() were combined
	// on one statement. The combination is rejected before attempting the write:
	// strict writes are commonly replay-unsafe, and mirror destinations may
	// retry or replay failed dispatches.
	ErrStrictMirrorUnsupported = errors.New("helix: Strict() and Mirror() cannot be combined")

	// ErrMirrorModeConflict indicates that both WithMirror and
	// WithMirrorPublisher were configured. The two mirror modes are
	// mutually exclusive: target mode dispatches writes from this process,
	// publisher mode publishes captures for an out-of-process consumer.
	ErrMirrorModeConflict = errors.New("helix: WithMirror and WithMirrorPublisher are mutually exclusive")

	// ErrNilMirrorTarget indicates that NewMirrorWorker or NewCQLClient
	// (with WithMirror) was called with a nil mirror destination CQLClient.
	ErrNilMirrorTarget = errors.New("helix: mirror target cannot be nil")

	// ErrNilMirrorPublisher indicates that NewCQLClient was called with
	// WithMirrorPublisher and a nil Replayer.
	ErrNilMirrorPublisher = errors.New("helix: mirror publisher cannot be nil")

	// ErrNotFound indicates that a query returned zero rows.
	//
	// This is the Helix sentinel for "not found" results, mapped from
	// gocql.ErrNotFound at the adapter layer. It is NOT treated as a cluster
	// health failure — Helix never records this as a read error or failover trigger.
	//
	// Use [IsNotFound] to check for this error, or errors.Is(err, ErrNotFound).
	ErrNotFound = errors.New("helix: not found")

	// ErrRowLimitExceeded indicates that a bounded multi-row read exceeded
	// its row limit (per-query MaxRows or Config.DefaultMaxRows).
	//
	// This is an application-level cap, not a cluster fault. Like [ErrNotFound],
	// it is NOT treated as a cluster health failure: Helix never records it
	// as a read error, never advances circuit-breaker / auto-refresh state,
	// and never triggers FallbackRead empty-retry. It is propagated to the
	// caller as-is across both clusters, including the FallbackRead alt path.
	//
	// Use [IsRowLimitExceeded] to check for this error, or
	// errors.Is(err, ErrRowLimitExceeded).
	ErrRowLimitExceeded = errors.New("helix: row limit exceeded")
)

Sentinel errors for common failure scenarios.

Functions

func IsNotFound added in v1.1.0 

func IsNotFound(err error) bool

IsNotFound reports whether err is a "not found" result.

Returns true for ErrNotFound and any error wrapping it. Use this instead of errors.Is(err, gocql.ErrNotFound) — the adapter layer maps the driver-specific error to this sentinel.

func IsOptionError added in v1.4.0 

func IsOptionError(err error) bool

IsOptionError reports whether err contains an OptionError.

func IsPartialWrite added in v1.4.0 

func IsPartialWrite(err error) bool

IsPartialWrite reports whether err contains a *PartialWriteError.

func IsRowLimitExceeded added in v1.5.0 

func IsRowLimitExceeded(err error) bool

IsRowLimitExceeded reports whether err is a row-limit-exceeded result.

Returns true for ErrRowLimitExceeded and any error wrapping it.

Types

type AdaptiveWriteMetrics added in v1.6.0 

type AdaptiveWriteMetrics interface {
	// SetWriteDegraded sets the degraded-state gauge for a cluster.
	// Metric: [prefix]_write_degraded{cluster="..."} (1=degraded, 0=healthy)
	SetWriteDegraded(cluster ClusterID, degraded bool)

	// IncWriteDegraded increments the healthy-to-degraded transition counter.
	// Metric: [prefix]_write_degraded_total{cluster="..."}
	IncWriteDegraded(cluster ClusterID)

	// IncWriteRecovered increments the degraded-to-healthy transition counter.
	// Metric: [prefix]_write_recovered_total{cluster="..."}
	IncWriteRecovered(cluster ClusterID)
}

AdaptiveWriteMetrics is an OPTIONAL interface that MetricsCollector implementations may satisfy to receive AdaptiveDualWrite health-state transitions (see policy.AdaptiveDualWrite).

The strategy type-asserts on this interface and silently no-ops if the configured collector does not implement it. By-hand MetricsCollector implementations stay source-compatible across this release and may opt in to the new metrics later by adding the three methods. Bundled collectors (e.g. contrib/metrics/vm) implement this interface directly.

All three methods are called after the cluster's state mutex has been released, once per actual transition — never per write. A cluster that is already degraded produces no further calls until it recovers.

Semantics:

  • SetWriteDegraded: gauge of the cluster's current write mode (true = degraded fire-and-forget, false = healthy synchronous). Gauge writes are sequenced per cluster: a transition report that another transition has already superseded skips the write, so the gauge always ends at the newest latched state.
  • IncWriteDegraded: incremented once per healthy-to-degraded transition (threshold-based or ForceDegrade), including superseded ones — the counter is cumulative.
  • IncWriteRecovered: incremented once per degraded-to-healthy transition (fast-strike recovery, ForceRecover, or Reset).

type BatchStatement 

type BatchStatement struct {
	// Query is the CQL statement.
	Query string

	// Args are the bound values for the query.
	Args []any
}

BatchStatement represents a single statement in a batch for replay.

type BatchType 

type BatchType byte

BatchType represents the type of batch operation. 

const (
	LoggedBatch   BatchType = 0
	UnloggedBatch BatchType = 1
	CounterBatch  BatchType = 2
)

Batch types matching gocql.

WARNING: CounterBatch operations are NOT idempotent. Counter updates (e.g., "UPDATE ... SET counter = counter + 1") are additive, so replaying them after a partial failure will cause double-counting. Do not use CounterBatch with the Helix Replay System if you require exactly-once semantics. Consider using a separate reconciliation strategy for counters.

type ClusterError 

type ClusterError struct {
	// Cluster identifies which cluster the error came from.
	Cluster string

	// Operation describes what operation failed.
	Operation string

	// Cause is the underlying error.
	Cause error
}

ClusterError wraps an error from a specific cluster.

func (*ClusterError) Error 

func (e *ClusterError) Error() string

Error implements the error interface.

If Cause is nil, "<nil>" is substituted in its place rather than panicking.

func (*ClusterError) Unwrap 

func (e *ClusterError) Unwrap() error

Unwrap returns the underlying cause for errors.Is/As compatibility.

type ClusterEvent added in v1.6.0 

type ClusterEvent struct {
	// Kind identifies the transition category.
	Kind ClusterEventKind

	// Cluster is the cluster this event primarily concerns.
	Cluster ClusterID

	// FromCluster and ToCluster describe direction for EventFailover.
	FromCluster ClusterID
	ToCluster   ClusterID

	// Timestamp is when the event was recorded (stamped at transition
	// time for policy events, at emission for client events).
	Timestamp time.Time

	// Err is the error that triggered the event, when applicable.
	Err error

	// Reason is a short human-readable cause (e.g. "slow-strike threshold
	// reached", "manual"), when applicable.
	Reason string

	// Count is a kind-specific counter (failure count, strike count),
	// when applicable.
	Count int
}

ClusterEvent describes an operationally significant cluster-health transition observed by a Helix client or one of its policies.

Delivery is asynchronous and best-effort: a bounded buffer absorbs bursts, and events are dropped (and counted) rather than ever blocking a read/write operation. Treat this as an alerting/notification stream, not a durable audit log. Every kind has a metric counterpart — read rates and current state from the metric and use the event as the push notification. See docs/cluster-events.md for the kind-to-metric table.

Ordering: events produced by circuit-breaker and adaptive-write state transitions are delivered in per-cluster transition order, per policy instance; the two policy types keep separate queues, so their events are not ordered against each other. Order holds among the events actually delivered — a drop can remove either end of an open/closed pair, so a handler must tolerate an unmatched close. Events from independent producers (failover, read divergence, replay drops, drain transitions, session refresh) are delivered in enqueue order with no cross-kind causal guarantee. Metric updates and log lines may become visible before or after the corresponding handler invocation.

EventFailover and EventReadDivergence fire once per affected read rather than once per state change, so they arrive at read rate during an outage and are the kinds most likely to be dropped.

Kind and Timestamp are always set. Of the remaining fields, only the ones relevant to a given Kind are populated; the rest hold zero values.

Field population by Kind:

  • EventFailover: FromCluster, ToCluster, Cluster (= ToCluster), Err
  • EventReadDivergence: Cluster (cluster missing the row), Reason (always "row found on alternative cluster after not-found")
  • EventCircuitBreakerOpen: Cluster, Count (failures at trip)
  • EventCircuitBreakerClosed: Cluster, Reason
  • EventWriteDegraded: Cluster, Count (slow strikes), Reason
  • EventWriteRecovered: Cluster, Reason
  • EventDrainEntered / EventDrainExited: Cluster
  • EventReplayDropped: Cluster (replay target), Err (enqueue error)
  • EventMirrorReplayDropped: Err (enqueue error), Reason; Cluster unset
  • EventSessionRefreshAttempt: Cluster, Count (qualifying failures)
  • EventSessionRefreshSuccess: Cluster
  • EventSessionRefreshError: Cluster, Err

type ClusterEventEmitter added in v1.6.0 

type ClusterEventEmitter interface {
	// EmitClusterEvent delivers one event. Should not block.
	EmitClusterEvent(event ClusterEvent)
}

ClusterEventEmitter delivers ClusterEvents to a registered handler.

Contract: implementations must be safe for concurrent use and should return quickly. Helix never invokes an emitter while holding policy state locks (policy transitions enqueue to an internal outbox and the emitter runs after the locks are released), so a slow emitter cannot deadlock or stall policy state — but it does delay the read/write goroutine that performed the transition, exactly like a slow Logger. Reentrant calls from an emitter back into the emitting policy are safe (they enqueue and return) but discouraged. Helix's internal dispatcher (registered via helix.WithOnClusterEvent) is non-blocking: atomics plus a buffered non-blocking send.

type ClusterEventKind added in v1.6.0 

type ClusterEventKind string

ClusterEventKind identifies the category of a ClusterEvent.

Values are stable snake_case strings suitable for direct use as log fields or alert labels. 

const (
	// EventFailover fires when a read fails on the selected cluster and is
	// retried on the alternative cluster. FromCluster/ToCluster identify
	// the direction; Err carries the error that triggered the failover.
	EventFailover ClusterEventKind = "failover"

	// EventReadDivergence fires when a fallback read finds a row on the
	// alternative cluster after the selected cluster returned not-found.
	// Cluster identifies the cluster that was missing the row (replay lag).
	// Reason is always "row found on alternative cluster after not-found".
	EventReadDivergence ClusterEventKind = "read_divergence"

	// EventCircuitBreakerOpen fires when a circuit breaker trips open for
	// a cluster. Count carries the consecutive-failure count at trip time.
	EventCircuitBreakerOpen ClusterEventKind = "circuit_breaker_open"

	// EventCircuitBreakerClosed fires when a previously open circuit
	// breaker closes. Reason distinguishes the two causes: a successful
	// operation ("operation succeeded"), or the reset timeout elapsing
	// with no recovery, which ends the open span on the next recorded
	// failure ("reset timeout elapsed").
	EventCircuitBreakerClosed ClusterEventKind = "circuit_breaker_closed"

	// EventWriteDegraded fires when AdaptiveDualWrite transitions a cluster
	// into degraded (fire-and-forget) mode. Count carries the slow-strike
	// count; Reason distinguishes threshold-based from manual transitions.
	EventWriteDegraded ClusterEventKind = "write_degraded"

	// EventWriteRecovered fires when AdaptiveDualWrite transitions a
	// cluster back to healthy. Reason distinguishes fast-strike recovery
	// from manual recovery ("manual", "manual reset").
	EventWriteRecovered ClusterEventKind = "write_recovered"

	// EventDrainEntered fires when a cluster enters drain mode via the
	// topology watcher.
	EventDrainEntered ClusterEventKind = "drain_entered"

	// EventDrainExited fires when a cluster exits drain mode.
	EventDrainExited ClusterEventKind = "drain_exited"

	// EventReplayDropped fires when a failed write cannot be enqueued for
	// replay (queue full or unavailable) — potential data loss. Cluster is
	// the replay target; Err carries the enqueue error. For payload
	// access, use helix.WithOnReplayDropped.
	EventReplayDropped ClusterEventKind = "replay_dropped"

	// EventMirrorReplayDropped fires when a failed mirror write cannot be
	// enqueued for mirror replay — potential mirror-target data loss. Err
	// carries the enqueue error. Cluster is unset: mirror payloads target
	// a logical sink, not one of this client's clusters. This event fires
	// only while Helix's internal mirror error handler is installed — a
	// caller-supplied mirror.WithOnError replaces that handler (existing
	// "caller options win" semantics) and with it this event.
	EventMirrorReplayDropped ClusterEventKind = "mirror_replay_dropped"

	// EventSessionRefreshAttempt fires when the auto-refresh detector
	// decides a cluster's session is permanently dead and invokes the
	// SessionRefresher. Count carries the qualifying consecutive-failure
	// count observed at the trigger decision.
	EventSessionRefreshAttempt ClusterEventKind = "session_refresh_attempt"

	// EventSessionRefreshSuccess fires after a successful session refresh.
	EventSessionRefreshSuccess ClusterEventKind = "session_refresh_success"

	// EventSessionRefreshError fires when a session refresh attempt fails.
	// Err carries the refresh error.
	EventSessionRefreshError ClusterEventKind = "session_refresh_error"
)

type ClusterEventMetrics added in v1.6.0 

type ClusterEventMetrics interface {
	// AddClusterEventsDropped adds n (always > 0) to the dropped-event
	// counter. Called with the delta accumulated since the previous call.
	// Metric: [prefix]_cluster_events_dropped_total
	AddClusterEventsDropped(n int)
}

ClusterEventMetrics is an OPTIONAL interface that MetricsCollector implementations may satisfy to receive the cluster event dispatcher's drop total (see helix.WithOnClusterEvent).

The dispatcher type-asserts on this interface at client construction and silently no-ops if the configured collector does not implement it. By-hand MetricsCollector implementations stay source-compatible across this release and may opt in later by adding the method. Bundled collectors (e.g. contrib/metrics/vm) implement this interface directly.

Call discipline: the method is never called from the read/write hot path. The dispatcher counts drops with an atomic and reconciles the counter into the metric from its own delivery goroutine (after each delivered event) and once more at shutdown, so the metric can lag the internal count while the handler is blocked. Drops that occur after the shutdown reconciliation (post-Close emissions from unjoined background goroutines) are counted internally but not reflected in the metric.

type ClusterID 

type ClusterID string

ClusterID identifies a cluster in the dual-cluster setup. 

const (
	// ClusterA represents the first cluster (often called "primary").
	ClusterA ClusterID = "A"
	// ClusterB represents the second cluster (often called "secondary").
	ClusterB ClusterID = "B"
)

func (ClusterID) String 

func (c ClusterID) String() string

String returns the string representation of the ClusterID.

type ClusterNamer 

type ClusterNamer interface {
	// SetClusterNames sets the display names for clusters.
	//
	// This method is called by the client after construction to propagate
	// custom cluster names configured via WithClusterNames.
	//
	// Parameters:
	//   - names: The cluster names to use for metrics and logging
	SetClusterNames(names ClusterNames)
}

ClusterNamer is an optional interface for components that can use custom cluster names.

Components implementing this interface will have their cluster names set by the client after construction. This allows centralized configuration of cluster names at the client level, which are then propagated to metrics collectors, loggers, policies, etc.

Example implementation: 

type MyPolicy struct {
    clusterNames types.ClusterNames
}

func (p *MyPolicy) SetClusterNames(names types.ClusterNames) {
    p.clusterNames = names
}

type ClusterNames 

type ClusterNames struct {
	// A is the display name for ClusterA. Defaults to "A".
	A string

	// B is the display name for ClusterB. Defaults to "B".
	B string
}

ClusterNames holds custom display names for clusters.

These names are used in metrics labels and log messages instead of the default "A" and "B". Names must be:

  • 1-32 characters long
  • Prometheus-compatible: start with letter or underscore, contain only alphanumeric characters and underscores
  • Different from each other

Example names: "us_east", "us_west", "primary", "secondary", "dc1", "dc2"

func DefaultClusterNames 

func DefaultClusterNames() ClusterNames

DefaultClusterNames returns the default cluster names ("A" and "B").

func (ClusterNames) Name 

func (n ClusterNames) Name(cluster ClusterID) string

Name returns the display name for the given cluster ID.

func (ClusterNames) Validate 

func (n ClusterNames) Validate() error

Validate checks that the cluster names are valid for use in metrics.

Returns:

  • error: Validation error, or nil if valid

type Consistency 

type Consistency uint16

Consistency represents the Cassandra consistency level. 

const (
	Any         Consistency = 0x00
	One         Consistency = 0x01
	Two         Consistency = 0x02
	Three       Consistency = 0x03
	Quorum      Consistency = 0x04
	All         Consistency = 0x05
	LocalQuorum Consistency = 0x06
	EachQuorum  Consistency = 0x07
	Serial      Consistency = 0x08
	LocalSerial Consistency = 0x09
	LocalOne    Consistency = 0x0A
)

Common consistency levels matching gocql.

type DualClusterError 

type DualClusterError struct {
	// ErrorA is the error from cluster A.
	ErrorA error

	// ErrorB is the error from cluster B.
	ErrorB error
}

DualClusterError represents failures from both clusters.

func (*DualClusterError) Error 

func (e *DualClusterError) Error() string

Error implements the error interface.

If either ErrorA or ErrorB is nil, the corresponding part is omitted from the message rather than panicking.

func (*DualClusterError) Unwrap 

func (e *DualClusterError) Unwrap() []error

Unwrap returns the wrapped errors for errors.Is/As compatibility. This allows checking for specific error types in either cluster's error. Nil cluster errors are excluded from the returned slice.

type Logger 

type Logger interface {
	// Debug logs a message at DebugLevel.
	// The message includes any fields passed at the log site,
	// as well as any fields accumulated on the logger.
	Debug(msg string, keysAndValues ...any)

	// Info logs a message at InfoLevel.
	// The message includes any fields passed at the log site,
	// as well as any fields accumulated on the logger.
	Info(msg string, keysAndValues ...any)

	// Warn logs a message at WarnLevel.
	// The message includes any fields passed at the log site,
	// as well as any fields accumulated on the logger.
	Warn(msg string, keysAndValues ...any)

	// Error logs a message at ErrorLevel.
	// The message includes any fields passed at the log site,
	// as well as any fields accumulated on the logger.
	Error(msg string, keysAndValues ...any)

	// Fatal logs a message at FatalLevel and calls os.Exit(1).
	// The message includes any fields passed at the log site,
	// as well as any fields accumulated on the logger.
	//
	// The logger then calls os.Exit(1), even if logging at FatalLevel is disabled.
	Fatal(msg string, keysAndValues ...any)
}

Logger defines methods for structured logging.

Compatible with zap.SugaredLogger and other structured loggers. All methods accept key-value pairs for structured fields.

Example usage with zap: 

logger, _ := zap.NewProduction()
client, _ := helix.NewCQLClient(sessionA, sessionB,
    helix.WithLogger(logger.Sugar()),
)

Example usage with a custom implementation: 

type MyLogger struct{}
func (l *MyLogger) Debug(msg string, keysAndValues ...any) { ... }
func (l *MyLogger) Info(msg string, keysAndValues ...any) { ... }
// ... etc

type MetricsCollector 

type MetricsCollector interface {

	// IncReadTotal increments the total read operations counter.
	// Metric: [prefix]_read_total{cluster="..."}
	IncReadTotal(cluster ClusterID)

	// IncReadError increments the read error counter.
	// Metric: [prefix]_read_errors_total{cluster="..."}
	IncReadError(cluster ClusterID)

	// ObserveReadDuration records a read operation duration in seconds.
	// Metric: [prefix]_read_duration_seconds{cluster="..."}
	ObserveReadDuration(cluster ClusterID, seconds float64)

	// IncReadDivergence increments the counter when a FallbackRead finds data
	// on the alternative cluster after the selected cluster returned not-found.
	// The cluster parameter is the cluster that was missing the row, allowing
	// operators to correlate divergence with replay lag on a specific cluster.
	// Metric: [prefix]_read_divergence_total{cluster="..."}
	IncReadDivergence(cluster ClusterID)

	// IncWriteTotal increments the total write operations counter.
	// Metric: [prefix]_write_total{cluster="..."}
	IncWriteTotal(cluster ClusterID)

	// IncWriteError increments the write error counter.
	// Metric: [prefix]_write_errors_total{cluster="..."}
	IncWriteError(cluster ClusterID)

	// IncWriteAsync increments the counter when a write is dispatched asynchronously
	// to a degraded cluster via fire-and-forget (AdaptiveDualWrite only).
	// This is an operational state, not a cluster error.
	// Metric: [prefix]_write_async_total{cluster="..."}
	IncWriteAsync(cluster ClusterID)

	// IncWriteDropped increments the counter when a fire-and-forget write is dropped
	// because the concurrency limit (semaphore) is full (AdaptiveDualWrite only).
	// The replay system handles reconciliation for dropped writes.
	// Metric: [prefix]_write_dropped_total{cluster="..."}
	IncWriteDropped(cluster ClusterID)

	// ObserveWriteDuration records a write operation duration in seconds.
	// Metric: [prefix]_write_duration_seconds{cluster="..."}
	ObserveWriteDuration(cluster ClusterID, seconds float64)

	// IncFailoverTotal increments the failover event counter.
	// Called when a read operation fails over from one cluster to another.
	// Metric: [prefix]_failover_total{from="...",to="..."}
	IncFailoverTotal(fromCluster, toCluster ClusterID)

	// SetCircuitBreakerState sets the circuit breaker state gauge.
	// State values: 0=closed, 1=half-open (reserved; no policy in this
	// module emits it today — a breaker admitting a probe after its reset
	// timeout still reports 2 until the probe's outcome closes or re-opens
	// it), 2=open.
	// Metric: [prefix]_circuit_breaker_state{cluster="..."}
	SetCircuitBreakerState(cluster ClusterID, state int)

	// IncCircuitBreakerTrip increments the counter when circuit breaker trips to open.
	// Metric: [prefix]_circuit_breaker_trips_total{cluster="..."}
	IncCircuitBreakerTrip(cluster ClusterID)

	// IncReplayEnqueued increments the counter when a write is enqueued for replay.
	// Metric: [prefix]_replay_enqueued_total{cluster="..."}
	IncReplayEnqueued(cluster ClusterID)

	// IncReplaySuccess increments the counter when a replay operation succeeds.
	// Metric: [prefix]_replay_success_total{cluster="..."}
	IncReplaySuccess(cluster ClusterID)

	// IncReplayError increments the counter when a replay operation fails.
	// Metric: [prefix]_replay_errors_total{cluster="..."}
	IncReplayError(cluster ClusterID)

	// IncReplayDropped increments the counter when a replay payload cannot be enqueued.
	// This indicates potential data loss if the replay queue is full or unavailable.
	// Metric: [prefix]_replay_dropped_total{cluster="..."}
	IncReplayDropped(cluster ClusterID)

	// SetReplayQueueDepth sets the current replay queue depth gauge.
	// Metric: [prefix]_replay_queue_depth{cluster="..."}
	SetReplayQueueDepth(cluster ClusterID, depth int)

	// ObserveReplayDuration records a replay operation duration in seconds.
	// Metric: [prefix]_replay_duration_seconds{cluster="..."}
	ObserveReplayDuration(cluster ClusterID, seconds float64)

	// SetClusterDraining sets the drain status gauge for a cluster.
	// Value: 1 if draining, 0 if healthy.
	// Metric: [prefix]_cluster_draining{cluster="..."}
	SetClusterDraining(cluster ClusterID, draining bool)

	// IncDrainModeEntered increments the counter when a cluster enters drain mode.
	// Metric: [prefix]_drain_mode_entered_total{cluster="..."}
	IncDrainModeEntered(cluster ClusterID)

	// IncDrainModeExited increments the counter when a cluster exits drain mode.
	// Metric: [prefix]_drain_mode_exited_total{cluster="..."}
	IncDrainModeExited(cluster ClusterID)
}

MetricsCollector defines methods for collecting operational metrics.

All cluster-scoped methods accept a ClusterID parameter for labeling. Implementations should be thread-safe as methods may be called concurrently.

Example usage with VictoriaMetrics (via contrib/metrics/vm): 

import vmmetrics "github.com/arloliu/helix/contrib/metrics/vm"

collector := vmmetrics.New(vmmetrics.WithPrefix("myapp"))
client, _ := helix.NewCQLClient(sessionA, sessionB,
    helix.WithMetrics(collector),
)

// Expose metrics via HTTP
http.HandleFunc("/metrics", collector.Handler)

type MirrorMetrics added in v1.4.0 

type MirrorMetrics interface {
	// IncMirrorEnqueueSuccess increments when a captured write enters the
	// engine's bounded queue.
	IncMirrorEnqueueSuccess()

	// IncMirrorEnqueueDropped increments when a captured write is rejected
	// (engine disabled, stopped, or queue full).
	IncMirrorEnqueueDropped()

	// IncMirrorExecSuccess increments when the engine's execute returned
	// nil (write landed at target, or publish.Enqueue returned nil).
	IncMirrorExecSuccess()

	// IncMirrorExecError increments when the engine's execute returned an
	// error.
	IncMirrorExecError()

	// ObserveMirrorExecDuration records execute duration in seconds. Called
	// for both success and error paths.
	ObserveMirrorExecDuration(seconds float64)

	// SetMirrorQueueDepth updates the gauge with the current queue depth.
	SetMirrorQueueDepth(depth int)

	// SetMirrorEnabled updates the gauge: true = accepting new captures,
	// false = disabled or stopped.
	SetMirrorEnabled(enabled bool)
}

MirrorMetrics is an OPTIONAL interface that MetricsCollector implementations may satisfy to receive async-mirror counters and gauges.

Helix's mirror engine type-asserts on this interface (when configured via [WithMirror] or [WithMirrorPublisher]) and silently no-ops if the configured collector does not implement it. By-hand MetricsCollector implementations stay source-compatible across this release and may opt in by adding the methods below. Bundled collectors (e.g. contrib/metrics/vm) implement this interface directly.

Mirror metrics are NOT cluster-scoped. Mirroring targets a separate helix dual-cluster pair (or a publisher); the per-cluster routing happens inside the mirror destination's own write path and is recorded against that destination's own metrics namespace, not the source client's. The methods below report only the source client's view of the mirror engine.

Counter semantics:

  • IncMirrorEnqueueSuccess: a captured write was accepted into the engine's bounded queue.
  • IncMirrorEnqueueDropped: a captured write was rejected because the engine was disabled, stopped, or its queue was full. Use the engine's drop log / [mirror.WithOnDrop] callback to disambiguate reasons.
  • IncMirrorExecSuccess: the engine successfully dispatched a captured write. In target mode this means the mirror destination's Exec returned nil; in publisher mode it means publisher.Enqueue returned nil.
  • IncMirrorExecError: the engine's dispatch returned an error. In target mode pair with [WithMirrorReplayer] for durable retry; in publisher mode pair with [mirror.WithOnError] for observability.

Gauges:

  • SetMirrorQueueDepth: current queue depth, updated after enqueue and after each dequeue.
  • SetMirrorEnabled: 1 when accepting new enqueues, 0 when disabled. Updated on engine start, on Enable/Disable transitions, and on Stop.

type MirrorReplayMetrics added in v1.6.0 

type MirrorReplayMetrics interface {
	// IncMirrorReplayDropped increments when a failed mirror write cannot
	// be enqueued for mirror replay — potential mirror-target data loss.
	// Metric: [prefix]_mirror_replay_dropped_total
	IncMirrorReplayDropped()
}

MirrorReplayMetrics is an OPTIONAL interface that MetricsCollector implementations may satisfy to receive mirror replay-enqueue drop counts (see helix.WithMirrorReplayer).

It is separate from MirrorMetrics so existing implementations of that interface stay source-compatible. Helix's internal mirror error handler type-asserts on this interface and silently no-ops if the configured collector does not implement it; a caller-supplied mirror.WithOnError replaces that handler and with it this metric, exactly as it does EventMirrorReplayDropped.

Like the MirrorMetrics methods, the counter is not cluster-scoped: mirror payloads target a logical sink, not one of this client's clusters.

type OptionError added in v1.4.0 

type OptionError struct {
	// Component identifies where the option is used (for example,
	// "policy.AdaptiveDualWrite").
	Component string

	// Option is the option function name (for example,
	// "WithAdaptiveStrikeThreshold").
	Option string

	// Reason describes why the value is invalid.
	Reason string
}

OptionError reports an invalid functional-option value.

func AsOptionError added in v1.4.0 

func AsOptionError(err error) (*OptionError, bool)

AsOptionError extracts an OptionError from err using errors.As. Returns the error and true if found, or nil and false otherwise.

func (*OptionError) Error added in v1.4.0 

func (e *OptionError) Error() string

Error implements the error interface.

type PartialWriteError 

type PartialWriteError struct {
	// Acknowledged is the cluster that returned OK.
	Acknowledged ClusterID

	// Unacknowledged is the cluster that did not ack (reason in Cause).
	Unacknowledged ClusterID

	// Cause is the underlying error: timeout, sentinel, or driver error.
	Cause error
}

PartialWriteError indicates that a Strict() write was acknowledged by exactly one cluster. The other cluster did not respond OK before the deadline; the mutation MAY OR MAY NOT have applied there.

Callers MUST NOT assume the unacknowledged cluster is in a known state. Compensating retries on non-idempotent operations (counters, list/set append) can double-apply.

func AsPartialWriteError added in v1.4.0 

func AsPartialWriteError(err error) (*PartialWriteError, bool)

AsPartialWriteError extracts a *PartialWriteError from err using errors.As. Returns the error and true if found, or nil and false otherwise.

func (*PartialWriteError) Error 

func (e *PartialWriteError) Error() string

Error implements the error interface.

func (*PartialWriteError) Unwrap 

func (e *PartialWriteError) Unwrap() error

Unwrap returns the underlying cause for errors.Is/As compatibility.

type PriorityLevel 

type PriorityLevel int

PriorityLevel defines the priority for replay operations. 

const (
	// PriorityHigh indicates critical writes that must be replayed ASAP.
	PriorityHigh PriorityLevel = iota
	// PriorityLow indicates best-effort writes that can be delayed.
	PriorityLow
)

type RecoveryProbeMetrics added in v1.4.0 

type RecoveryProbeMetrics interface {
	// IncRecoveryProbeSuccess is called after a successful probe (err == nil)
	// against a degraded cluster.
	IncRecoveryProbeSuccess(cluster ClusterID)

	// IncRecoveryProbeFailure is called after a failing probe (err != nil)
	// against a degraded cluster.
	IncRecoveryProbeFailure(cluster ClusterID)
}

RecoveryProbeMetrics is an OPTIONAL interface that MetricsCollector implementations may satisfy to receive recovery-probe counters from the CQLClient's background probe goroutines (see helix.WithRecoveryProbe).

Helix's probe loop type-asserts on this interface and silently no-ops if the configured collector does not implement it. By-hand MetricsCollector implementations stay source-compatible and may opt in by adding the two methods below. Bundled collectors (e.g. contrib/metrics/vm) implement this interface directly.

Counter semantics:

  • IncRecoveryProbeSuccess: the probe returned nil; the cluster has been credited with one recovery point via [AdaptiveDualWrite.RecordProbeSuccess].
  • IncRecoveryProbeFailure: the probe returned a non-nil error; the cluster remains degraded and no recovery point was credited.

Probes run only against degraded clusters, so a healthy cluster produces neither counter.

type ReplayPayload 

type ReplayPayload struct {
	// TargetCluster identifies which cluster failed and needs replay.
	TargetCluster ClusterID

	// Query is the CQL statement to replay (for single-query writes).
	// Empty when IsBatch is true.
	Query string

	// Args are the bound values for the query.
	// Empty when IsBatch is true.
	Args []any

	// IsBatch indicates if this is a batch operation.
	IsBatch bool

	// BatchType is the type of batch (Logged, Unlogged, Counter).
	// Only used when IsBatch is true.
	BatchType BatchType

	// BatchStatements contains the statements in a batch.
	// Only used when IsBatch is true.
	BatchStatements []BatchStatement

	// Timestamp is the client-generated timestamp for idempotency.
	// This ensures replays don't overwrite newer data.
	Timestamp int64

	// Priority indicates the importance of this replay.
	Priority PriorityLevel
}

ReplayPayload contains the information needed to replay a failed write.

type SessionRefreshMetrics added in v1.2.1 

type SessionRefreshMetrics interface {
	// IncSessionRefreshAttempt is called by the auto-refresh detector
	// just before invoking the SessionRefresher.
	IncSessionRefreshAttempt(cluster ClusterID)
	// IncSessionRefreshSuccess is called after a successful auto-refresh
	// (refresher returned a non-nil session and the swap installed it).
	IncSessionRefreshSuccess(cluster ClusterID)
	// IncSessionRefreshError is called when the refresher returned an
	// error, returned a nil session, or the swap failed.
	IncSessionRefreshError(cluster ClusterID)
}

SessionRefreshMetrics is an OPTIONAL interface that MetricsCollector implementations may satisfy to receive auto-refresh counters from the CQLClient's auto-refresh detector (see helix.WithAutoRefresh).

Helix's auto-refresh path type-asserts on this interface and silently no-ops if the configured collector does not implement it. By-hand MetricsCollector implementations stay source-compatible across this release and may opt in to the new metrics later by adding the three methods. Bundled collectors (e.g. contrib/metrics/vm) implement this interface directly.

Counter semantics:

  • IncSessionRefreshAttempt: incremented every time the auto-refresh detector decides a cluster's session is permanently dead and is about to invoke the SessionRefresher. Stamped before the refresher call so monitoring sees attempts even if the refresher hangs.
  • IncSessionRefreshSuccess: incremented after a successful RefreshSession (refresher returned a non-nil session and the swap installed it). Old session has been closed at this point.
  • IncSessionRefreshError: incremented when the refresher returned an error, returned a nil session, or the swap failed (e.g., client was closed mid-operation).

On every attempt: Attempt is incremented exactly once; either Success XOR Error is incremented exactly once afterward. Sum of (Success + Error) always equals Attempt.

type StrictMetrics added in v1.4.0 

type StrictMetrics interface {
	// IncWriteSkipped is called when a cluster is skipped by a Strict() write
	// due to its degraded or draining state.
	IncWriteSkipped(cluster ClusterID)
}

StrictMetrics is an OPTIONAL interface that MetricsCollector implementations may satisfy to receive counters from [Strict] write paths.

Helix's strict write orchestrator type-asserts on this interface and silently no-ops if the configured collector does not implement it. By-hand MetricsCollector implementations stay source-compatible and may opt in by adding the method below. Bundled collectors (e.g. contrib/metrics/vm) implement this interface directly.

Counter semantics:

  • IncWriteSkipped: the cluster was not written to because it was degraded (AdaptiveDualWrite) or draining (drain mode). This is an operational state, not a cluster error — MetricsCollector.IncWriteError is NOT incremented for skipped writes.

Source Files

View all Source files

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.