cacheutil

Documentation

Index

• Variables
• func NewMemcachedClient(logger log.Logger, name string, conf []byte, reg prometheus.Registerer) (*memcachedClient, error)
• func NewMemcachedClientWithConfig(logger log.Logger, name string, config MemcachedClientConfig, ...) (*memcachedClient, error)
• type AddressProvider
• type AsyncOperationProcessor
  • func NewAsyncOperationProcessor(bufferSize, concurrency int) *AsyncOperationProcessor
  • func (p *AsyncOperationProcessor) EnqueueAsync(op func()) error
  • func (p *AsyncOperationProcessor) Stop()
• type CircuitBreaker
• type CircuitBreakerConfig
• type MemcachedClient
• type MemcachedClientConfig
• type MemcachedJumpHashSelector
  • func (s *MemcachedJumpHashSelector) Each(f func(net.Addr) error) error
  • func (s *MemcachedJumpHashSelector) PickServer(key string) (net.Addr, error)
  • func (s *MemcachedJumpHashSelector) SetServers(servers ...string) error
• type RedisClient
  • func NewRedisClient(logger log.Logger, name string, conf []byte, reg prometheus.Registerer) (*RedisClient, error)
  • func NewRedisClientWithConfig(logger log.Logger, name string, config RedisClientConfig, ...) (*RedisClient, error)
  • func (c *RedisClient) GetMulti(ctx context.Context, keys []string) map[string][]byte
  • func (c *RedisClient) SetAsync(key string, value []byte, ttl time.Duration) error
  • func (c *RedisClient) SetMulti(data map[string][]byte, ttl time.Duration)
  • func (c *RedisClient) Stop()
• type RedisClientConfig
• type RemoteCacheClient
• type TLSConfig

Variables

var (
	// DefaultRedisClientConfig is default redis config.
	DefaultRedisClientConfig = RedisClientConfig{
		DialTimeout:            time.Second * 5,
		ReadTimeout:            time.Second * 3,
		WriteTimeout:           time.Second * 3,
		MaxGetMultiConcurrency: 100,
		GetMultiBatchSize:      100,
		MaxSetMultiConcurrency: 100,
		SetMultiBatchSize:      100,
		TLSEnabled:             false,
		TLSConfig:              TLSConfig{},
		MaxAsyncConcurrency:    20,
		MaxAsyncBufferSize:     10000,

		SetAsyncCircuitBreaker: defaultCircuitBreakerConfig,
	}
)

var (
	ErrAsyncBufferFull = errors.New("the async buffer is full")
)

Functions

func NewMemcachedClient

func NewMemcachedClient(logger log.Logger, name string, conf []byte, reg prometheus.Registerer) (*memcachedClient, error)

NewMemcachedClient makes a new RemoteCacheClient.

func NewMemcachedClientWithConfig

func NewMemcachedClientWithConfig(logger log.Logger, name string, config MemcachedClientConfig, reg prometheus.Registerer) (*memcachedClient, error)

NewMemcachedClientWithConfig makes a new RemoteCacheClient.

Types

type AddressProvider

type AddressProvider interface {
	// Resolves the provided list of memcached cluster to the actual nodes
	Resolve(context.Context, []string, bool) error

	// Returns the nodes
	Addresses() []string
}

AddressProvider performs node address resolution given a list of clusters.

type AsyncOperationProcessor

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

func NewAsyncOperationProcessor

func NewAsyncOperationProcessor(bufferSize, concurrency int) *AsyncOperationProcessor

NewAsyncOperationProcessor creates an async processor with given bufferSize and concurrency.

func (*AsyncOperationProcessor) EnqueueAsync

func (p *AsyncOperationProcessor) EnqueueAsync(op func()) error

EnqueueAsync enqueues op to async queue. If enqueue failed, ErrAsyncBufferFull is returned.

func (*AsyncOperationProcessor) Stop

func (p *AsyncOperationProcessor) Stop()

type CircuitBreaker

type CircuitBreaker interface {
	Execute(func() error) error
}

CircuitBreaker implements the circuit breaker pattern https://en.wikipedia.org/wiki/Circuit_breaker_design_pattern.

type CircuitBreakerConfig

type CircuitBreakerConfig struct {
	// Enabled enables circuite breaker.
	Enabled bool `yaml:"enabled"`

	// HalfOpenMaxRequests is the maximum number of requests allowed to pass through
	// when the circuit breaker is half-open.
	// If set to 0, the circuit breaker allows only 1 request.
	HalfOpenMaxRequests uint32 `yaml:"half_open_max_requests"`
	// OpenDuration is the period of the open state after which the state of the circuit breaker becomes half-open.
	// If set to 0, the circuit breaker utilizes the default value of 60 seconds.
	OpenDuration time.Duration `yaml:"open_duration"`
	// MinRequests is minimal requests to trigger the circuit breaker.
	MinRequests uint32 `yaml:"min_requests"`
	// ConsecutiveFailures represents consecutive failures based on CircuitBreakerMinRequests to determine if the circuit breaker should open.
	ConsecutiveFailures uint32 `yaml:"consecutive_failures"`
	// FailurePercent represents the failure percentage, which is based on CircuitBreakerMinRequests, to determine if the circuit breaker should open.
	FailurePercent float64 `yaml:"failure_percent"`
}

CircuitBreakerConfig is the config for the circuite breaker.

type MemcachedClient

type MemcachedClient = RemoteCacheClient

MemcachedClient for compatible.

type MemcachedClientConfig

type MemcachedClientConfig struct {
	// Addresses specifies the list of memcached addresses. The addresses get
	// resolved with the DNS provider.
	Addresses []string `yaml:"addresses"`

	// Timeout specifies the socket read/write timeout.
	Timeout time.Duration `yaml:"timeout"`

	// MaxIdleConnections specifies the maximum number of idle connections that
	// will be maintained per address. For better performances, this should be
	// set to a number higher than your peak parallel requests.
	MaxIdleConnections int `yaml:"max_idle_connections"`

	// MaxAsyncConcurrency specifies the maximum number of SetAsync goroutines.
	MaxAsyncConcurrency int `yaml:"max_async_concurrency"`

	// MaxAsyncBufferSize specifies the queue buffer size for SetAsync operations.
	MaxAsyncBufferSize int `yaml:"max_async_buffer_size"`

	// MaxGetMultiConcurrency specifies the maximum number of concurrent GetMulti() operations.
	// If set to 0, concurrency is unlimited.
	MaxGetMultiConcurrency int `yaml:"max_get_multi_concurrency"`

	// MaxItemSize specifies the maximum size of an item stored in memcached.
	// Items bigger than MaxItemSize are skipped.
	// If set to 0, no maximum size is enforced.
	MaxItemSize model.Bytes `yaml:"max_item_size"`

	// MaxGetMultiBatchSize specifies the maximum number of keys a single underlying
	// GetMulti() should run. If more keys are specified, internally keys are split
	// into multiple batches and fetched concurrently, honoring MaxGetMultiConcurrency parallelism.
	// If set to 0, the max batch size is unlimited.
	MaxGetMultiBatchSize int `yaml:"max_get_multi_batch_size"`

	// DNSProviderUpdateInterval specifies the DNS discovery update interval.
	DNSProviderUpdateInterval time.Duration `yaml:"dns_provider_update_interval"`

	// AutoDiscovery configures memached client to perform auto-discovery instead of DNS resolution
	AutoDiscovery bool `yaml:"auto_discovery"`

	// SetAsyncCircuitBreaker configures the circuit breaker for SetAsync operations.
	SetAsyncCircuitBreaker CircuitBreakerConfig `yaml:"set_async_circuit_breaker_config"`
}

MemcachedClientConfig is the config accepted by RemoteCacheClient.

type MemcachedJumpHashSelector

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

MemcachedJumpHashSelector implements the memcache.ServerSelector interface, utilizing a jump hash to distribute keys to servers.

While adding or removing servers only requires 1/N keys to move, servers are treated as a stack and can only be pushed/popped. Therefore, MemcachedJumpHashSelector works best for servers with consistent DNS names where the naturally sorted order is predictable (ie. Kubernetes statefulsets).

func (*MemcachedJumpHashSelector) Each

func (s *MemcachedJumpHashSelector) Each(f func(net.Addr) error) error

Each iterates over each server and calls the given function. If f returns a non-nil error, iteration will stop and that error will be returned.

func (*MemcachedJumpHashSelector) PickServer

func (s *MemcachedJumpHashSelector) PickServer(key string) (net.Addr, error)

PickServer returns the server address that a given item should be shared onto.

func (*MemcachedJumpHashSelector) SetServers

func (s *MemcachedJumpHashSelector) SetServers(servers ...string) error

SetServers changes a MemcachedJumpHashSelector's set of servers at runtime and is safe for concurrent use by multiple goroutines.

Each server is given equal weight. A server is given more weight if it's listed multiple times.

SetServers returns an error if any of the server names fail to resolve. No attempt is made to connect to the server. If any error occurs, no changes are made to the internal server list.

To minimize the number of rehashes for keys when scaling the number of servers in subsequent calls to SetServers, servers are stored in natural sort order.

type RedisClient

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

func NewRedisClient

func NewRedisClient(logger log.Logger, name string, conf []byte, reg prometheus.Registerer) (*RedisClient, error)

NewRedisClient makes a new RedisClient.

func NewRedisClientWithConfig

func NewRedisClientWithConfig(logger log.Logger, name string, config RedisClientConfig,
	reg prometheus.Registerer) (*RedisClient, error)

NewRedisClientWithConfig makes a new RedisClient.

func (*RedisClient) GetMulti

func (c *RedisClient) GetMulti(ctx context.Context, keys []string) map[string][]byte

GetMulti implement RemoteCacheClient.

func (*RedisClient) SetAsync

func (c *RedisClient) SetAsync(key string, value []byte, ttl time.Duration) error

SetAsync implement RemoteCacheClient.

func (*RedisClient) SetMulti

func (c *RedisClient) SetMulti(data map[string][]byte, ttl time.Duration)

SetMulti set multiple keys and value.

func (*RedisClient) Stop

func (c *RedisClient) Stop()

Stop implement RemoteCacheClient.

type RedisClientConfig

type RedisClientConfig struct {
	// Addr specifies the addresses of redis server.
	Addr string `yaml:"addr"`

	// Use the specified Username to authenticate the current connection
	// with one of the connections defined in the ACL list when connecting
	// to a Redis 6.0 instance, or greater, that is using the Redis ACL system.
	Username string `yaml:"username"`
	// Optional password. Must match the password specified in the
	// requirepass server configuration option (if connecting to a Redis 5.0 instance, or lower),
	// or the User Password when connecting to a Redis 6.0 instance, or greater,
	// that is using the Redis ACL system.
	Password string `yaml:"password"`

	// DB Database to be selected after connecting to the server.
	DB int `yaml:"db"`

	// DialTimeout specifies the client dial timeout.
	DialTimeout time.Duration `yaml:"dial_timeout"`

	// ReadTimeout specifies the client read timeout.
	ReadTimeout time.Duration `yaml:"read_timeout"`

	// WriteTimeout specifies the client write timeout.
	WriteTimeout time.Duration `yaml:"write_timeout"`

	// MaxGetMultiConcurrency specifies the maximum number of concurrent GetMulti() operations.
	// If set to 0, concurrency is unlimited.
	MaxGetMultiConcurrency int `yaml:"max_get_multi_concurrency"`

	// GetMultiBatchSize specifies the maximum size per batch for mget.
	GetMultiBatchSize int `yaml:"get_multi_batch_size"`

	// MaxSetMultiConcurrency specifies the maximum number of concurrent SetMulti() operations.
	// If set to 0, concurrency is unlimited.
	MaxSetMultiConcurrency int `yaml:"max_set_multi_concurrency"`

	// SetMultiBatchSize specifies the maximum size per batch for pipeline set.
	SetMultiBatchSize int `yaml:"set_multi_batch_size"`

	// TLSEnabled enable tls for redis connection.
	TLSEnabled bool `yaml:"tls_enabled"`

	// TLSConfig to use to connect to the redis server.
	TLSConfig TLSConfig `yaml:"tls_config"`

	// If not zero then client-side caching is enabled.
	// Client-side caching is when data is stored in memory
	// instead of fetching data each time.
	// See https://redis.io/docs/manual/client-side-caching/ for info.
	CacheSize model.Bytes `yaml:"cache_size"`

	// MasterName specifies the master's name. Must be not empty
	// for Redis Sentinel.
	MasterName string `yaml:"master_name"`

	// MaxAsyncBufferSize specifies the queue buffer size for SetAsync operations.
	MaxAsyncBufferSize int `yaml:"max_async_buffer_size"`

	// MaxAsyncConcurrency specifies the maximum number of SetAsync goroutines.
	MaxAsyncConcurrency int `yaml:"max_async_concurrency"`

	// SetAsyncCircuitBreaker configures the circuit breaker for SetAsync operations.
	SetAsyncCircuitBreaker CircuitBreakerConfig `yaml:"set_async_circuit_breaker_config"`
}

RedisClientConfig is the config accepted by RedisClient.

type RemoteCacheClient

type RemoteCacheClient interface {
	// GetMulti fetches multiple keys at once from remoteCache. In case of error,
	// an empty map is returned and the error tracked/logged.
	GetMulti(ctx context.Context, keys []string) map[string][]byte

	// SetAsync enqueues an asynchronous operation to store a key into memcached.
	// Returns an error in case it fails to enqueue the operation. In case the
	// underlying async operation will fail, the error will be tracked/logged.
	SetAsync(key string, value []byte, ttl time.Duration) error

	// Stop client and release underlying resources.
	Stop()
}

RemoteCacheClient is a high level client to interact with remote cache.

type TLSConfig

type TLSConfig struct {
	// The CA cert to use for the targets.
	CAFile string `yaml:"ca_file"`
	// The client cert file for the targets.
	CertFile string `yaml:"cert_file"`
	// The client key file for the targets.
	KeyFile string `yaml:"key_file"`
	// Used to verify the hostname for the targets. See https://tools.ietf.org/html/rfc4366#section-3.1
	ServerName string `yaml:"server_name"`
	// Disable target certificate validation.
	InsecureSkipVerify bool `yaml:"insecure_skip_verify"`
}

TLSConfig configures TLS connections.