cacheclient

package module
v1.11.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	// ModeAuto automatically detects cluster vs simple mode (default)
	ModeAuto ConnectionMode = "auto"
	// ModeSimple uses direct connections without topology service
	ModeSimple ConnectionMode = "simple"
	// ModeCluster uses topology service for smart routing
	ModeCluster ConnectionMode = "cluster"

	// DefaultRefreshInterval is the default topology refresh interval
	DefaultRefreshInterval = 30 * time.Second
	// MaxMessageSize is the maximum message size for gRPC
	MaxMessageSize = 128 * 1024 * 1024 // 128MB
	// TopologyDetectTimeout is the timeout for detecting cluster topology
	TopologyDetectTimeout = 2 * time.Second
	// DefaultBufferSize is the default buffer size for streaming operations
	DefaultBufferSize = 64 * 1024 // 64KB
	// ConnectionHealthCheckInterval is the interval for checking connection health
	ConnectionHealthCheckInterval = 30 * time.Second
	// ConnectionErrorWindow is the time window for tracking connection errors
	ConnectionErrorWindow = 30 * time.Second
	// DefaultConnectionPoolSize is the default number of connections per address
	DefaultConnectionPoolSize = 4
	// DefaultKeepaliveTime is the default keepalive time for client connections
	DefaultKeepaliveTime = 30 * time.Second
	// DefaultKeepaliveTimeout is the default keepalive timeout for client connections
	DefaultKeepaliveTimeout = 10 * time.Second

	// MaxPageLimit is the maximum number of keys to return in a single page
	MaxPageLimit = 1000
)

Variables

This section is empty.

Functions

func DefaultDialOptions 

func DefaultDialOptions() []grpc.DialOption

DefaultDialOptions returns the default gRPC dial options

Types

type CacheClient 

type CacheClient interface {
	// Basic operations
	Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error
	Get(ctx context.Context, key string) ([]byte, error)
	Delete(ctx context.Context, key string) error
	List(ctx context.Context, prefix string) ([]string, error)
	ListPage(ctx context.Context, prefix string, limit int, continuationToken string) (keys []string, nextToken string, hasMore bool, err error)
	ListPageWithValues(ctx context.Context, prefix string, limit int, continuationToken string) (entries []KeyValue, nextToken string, hasMore bool, err error)

	// Streaming operations
	PutStream(ctx context.Context, key string, r io.Reader, ttlSeconds int64) error
	GetStream(ctx context.Context, key string, w io.Writer) error

	// Range operations
	GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)
	GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

	// Lifecycle
	Close() error

	// Info
	GetMode() ConnectionMode
	GetConnectedNodes() []string
}

CacheClient is the common interface for both SimpleClient and ClusterClient

type Client 

type Client struct {
	CacheClient
	// contains filtered or unexported fields
}

Client is a wrapper that maintains backward compatibility It delegates to either SimpleClient or ClusterClient based on mode detection

func New 

func New(addrs ...string) (*Client, error)

New creates a new client with default configuration

func NewWithConfig 

func NewWithConfig(config *ClientConfig) (*Client, error)

NewWithConfig creates a new client with custom configuration

func (*Client) FetchClusterState 

func (c *Client) FetchClusterState() (*clusterpb.ClusterState, error)

FetchClusterState fetches the current cluster state (cluster mode only)

func (*Client) FetchTopology 

func (c *Client) FetchTopology() (*clusterpb.ClusterTopology, error)

FetchTopology fetches the current cluster topology (cluster mode only)

func (*Client) GetMode 

func (c *Client) GetMode() ConnectionMode

GetMode returns the actual connection mode being used

func (*Client) GetNodeIDForKey 

func (c *Client) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID that owns the given key (cluster mode only)

func (*Client) GetNodeInfoForKey 

func (c *Client) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address that owns the given key (cluster mode only)

func (*Client) GetTopologyEpoch 

func (c *Client) GetTopologyEpoch() uint64

GetTopologyEpoch returns the current topology epoch (cluster mode only)

func (*Client) HasRing 

func (c *Client) HasRing() bool

HasRing returns true if the consistent hash ring is initialized (cluster mode only)

func (*Client) IsClusterMode 

func (c *Client) IsClusterMode() bool

IsClusterMode returns true if the client is in cluster mode

type ClientConfig 

type ClientConfig struct {
	Addrs              []string          // One or more server addresses
	Mode               ConnectionMode    // Connection mode (default: "auto")
	RefreshInterval    time.Duration     // Topology refresh for cluster mode (default: 30s)
	ConnectionPoolSize int               // Number of connections per address (default: 4)
	DialOpts           []grpc.DialOption // Optional gRPC dial options
}

ClientConfig contains configuration for the unified Client

func (*ClientConfig) SetDefaults 

func (c *ClientConfig) SetDefaults()

SetDefaults sets default values for unspecified config fields

type ClusterClient 

type ClusterClient struct {
	*Operations // Embedded for shared operations
	// contains filtered or unexported fields
}

ClusterClient implements a cluster-aware cache client with topology support

func NewClusterClient 

func NewClusterClient(config *ClientConfig) (*ClusterClient, error)

NewClusterClient creates a new ClusterClient with the given configuration

func (*ClusterClient) Close 

func (c *ClusterClient) Close() error

Close closes all connections and stops background goroutines

func (*ClusterClient) Delete 

func (c *ClusterClient) Delete(ctx context.Context, key string) error

Delete removes a key with retry logic

func (*ClusterClient) FetchClusterState 

func (c *ClusterClient) FetchClusterState() (*clusterpb.ClusterState, error)

FetchClusterState fetches the current cluster state from any available node.

func (*ClusterClient) FetchTopology 

func (c *ClusterClient) FetchTopology() (*clusterpb.ClusterTopology, error)

FetchTopology fetches the current topology (exposed for testing)

func (*ClusterClient) Get 

func (c *ClusterClient) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves a value with retry logic

func (*ClusterClient) GetConnectedNodes 

func (c *ClusterClient) GetConnectedNodes() []string

GetConnectedNodes returns the addresses of all connected nodes

func (*ClusterClient) GetConnectionCount 

func (c *ClusterClient) GetConnectionCount() int

GetConnectionCount returns the number of active connections (exposed for testing)

func (*ClusterClient) GetMode 

func (c *ClusterClient) GetMode() ConnectionMode

GetMode returns the connection mode

func (*ClusterClient) GetNodeIDForKey 

func (c *ClusterClient) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID that owns the given key.

func (*ClusterClient) GetNodeInfoForKey 

func (c *ClusterClient) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address that owns the given key.

func (*ClusterClient) GetRange 

func (c *ClusterClient) GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)

GetRange retrieves a byte range with retry logic

func (*ClusterClient) GetRangeStream 

func (c *ClusterClient) GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

GetRangeStream streams a byte range with retry logic

func (*ClusterClient) GetStream 

func (c *ClusterClient) GetStream(ctx context.Context, key string, w io.Writer) error

GetStream streams a value with retry logic

func (*ClusterClient) GetTopologyEpoch 

func (c *ClusterClient) GetTopologyEpoch() uint64

GetTopologyEpoch returns the current topology epoch

func (*ClusterClient) HasRing 

func (c *ClusterClient) HasRing() bool

HasRing returns true if the token ring is initialized and has tokens

func (*ClusterClient) Put 

func (c *ClusterClient) Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error

Put stores a value with retry logic for routing errors

func (*ClusterClient) RoundRobinRoute 

func (c *ClusterClient) RoundRobinRoute() (*connection, error)

RoundRobinRoute selects a connection using round-robin (for List operation) Implements Router interface

func (*ClusterClient) Route 

func (c *ClusterClient) Route(key string) (*connection, error)

Route determines which connection to use for a given key Implements Router interface Optimized to minimize lock contention using cached routing decisions

func (*ClusterClient) UpdateTopology 

func (c *ClusterClient) UpdateTopology(topology *clusterpb.ClusterTopology) error

UpdateTopology manually updates the topology (exposed for testing)

type ConnectionMode 

type ConnectionMode string

ConnectionMode defines how the client connects to servers

type EpochGetter 

type EpochGetter func() uint64

EpochGetter is a function that returns the current client epoch

type EpochMismatchHandler 

type EpochMismatchHandler func(clientEpoch, serverEpoch uint64)

EpochMismatchHandler is called when server epoch differs from client epoch

type KeyValue 

type KeyValue struct {
	Key   string
	Value []byte
	// ValueLength is the size of the value in bytes, set even when the value was
	// omitted for exceeding the List-with-values per-value size cap.
	ValueLength int64
	// ValueOmitted is true when the value was omitted for exceeding that cap;
	// Value is nil in that case.
	ValueOmitted bool
}

KeyValue holds a key and its associated value bytes.

type MemoryCache 

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

MemoryCache implements CacheClient with in-memory storage. Useful for testing without a real cache server.

func NewMemoryCache 

func NewMemoryCache() *MemoryCache

NewMemoryCache creates a new in-memory cache.

func (*MemoryCache) Close 

func (m *MemoryCache) Close() error

Close clears the cache. This is a no-op for cleanup purposes.

func (*MemoryCache) Delete 

func (m *MemoryCache) Delete(ctx context.Context, key string) error

Delete removes a key from the cache.

func (*MemoryCache) Get 

func (m *MemoryCache) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves data by key.

func (*MemoryCache) GetConnectedNodes 

func (m *MemoryCache) GetConnectedNodes() []string

GetConnectedNodes returns a single "memory" node identifier.

func (*MemoryCache) GetMode 

func (m *MemoryCache) GetMode() ConnectionMode

GetMode returns ModeSimple since MemoryCache is a single-node implementation.

func (*MemoryCache) GetRange 

func (m *MemoryCache) GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)

GetRange retrieves a byte range from the cached data.

func (*MemoryCache) GetRangeStream 

func (m *MemoryCache) GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

GetRangeStream retrieves a byte range and writes it to the writer.

func (*MemoryCache) GetStream 

func (m *MemoryCache) GetStream(ctx context.Context, key string, w io.Writer) error

GetStream retrieves data and writes it to the writer.

func (*MemoryCache) List 

func (m *MemoryCache) List(ctx context.Context, prefix string) ([]string, error)

List returns all keys matching the prefix.

func (*MemoryCache) ListPage 

func (m *MemoryCache) ListPage(ctx context.Context, prefix string, limit int, continuationToken string) (keys []string, nextToken string, hasMore bool, err error)

ListPage returns a paginated list of keys matching the prefix.

func (*MemoryCache) ListPageWithValues 

func (m *MemoryCache) ListPageWithValues(ctx context.Context, prefix string, limit int, continuationToken string) (entries []KeyValue, nextToken string, hasMore bool, err error)

ListPageWithValues returns a paginated list of key-value pairs matching the prefix.

func (*MemoryCache) Put 

func (m *MemoryCache) Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error

Put stores data with an optional TTL (0 means no expiration).

func (*MemoryCache) PutStream 

func (m *MemoryCache) PutStream(ctx context.Context, key string, r io.Reader, ttlSeconds int64) error

PutStream reads all data from the reader and stores it.

type Operations 

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

Operations provides shared implementation of cache operations

func NewOperations 

func NewOperations(router Router) *Operations

NewOperations creates a new Operations instance

func (*Operations) Delete 

func (o *Operations) Delete(ctx context.Context, key string) error

Delete removes a key from the cache

func (*Operations) Get 

func (o *Operations) Get(ctx context.Context, key string) ([]byte, error)

Get retrieves a value from the cache

func (*Operations) GetRange 

func (o *Operations) GetRange(ctx context.Context, key string, start, end int64) ([]byte, error)

GetRange retrieves a byte range from the cache

func (*Operations) GetRangeStream 

func (o *Operations) GetRangeStream(ctx context.Context, key string, start, end int64, w io.Writer) error

GetRangeStream streams a byte range from the cache

func (*Operations) GetStream 

func (o *Operations) GetStream(ctx context.Context, key string, w io.Writer) error

GetStream streams a value from the cache

func (*Operations) List 

func (o *Operations) List(ctx context.Context, prefix string) ([]string, error)

List lists keys with optional prefix Returns all keys matching the prefix (automatically handles pagination)

func (*Operations) ListPage 

func (o *Operations) ListPage(ctx context.Context, prefix string, limit int, continuationToken string) ([]string, string, bool, error)

ListPage returns a single page of keys with pagination support Returns: (keys, continuationToken, hasMore, error)

func (*Operations) ListPageWithValues 

func (o *Operations) ListPageWithValues(ctx context.Context, prefix string, limit int, continuationToken string) ([]KeyValue, string, bool, error)

ListPageWithValues returns a single page of key-value pairs with pagination support. Returns: (entries, continuationToken, hasMore, error)

func (*Operations) Put 

func (o *Operations) Put(ctx context.Context, key string, data []byte, ttlSeconds int64) error

Put stores a value in the cache

func (*Operations) PutStream 

func (o *Operations) PutStream(ctx context.Context, key string, r io.Reader, ttlSeconds int64) error

PutStream streams data to the cache

type Router 

type Router interface {
	Route(key string) (*connection, error)
	RoundRobinRoute() (*connection, error)
}

Router is an interface for routing keys to connections

type SimpleClient 

type SimpleClient struct {
	*Operations // Embedded for shared operations
	// contains filtered or unexported fields
}

SimpleClient implements a simple round-robin cache client

func NewSimpleClient 

func NewSimpleClient(config *ClientConfig) (*SimpleClient, error)

NewSimpleClient creates a new SimpleClient with the given configuration

func (*SimpleClient) Close 

func (c *SimpleClient) Close() error

Close closes all connections

func (*SimpleClient) GetConnectedNodes 

func (c *SimpleClient) GetConnectedNodes() []string

GetConnectedNodes returns the addresses of all connected nodes

func (*SimpleClient) GetMode 

func (c *SimpleClient) GetMode() ConnectionMode

GetMode returns the connection mode

func (*SimpleClient) RoundRobinRoute 

func (c *SimpleClient) RoundRobinRoute() (*connection, error)

RoundRobinRoute selects a connection using round-robin (for operations without keys) Implements Router interface

func (*SimpleClient) Route 

func (c *SimpleClient) Route(key string) (*connection, error)

Route selects a connection using hash-based routing for better key locality Implements Router interface

type TokenRing 

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

TokenRing implements token-based consistent hashing matching the server's dskit ring. It uses FNV-1a 32-bit hash and binary search for O(log n) lookups. The ring is structured as a sorted array of tokens. Each token is owned by a specific node.

Thread Safety: Updates use atomic pointer swapping, so reads are completely lock-free.

func NewTokenRing 

func NewTokenRing() *TokenRing

NewTokenRing creates a new empty token ring

func (*TokenRing) GetNodeAddresses 

func (r *TokenRing) GetNodeAddresses() map[string]string

GetNodeAddresses returns a copy of all node addresses

func (*TokenRing) GetNodeForKey 

func (r *TokenRing) GetNodeForKey(key string) (string, error)

GetNodeForKey returns the node address that owns the given key. Uses FNV-1a 32-bit hash (same as server) + binary search.

The algorithm: 1. Hash the key to get a 32-bit token 2. Binary search for the first token in the ring >= our hash 3. If we're past the last token, wrap around to the first (ring semantics) 4. Return the address of the node that owns that token

This method is lock-free - it reads from an atomically swapped pointer.

func (*TokenRing) GetNodeIDForKey 

func (r *TokenRing) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID that owns the given key. This is useful for debugging and testing. This method is lock-free.

func (*TokenRing) GetNodeInfoForKey 

func (r *TokenRing) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address that owns the given key. This method is lock-free.

func (*TokenRing) IsEmpty 

func (r *TokenRing) IsEmpty() bool

IsEmpty returns true if the ring has no tokens

func (*TokenRing) NodeCount 

func (r *TokenRing) NodeCount() int

NodeCount returns the number of unique nodes in the ring

func (*TokenRing) TokenCount 

func (r *TokenRing) TokenCount() int

TokenCount returns the total number of tokens in the ring

func (*TokenRing) Update 

func (r *TokenRing) Update(nodeTokens map[string][]uint32, nodeAddresses map[string]string)

Update rebuilds the ring with new token assignments. nodeTokens is a map of nodeID -> list of tokens owned by that node. nodeAddresses is a map of nodeID -> listen address for that node.

This method uses atomic pointer swapping, so concurrent reads see either the old state or the new state atomically - never a partially updated state. This eliminates contention between topology updates and client requests.

type TopologyManager 

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

TopologyManager manages cluster topology for ClusterClient. It uses a token-based ring that matches the server's dskit ring implementation, ensuring consistent key routing between client and server.

func NewTopologyManager 

func NewTopologyManager(seedAddrs []string, refreshInterval time.Duration, dialOpts []grpc.DialOption) (*TopologyManager, error)

NewTopologyManager creates a new topology manager

func (*TopologyManager) FetchTopology 

func (tm *TopologyManager) FetchTopology(ctx context.Context) (*clusterpb.ClusterTopology, error)

FetchTopology fetches the cluster topology from available nodes

func (*TopologyManager) GetNodeAddresses 

func (tm *TopologyManager) GetNodeAddresses() map[string]string

GetNodeAddresses returns all node addresses

func (*TopologyManager) GetNodeForKey 

func (tm *TopologyManager) GetNodeForKey(key string) (string, error)

GetNodeForKey returns the node address for a given key. Uses FNV-1a 32-bit hash + binary search (same as server).

func (*TopologyManager) GetNodeIDForKey 

func (tm *TopologyManager) GetNodeIDForKey(key string) (string, error)

GetNodeIDForKey returns the node ID for a given key. Useful for debugging and testing.

func (*TopologyManager) GetNodeInfoForKey 

func (tm *TopologyManager) GetNodeInfoForKey(key string) (nodeID, address string, err error)

GetNodeInfoForKey returns both the node ID and address for a given key.

func (*TopologyManager) GetRing 

func (tm *TopologyManager) GetRing() *TokenRing

GetRing returns the underlying token ring

func (*TopologyManager) GetTopologyEpoch 

func (tm *TopologyManager) GetTopologyEpoch() uint64

GetTopologyEpoch returns the current topology epoch. Uses atomic load for lock-free access.

func (*TopologyManager) RefreshTopology 

func (tm *TopologyManager) RefreshTopology(ctx context.Context) (bool, error)

RefreshTopology refreshes the topology

func (*TopologyManager) TopologyRefreshLoop 

func (tm *TopologyManager) TopologyRefreshLoop(ctx context.Context, updateFn func())

TopologyRefreshLoop periodically refreshes the cluster topology

func (*TopologyManager) UpdateTopology 

func (tm *TopologyManager) UpdateTopology(topology *clusterpb.ClusterTopology) (map[string]bool, bool)

UpdateTopology updates the internal state based on new topology. With content-addressable epochs, same epoch = same ring state, so we use equality check (not >=) to detect changes.

Source Files

View all Source files

Directories

Path Synopsis
cmd
ycsb

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.