ring

package
v1.9.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	// DefaultHeartbeatPeriod is the default interval for heartbeat updates
	DefaultHeartbeatPeriod = 500 * time.Millisecond

	// MinHeartbeatTimeout is the default timeout before marking a node unhealthy.
	// Set long enough to handle rolling updates (60s allows for typical pod restarts).
	MinHeartbeatTimeout = 60 * time.Second

	// DefaultNumTokens is the default number of tokens per instance.
	// 512 provides good distribution across the ring.
	DefaultNumTokens = 512

	// DefaultReplicationFactor is the default replication factor (1 = no replication)
	DefaultReplicationFactor = 1

	// RingName is the name used for the ring in metrics and KV store
	RingName = "ocache"

	// RingKey is the key used to store the ring in the KV store
	RingKey = "ring"
)
View Source 
const (
	// RecommendedAnnounceTimeout is the recommended minimum timeout for AnnounceLeaving.
	// CAS retries can take significant time when there's contention with background
	// heartbeat operations, especially under race detection or slow CI environments.
	RecommendedAnnounceTimeout = 10 * time.Second

	// DefaultGossipPropagationDelay is the time to wait for gossip to propagate to other nodes.
	// Memberlist gossip typically propagates within 200-500ms.
	DefaultGossipPropagationDelay = 500 * time.Millisecond
)

Variables

This section is empty.

Functions

func ComputeRingEpoch 

func ComputeRingEpoch(ringDesc *ring.Desc) uint64

ComputeRingEpoch creates a deterministic hash of ring state. This function is exported for testing purposes.

The hash includes: - Node IDs (sorted for determinism) - Node states (to detect JOINING→ACTIVE transitions) - Token counts (to detect if tokens were modified)

We intentionally do NOT include full tokens because: - Tokens are assigned once and persisted (dskit's token persistence) - Hashing 512 tokens × N nodes would be expensive - Token count is sufficient to detect "has tokens been modified"

func GetEpochFromRing 

func GetEpochFromRing(rm *RingManager) uint64

GetEpochFromRing is a convenience function to safely get epoch from a potentially nil RingManager.

func SetupLifecyclerConfig 

func SetupLifecyclerConfig(nodeID, listenAddr, diskPath string, baseConfig *LifecyclerConfig) error

SetupLifecyclerConfig creates a LifecyclerConfig from coordinator parameters. This is the preferred way to create a LifecyclerConfig as it ensures all required fields are set. Parameters:

  • nodeID: unique identifier for this instance
  • listenAddr: the address this instance listens on for client requests (e.g., ":9001" or "0.0.0.0:9001")
  • diskPath: the path where the ring tokens will be persisted
  • baseConfig: the base configuration to use

Types

type Config 

type Config struct {
	// KVStore configures the key-value store backend (memberlist)
	KVStore kv.Config `yaml:"kvstore"`

	// HeartbeatPeriod is the interval at which this instance sends heartbeats
	HeartbeatPeriod time.Duration `yaml:"heartbeat_period"`

	// HeartbeatTimeout is the time after which an instance is considered unhealthy
	// if no heartbeat is received. Should be long enough for rolling updates.
	HeartbeatTimeout time.Duration `yaml:"heartbeat_timeout"`

	// ReplicationFactor is the number of replicas for each key (1 = no replication)
	ReplicationFactor int `yaml:"replication_factor"`
}

Config holds configuration for the dskit ring integration

func (*Config) ApplyDefaults 

func (c *Config) ApplyDefaults()

ApplyDefaults applies default values for any unset or invalid configuration fields. This should be called before using the configuration.

func (*Config) ToRingConfig 

func (c *Config) ToRingConfig() ring.Config

ToRingConfig converts to dskit ring.Config

type Epoch 

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

Epoch tracks the ring version using content-addressable hashing. Nodes with identical ring views will have identical epochs, enabling reliable cross-node comparisons and eliminating unnecessary topology refreshes.

The epoch is computed as a deterministic hash of the ring state: - Sorted node IDs (for determinism) - Node states (JOINING, ACTIVE, LEAVING, etc.) - Token counts (not full tokens - too expensive, and tokens are immutable)

This is an O(1) atomic load operation for reading - safe for hot paths. Computing the epoch is O(N) where N = number of nodes, but this only happens during heartbeat callbacks when ring state changes.

func NewEpoch 

func NewEpoch() *Epoch

NewEpoch creates a new Epoch tracker initialized to 0.

func (*Epoch) Get 

func (e *Epoch) Get() uint64

Get returns the current epoch value. This is O(1) - just an atomic load, safe to call from hot paths.

func (*Epoch) Set 

func (e *Epoch) Set(ringDesc *ring.Desc) uint64

Set computes epoch from ring membership state and stores it. Nodes with identical ring views will compute identical epochs.

This is O(N) where N = number of nodes, but is only called during heartbeat callbacks when ring state may have changed.

Returns the new epoch value.

type LifecyclerConfig 

type LifecyclerConfig struct {
	// RingConfig is the shared ring configuration
	RingConfig Config `yaml:"ring"`

	// InstanceID is the unique identifier for this instance
	InstanceID string `yaml:"instance_id"`

	// InstanceAddr is the address other instances use to reach this one (for client requests)
	InstanceAddr string `yaml:"instance_addr"`

	// InstancePort is the port this instance listens on for client requests
	InstancePort int `yaml:"instance_port"`

	// NumTokens is the number of tokens this instance claims on the ring
	NumTokens int `yaml:"num_tokens"`

	// DiskPath is the base directory for persistent storage (e.g., the -disk flag value).
	// Used to derive TokensFilePath if not explicitly set.
	DiskPath string `yaml:"disk_path"`

	// TokensFilePath is the path to persist tokens for stable ownership across restarts.
	// If empty, defaults to <DiskPath>/coordinator/ring-tokens.
	// Token persistence is essential for stable ownership across restarts.
	TokensFilePath string `yaml:"tokens_file_path"`

	// ObservePeriod is the time to wait after joining before marking as ACTIVE.
	// Used to observe the ring state before fully joining.
	ObservePeriod time.Duration `yaml:"observe_period"`

	// MinReadyDuration is the minimum time this instance must be in ACTIVE state
	// before the /ready endpoint returns ready.
	MinReadyDuration time.Duration `yaml:"min_ready_duration"`

	// UnregisterOnShutdown controls whether this instance is removed from the ring on shutdown.
	// If true (default), the instance transitions to LEAVING then leaves the ring.
	// If false, the instance stays in the ring and will be detected as unhealthy via heartbeat timeout.
	UnregisterOnShutdown bool `yaml:"unregister_on_shutdown"`
}

LifecyclerConfig holds configuration for an individual instance's lifecycle

func (*LifecyclerConfig) ApplyDefaults 

func (c *LifecyclerConfig) ApplyDefaults()

ApplyDefaults applies default values for any unset or invalid configuration fields. This should be called before using the configuration. Note: This method mutates the config to apply defaults.

func (*LifecyclerConfig) ToBasicLifecyclerConfig 

func (c *LifecyclerConfig) ToBasicLifecyclerConfig() ring.BasicLifecyclerConfig

ToBasicLifecyclerConfig converts to dskit ring.BasicLifecyclerConfig

type NodeInfo 

type NodeInfo struct {
	ID            string
	Address       string // Cluster communication address (for gossip/heartbeats)
	ListenAddress string // Service listen address for client requests (Put/Get/Delete)
	Status        NodeStatus
	JoinedAt      time.Time
	Weight        float64
	Available     bool
}

NodeInfo stores information about a node in the cluster. This maintains API compatibility with the existing coordinator package.

type NodeStatus 

type NodeStatus int

NodeStatus represents the status of a node in the cluster 

const (
	NodeStatusActive NodeStatus = iota
	NodeStatusJoining
	NodeStatusLeaving
	NodeStatusDown
)

func (NodeStatus) String 

func (s NodeStatus) String() string

type RingManager 

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

RingManager wraps dskit's ring and lifecycler to provide the same interface as the existing coordinator.Ring but with production-grade features: - Gossip-based membership via memberlist - Token persistence for stable ownership - Proper lifecycle state machine - Epoch tracking via heartbeat callbacks

func NewRingManager 

func NewRingManager(cfg LifecyclerConfig, kvClient kv.Client, logger log.Logger, reg prometheus.Registerer) (*RingManager, error)

NewRingManager creates a new RingManager with dskit ring integration

func (*RingManager) AnnounceLeaving 

func (rm *RingManager) AnnounceLeaving(ctx context.Context) error

AnnounceLeaving transitions this node to LEAVING state and waits briefly for gossip propagation. This should be called BEFORE Stop() to ensure other nodes are notified of the departure. The caller should provide a context with an appropriate timeout (recommend at least 10 seconds to allow CAS retries to succeed under contention).

func (*RingManager) GetActiveNodes 

func (rm *RingManager) GetActiveNodes() []*NodeInfo

GetActiveNodes returns all active nodes in the cluster. Returns an empty slice (not nil) if no active nodes are available.

func (*RingManager) GetAllNodes 

func (rm *RingManager) GetAllNodes() []*NodeInfo

GetAllNodes returns all nodes in the cluster. Returns an empty slice (not nil) if no nodes are available to ensure consistent behavior.

func (*RingManager) GetAvailableNodes 

func (rm *RingManager) GetAvailableNodes() []*NodeInfo

GetAvailableNodes returns nodes that are available for routing

func (*RingManager) GetEpoch 

func (rm *RingManager) GetEpoch() uint64

GetEpoch returns the current ring epoch. The epoch is a monotonically increasing counter that increments whenever ring membership changes (nodes join, leave, or change state). Clients can use this to detect stale topology information.

func (*RingManager) GetNode 

func (rm *RingManager) GetNode(key string) (*NodeInfo, error)

GetNode returns the available node that owns the key.

func (*RingManager) GetNodeStatus 

func (rm *RingManager) GetNodeStatus(id string) (NodeStatus, error)

GetNodeStatus returns the status of a specific node

func (*RingManager) GetNodeTokens 

func (rm *RingManager) GetNodeTokens() map[string][]uint32

GetNodeTokens returns token assignments for all active nodes in the ring. Used by GetClusterTopology to provide clients with token data for routing. Returns a map of nodeID -> sorted list of tokens.

Important: This only returns tokens for ACTIVE nodes because: 1. JOINING/PENDING nodes are not yet ready to serve requests 2. LEAVING nodes are transitioning out and should not receive new requests 3. Temporarily unhealthy nodes (missed heartbeats) are filtered by GetAllHealthy

func (*RingManager) GetPrimaryNode 

func (rm *RingManager) GetPrimaryNode(key string) (*NodeInfo, error)

GetPrimaryNode returns the primary owner regardless of availability. This includes nodes in JOINING, PENDING, and LEAVING states, but NOT LEFT (nodes that have already departed the cluster).

func (*RingManager) GetState 

func (rm *RingManager) GetState() ring.InstanceState

GetState returns the current lifecycler state

func (*RingManager) HealthyInstancesCount 

func (rm *RingManager) HealthyInstancesCount() int

HealthyInstancesCount returns the count of healthy instances

func (*RingManager) IsLocal 

func (rm *RingManager) IsLocal(key string) bool

IsLocal checks if the local node is the owner of the key.

func (*RingManager) IsNodeAvailable 

func (rm *RingManager) IsNodeAvailable(nodeID string) bool

IsNodeAvailable checks if a specific node is available

func (*RingManager) IsReady 

func (rm *RingManager) IsReady() bool

IsReady returns true if this instance is ready to serve requests

func (*RingManager) MarkReady 

func (rm *RingManager) MarkReady()

MarkReady signals that this node can serve requests, releasing the gate that holds it in JOINING and allowing it to advertise ACTIVE. It is called once storage has booted and the gRPC server is listening. Idempotent, and safe to call before or after the lifecycler has assigned tokens (it just closes the gate; the activation goroutine proceeds whenever it observes it closed).

func (*RingManager) Start 

func (rm *RingManager) Start(ctx context.Context) error

Start starts the ring manager and its subservices

func (*RingManager) Stop 

func (rm *RingManager) Stop(ctx context.Context) error

Stop gracefully stops the ring manager

func (*RingManager) WaitReady 

func (rm *RingManager) WaitReady(ctx context.Context) error

WaitReady blocks until the instance reaches ACTIVE state or the context is cancelled. This is useful for callers that need to wait for the ring to be ready before proceeding. Returns nil if ACTIVE state is reached, or context error if cancelled/timed out.

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.