embedded

package module
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: 17 Imported by: 0

Documentation

Overview

Package embedded provides an embedded ocache client for use in other services. This allows services like TAG to embed ocache and get full cluster-aware caching with metrics, routing, and cluster-wide list operations.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Client

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

Client provides embedded cache access with cluster routing. It implements the cacheclient.CacheClient interface.

func New

func New(cfg *Config) (*Client, error)

New creates a new embedded cache client. The client provides direct access to local storage with automatic routing to remote nodes when running in cluster mode.

func (*Client) Close

func (c *Client) Close() error

Close shuts down the embedded client and releases all resources.

func (*Client) Coordinator

func (c *Client) Coordinator() *coordinator.Coordinator

Coordinator returns the underlying coordinator. Returns nil if clustering is not enabled.

func (*Client) Delete

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

Delete removes a key from the cache.

func (*Client) Get

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

Get retrieves data for the given key.

func (*Client) GetConnectedNodes

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

GetConnectedNodes returns the list of connected nodes.

func (*Client) GetGRPCServer

func (c *Client) GetGRPCServer() *grpc.Server

GetGRPCServer returns the gRPC server. Returns nil if StartGRPCServer has not been called.

func (*Client) GetMode

func (c *Client) GetMode() cacheclient.ConnectionMode

GetMode returns the connection mode.

func (*Client) GetRange

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

GetRange retrieves a byte range for the given key.

func (*Client) GetRangeStream

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

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

func (*Client) GetStream

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

GetStream retrieves data and writes it to the provided writer.

func (*Client) IsLocal added in v1.9.0

func (c *Client) IsLocal(key string) bool

IsLocal reports whether key is owned by this node — i.e. a read for it is served from local storage rather than routed to a peer over gRPC. In single-node (non-cluster) mode every key is local. Callers use this to observe cross-node serve ratios without reaching into the coordinator.

func (*Client) IsReady

func (c *Client) IsReady() bool

IsReady returns true if the client is ready to serve requests.

func (*Client) List

func (c *Client) List(ctx context.Context, prefix string) ([]string, error)

List returns all keys matching the given prefix across the entire cluster.

func (*Client) ListPage

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

ListPage returns a page of keys with pagination support.

func (*Client) ListPageWithValues

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

ListPageWithValues returns a page of key-value pairs with pagination support.

func (*Client) Operations

func (c *Client) Operations() *operations.Operations

Operations returns the underlying operations layer. This provides direct access to the routing logic for advanced use cases.

func (*Client) Put

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

Put stores data for the given key.

func (*Client) PutStream

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

PutStream stores data from a reader for the given key.

func (*Client) Service

func (c *Client) Service() *service.CacheService

Service returns the gRPC service. This is useful for registering additional handlers or middleware.

func (*Client) StartGRPCServer

func (c *Client) StartGRPCServer() error

StartGRPCServer starts the gRPC server for handling remote requests. This must be called in cluster mode to allow other nodes to route requests here.

func (*Client) Storage

func (c *Client) Storage() *stor.Storage

Storage returns the underlying storage layer. This provides direct access to local storage for advanced use cases.

func (*Client) WaitReady

func (c *Client) WaitReady(ctx context.Context) error

WaitReady waits for the client to be ready to serve requests. In cluster mode, this waits for the coordinator to reach ACTIVE state.

type Config

type Config struct {
	// DiskPath is the path to the cache data directory (required)
	DiskPath string

	// TTL is the default time-to-live for cache entries (required)
	TTL time.Duration

	// MaxDiskUsage is the maximum disk usage in bytes (0 = unlimited)
	MaxDiskUsage int64

	// InlineThreshold is the size threshold for inline vs file storage (default: 64KB)
	// Objects smaller than this are stored in RocksDB, larger ones as files.
	InlineThreshold int

	// NodeID is the unique identifier for this node in cluster mode
	NodeID string

	// ClusterAddr is the address for cluster membership (gossip) protocol
	// Example: ":7000"
	ClusterAddr string

	// GRPCAddr is the address for the gRPC server to listen on
	// Example: ":9000"
	GRPCAddr string

	// AdvertiseAddr is the address advertised to other nodes for gRPC connections
	// Example: "node1.cluster:9000"
	AdvertiseAddr string

	// SeedNodes is a list of seed nodes for cluster discovery
	// Example: []string{"node1:7000", "node2:7000"}
	SeedNodes []string

	// RequestLogging enables logging of gRPC requests (default: false)
	RequestLogging bool

	// GRPCServerOptions are additional gRPC server options (e.g., auth interceptors).
	// These are appended after the default options (message size limits, epoch interceptors).
	GRPCServerOptions []grpc.ServerOption

	// GRPCDialOptions are additional gRPC dial options for inter-node connections (e.g., auth interceptors).
	// These are appended after the default options (transport credentials, keepalive, message size).
	GRPCDialOptions []grpc.DialOption

	// Storage provides advanced tuning for the local storage layer (compaction,
	// segment sizing, RocksDB block cache, fd cache, TTL cleanup interval, etc.).
	// When set, its fields are applied as the base; the top-level DiskPath, TTL,
	// MaxDiskUsage, and InlineThreshold fields above take precedence when non-zero.
	// Unset fields fall through to storage-layer defaults.
	Storage *stor.StorageConfig

	// Lifecycler provides advanced cluster ring tuning (heartbeat period/timeout,
	// replication factor, num tokens, observe period, tokens file path, etc.).
	// Ignored when cluster mode is not enabled.
	Lifecycler *ring.LifecyclerConfig

	// Router provides advanced inter-node routing tuning (connection timeout,
	// retry/backoff, keepalive, circuit breaker, message sizes).
	// When nil, coordinator defaults are used.
	// Ignored when cluster mode is not enabled.
	Router *coordinator.RouterConfig

	// Registerer is the Prometheus registerer used for cluster metrics.
	// When nil, prometheus.DefaultRegisterer is used. Provide a dedicated
	// registerer to avoid duplicate-registration panics when embedding
	// multiple instances or integrating with a custom metrics registry.
	Registerer prometheus.Registerer
}

Config contains configuration for the embedded cache client.

func (*Config) IsClusterMode

func (c *Config) IsClusterMode() bool

IsClusterMode returns true if cluster configuration is provided.

func (*Config) SetDefaults

func (c *Config) SetDefaults()

SetDefaults sets default values for unspecified config fields. Storage-layer fields (InlineThreshold, segment sizes, etc.) are defaulted by storage.NewStorageWithConfig and are intentionally not touched here so that values set via Config.Storage flow through unchanged.

func (*Config) Validate

func (c *Config) Validate() error

Validate checks that required configuration is provided.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL