kv

Documentation

Index

• Constants
• Variables
• type BatchOperation
• type KV
  • func New(store Storer, cfg *KVConfig) (*KV, error)
  • func (kv *KV) Close() error
  • func (kv *KV) Delete(key string) error
  • func (kv *KV) Get(key string) ([]byte, error)
  • func (kv *KV) GetServers() ([]*api.Server, error)
  • func (kv *KV) Join(id, addr string) error
  • func (kv *KV) Leave(id string) error
  • func (kv *KV) List() <-chan []byte
  • func (kv *KV) Set(key string, value []byte) error
  • func (kv *KV) WaitForLeader(timeout time.Duration) error
• type KVConfig
• type RaftConfig
• type RequestType
• type Storer
• type StreamLayer
  • func NewStreamLayer(ln net.Listener) *StreamLayer
  • func (s *StreamLayer) Accept() (net.Conn, error)
  • func (s *StreamLayer) Addr() net.Addr
  • func (s *StreamLayer) Close() error
  • func (s *StreamLayer) Dial(addr raft.ServerAddress, timeout time.Duration) (net.Conn, error)

Constants

const (
	// RaftRPC is the first byte sent on a connection to identify it as a Raft command
	// for the connection multiplexer.
	RaftRPC = 1

	// RetainSnapshotCount defines how many snapshots to retain.
	RetainSnapshotCount = 2

	// RaftTimeout defines the generic timeout for Raft operations.
	RaftTimeout = 10 * time.Second
)

Variables

var (
	ErrLeaderTimeout = errors.New("timed out waiting for leader")
	ErrNotRaftRPC    = errors.New("not a raft rpc connection")
)

Types

type BatchOperation

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

type KV

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

KV is a distributed key-value store implementation using Raft consensus.

func New

func New(store Storer, cfg *KVConfig) (*KV, error)

New creates and initializes a new distributed KV.

func (*KV) Close

func (kv *KV) Close() error

Close gracefully shuts down the Raft node.

func (*KV) Delete

func (kv *KV) Delete(key string) error

Delete applies a delete operation to the distributed store via Raft.

func (*KV) Get

func (kv *KV) Get(key string) ([]byte, error)

Get retrieves a value from the local store.

func (*KV) GetServers

func (kv *KV) GetServers() ([]*api.Server, error)

GetServers retrieves the current list of servers in the Raft configuration.

func (*KV) Join

func (kv *KV) Join(id, addr string) error

Join adds a new node (voter) to the Raft cluster. This operation must be executed on the cluster leader.

func (*KV) Leave

func (kv *KV) Leave(id string) error

Leave removes a node from the Raft cluster.

func (*KV) List

func (kv *KV) List() <-chan []byte

List returns a channel that streams all values from the local store.

func (*KV) Set

func (kv *KV) Set(key string, value []byte) error

Set applies a set operation to the distributed store via Raft.

func (*KV) WaitForLeader

func (kv *KV) WaitForLeader(timeout time.Duration) error

WaitForLeader blocks until the Raft node detects a leader in the cluster or times out.

type KVConfig

type KVConfig struct {
	Raft    RaftConfig
	DataDir string
}

KVConfig holds the general configuration for the KV store.

type RaftConfig

type RaftConfig struct {
	raft.Config
	BindAddr    string
	StreamLayer *StreamLayer
	Bootstrap   bool
	RPCPort     string
}

RaftConfig holds configuration specific to the Raft consensus mechanism.

type RequestType

type RequestType uint8

RequestType identifies the type of operation being applied to the Raft log.

const (
	RequestTypeSet RequestType = iota
	RequestTypeDelete
)

type Storer

type Storer interface {
	Get(key string) ([]byte, error)
	Set(key string, value []byte) error
	Delete(key string) error
	List() <-chan []byte
}

Storer defines the interface for a the key-value storage.

type StreamLayer

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

StreamLayer implements raft.StreamLayer to allow using a custom listener (e.g., from cmux) with the Hashicorp Raft library.

func NewStreamLayer

func NewStreamLayer(ln net.Listener) *StreamLayer

func (*StreamLayer) Accept

func (s *StreamLayer) Accept() (net.Conn, error)

Accept accepts a new incoming connection. It consumes the first byte (the multiplexing header) to verify the protocol and provides the clean connection to the Raft library.

func (*StreamLayer) Addr

func (s *StreamLayer) Addr() net.Addr

Addr returns the listener's network address.

func (*StreamLayer) Close

func (s *StreamLayer) Close() error

Close closes the listener.

func (*StreamLayer) Dial

func (s *StreamLayer) Dial(
	addr raft.ServerAddress,
	timeout time.Duration,
) (net.Conn, error)

Dial establishes an outgoing connection to another Raft node. It prepends the RaftRPC byte to identify the protocol to the remote multiplexer.