drax

README

Drax

An embeddable distributed k/v store for Go

Drax is a distributed k/v store based on raft (specifically https://github.com/hashicorp/raft) The intent is to be relatively light-weight and embeddable, while implementing the API from https://github.com/docker/libkv

Drax is NOT intended for production use. Use at your own risk. This should be considered ALPHA quality.

Boltdb is used to persist raft logs (and as such the K/V data)

Usage

For running the server, see cmd/kv.go as an example to get going.

Server Side

  // on node 1
  listener, _ := net.Listen("tcp", "10.0.0.1:2380")
  peerAddr := ""
  dialerFn := func(addr string, timeout time.Duration) (net.Conn, error) {
    return net.DialTimeout("tcp", addr, timeout)
  }
  cluster, _ := drax.New(listener, dialerFn, "/var/lib/drax", listener.Addr().String(), peerAddr)


  // on node 2
  listener, _ := net.Listen("tcp", "10.0.0.2:2380")
  peerAddr := "10.0.0.1:2380"
  dialerFn := func(addr string, timeout time.Duration) (net.Conn, error) {
    return net.DialTimeout("tcp", addr, timeout)
  }
  cluster, _ := drax.New(listener, dialerFn, "/var/lib/drax", listener.Addr().String(), peerAddr)

A node that does not specify a peer, and that does not already contain a peer in it's peer store will be setup in single node mode and made the leader. Once it is the leader it will exit single-node mode.

You can join a cluster by specifying any active peer's address, it does not need to be the leader.

Client side

  dialerFn := func(addr string, timeout time.Duration) (net.Conn, error) {
    return net.DialTimeout("tcp", addr, timeout)
  }
  retryTimeout := 10*time.Second
  client := client.New("10.0.0.1:2380", retryTimeout, dailerFn)
  kvPair, err := client.Get("/foo")

Requests to the K/V store can be sent to any peer and it will be forwarded to the leader. This implements libkv's Store interface.

You can also communicate with the k/v store directly from the cluster:

  kvPair, err := cluster.KVStore().Get("/foo")

This also implements libkv's Store interface.

###TODO:

• Add (more) tests
• Improve RPC semantics
• Look at using something other than JSON for encoding/decoding K/V messages, and RPC messages
• Implement cluster management, adding/removing nodes to the store cluster as needed

Documentation

Index

• Variables
• type Cluster
  • func New(l net.Listener, rpcDialer rpc.DialerFn, home, addr, peer string, ...) (*Cluster, error)
  • func (c *Cluster) Addr() string
  • func (c *Cluster) Errors() <-chan error
  • func (c *Cluster) Home() string
  • func (c *Cluster) KVStore() libkvstore.Store
  • func (c *Cluster) Shutdown()
  • func (c *Cluster) ShutdownCh() <-chan struct{}
• type Raft
  • func (r *Raft) AddPeer(peer string) error
  • func (r *Raft) Apply(b []byte) error
  • func (r *Raft) Close() error
  • func (r *Raft) GetLeader() string
  • func (r *Raft) IsLeader() bool
  • func (r *Raft) Leader() string
  • func (r *Raft) LeaderCh() <-chan interface{}
  • func (r *Raft) Peers() ([]string, error)
  • func (r *Raft) RemovePeer(peer string) error
  • func (r *Raft) SetPeers(peers []string) error
  • func (r *Raft) ShutdownCh() <-chan struct{}

Variables

var (
	// ErrKeyNotFound - Key does not exist in the store
	ErrKeyNotFound = libkvstore.ErrKeyNotFound
	// ErrKeyModified - Key was modified during atomic operation
	ErrKeyModified = libkvstore.ErrKeyModified
	// ErrCallNotSupported - call is not supported
	ErrCallNotSupported = libkvstore.ErrCallNotSupported
)

Types

type Cluster

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

Cluster is used to manage all the cluster state for a given node

func New

func New(l net.Listener, rpcDialer rpc.DialerFn, home, addr, peer string, logger io.Writer) (*Cluster, error)

New creates a new Cluster and starts it

func (*Cluster) Addr

func (c *Cluster) Addr() string

Addr cluster is available on

func (*Cluster) Errors

func (c *Cluster) Errors() <-chan error

Errors returns a channel receiver that callers can use to listen for cluster errors

func (*Cluster) Home

func (c *Cluster) Home() string

Home is the home path where the cluster state is stored

func (*Cluster) KVStore

func (c *Cluster) KVStore() libkvstore.Store

KVStore provides access to the underlying KV store

func (*Cluster) Shutdown

func (c *Cluster) Shutdown()

Shutdown stops the local cluster node

func (*Cluster) ShutdownCh

func (c *Cluster) ShutdownCh() <-chan struct{}

ShutdownCh returns a channel that can be watched to determine if the cluster has been shut down

type Raft

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

Raft wraps the underlying raft.Raft with some extra helpers to expose to other, non-raft components

func (*Raft) AddPeer

func (r *Raft) AddPeer(peer string) error

AddPeer adds the given peer to the raft cluster

func (*Raft) Apply

func (r *Raft) Apply(b []byte) error

Apply applies the bytes to the raft log If this is not the leader node, the request is sent to the leader

func (*Raft) Close

func (r *Raft) Close() error

Close shutsdown the raft for the local node

func (*Raft) GetLeader

func (r *Raft) GetLeader() string

GetLeader loops, waiting for the leader

func (*Raft) IsLeader

func (r *Raft) IsLeader() bool

IsLeader returns whether this node is the raft leader

func (*Raft) Leader

func (r *Raft) Leader() string

Leader returns the current leader

func (*Raft) LeaderCh

func (r *Raft) LeaderCh() <-chan interface{}

LeaderCh can be watched for changes in raft node state

func (*Raft) Peers

func (r *Raft) Peers() ([]string, error)

Peers returns the currnet list of raft peers

func (*Raft) RemovePeer

func (r *Raft) RemovePeer(peer string) error

RemovePeer removes the given peer from the raft cluster

func (*Raft) SetPeers

func (r *Raft) SetPeers(peers []string) error

SetPeers sets the list of raft peers

func (*Raft) ShutdownCh

func (r *Raft) ShutdownCh() <-chan struct{}

ShutdownCh can be watched to determine if the raft has been shutdown