dolly

README

Dolly

ZMQ centralized key/value clone model with PUB/SUB

This simple network of peers replicates a key/value store using PUB/SUB ZMQ sockets. A client can make a GET request to any replica and receive that replica's current value of the key. Consistency is maintained by serializing all PUT requests through the leader -- which is selected as the replica with the lowest PID value. If the leader goes down, GET requests can still be made, but all PUTs are dropped until the leader comes back online.

Getting Started

To run the code, first get and install the package:

$ go get github.com/bbengfort/dolly

Note that this will require the installation of ZMQ, which can be tricky depending on your environment. Then create a peers.json file to define the replica network:

[
  {
    "pid": 1,
    "name": "alpha",
    "address": "localhost",
    "host": "apollo",
    "ipaddr": "127.0.0.1",
    "updates": 3264,
    "snapshots": 3265,
    "requests": 3266
  },
  {
    "pid": 2,
    "name": "bravo",
    "address": "localhost",
    "host": "apollo",
    "ipaddr": "127.0.0.1",
    "updates": 3267,
    "snapshots": 3268,
    "requests": 3269
  },
  {
    "pid": 3,
    "name": "charlie",
    "address": "localhost",
    "host": "apollo",
    "ipaddr": "127.0.0.1",
    "updates": 3270,
    "snapshots": 3271,
    "requests": 3272
  }
]

Note that each replica requires three ports to bind or connect sockets on.

• updates: the port where the leader binds PUB and replicas connect SUB to get key/value updates.
• snapshots: the port where the leader binds ROUTER and replicas connect DEALER so that a late replica can catch up with the leader.
• requests: the port where the leader binds PULL and replicas and clients bind PUSH so that the leader can have its state updated.

Documentation

Index

• Constants
• func LogLevel() string
• func SetLogLevel(level uint8)
• type Client
  • func (c *Client) Close() error
  • func (c *Client) Connect() (err error)
  • func (c *Client) Get(key string, timeout time.Duration) error
  • func (c *Client) Put(key, val string, timeout time.Duration) error
• type Leader
  • func (l *Leader) Bind() (err error)
  • func (l *Leader) Serve(ctx *zmq.Context, echan chan<- error)
• type Message
  • func RecvMessage(sock *zmq.Socket, route bool) (*Message, []byte, error)
  • func (m *Message) Send(sock *zmq.Socket, route []byte) error
• type Network
  • func New(peers string) (*Network, error)
  • func (n *Network) Client(name string) (*Client, error)
  • func (n *Network) Run(name string) (err error)
• type Replica
  • func (r *Replica) Bind() (err error)
  • func (r *Replica) Connect(leader *Replica) (err error)
  • func (r *Replica) Serve(leader *Replica, ctx *zmq.Context, echan chan<- error)
  • func (r *Replica) Snapshot() (err error)
• type Replicas
  • func (r Replicas) Get(name string) (*Replica, error)
  • func (r Replicas) Leader() (*Replica, error)

Constants

const (
	Trace uint8 = iota
	Debug
	Info
	Warn
	Silent
)

Levels for implementing the debug and trace message functionality.

const (
	MethodGet      = "Get"
	MethodPut      = "Put"
	MethodError    = "Error"
	MethodSnapshot = "Snapshot"
	MethodTerm     = "Terminate"
)

Method constants

Functions

func LogLevel

func LogLevel() string

LogLevel returns a string representation of the current level

func SetLogLevel

func SetLogLevel(level uint8)

SetLogLevel modifies the log level for messages at runtime. Ensures that the highest level that can be set is the silent level.

Types

type Client

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

Client connects to the a replica and makes requests.

func (*Client) Close

func (c *Client) Close() error

Close all sockets on the client and stop resonding to requests.

func (*Client) Connect

func (c *Client) Connect() (err error)

Connect all sockets from the client to the leader.

func (*Client) Get

func (c *Client) Get(key string, timeout time.Duration) error

Get the value for the specified key

func (*Client) Put

func (c *Client) Put(key, val string, timeout time.Duration) error

Put a value for the specified key.

type Leader

type Leader struct {
	Replica
}

Leader defines a server that can respond to both Get and Put requests and publishes state to all replica subscribers.

func (*Leader) Bind

func (l *Leader) Bind() (err error)

Bind the sockets on the appropriate ports

func (*Leader) Serve

func (l *Leader) Serve(ctx *zmq.Context, echan chan<- error)

Serve the leader, publishing state updates and responding to snapshot requests as well as GET and PUT requests.

type Message

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

Message represents a message that can be read off the wire.

func RecvMessage

func RecvMessage(sock *zmq.Socket, route bool) (*Message, []byte, error)

RecvMessage off the socket, serializing correctly. If route is true, then the message is read with the identity, otherwise it is treated as a subscription message.

func (*Message) Send

func (m *Message) Send(sock *zmq.Socket, route []byte) error

Send the message on the socket

type Network

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

Network defines all sockets for the local process.

func New

func New(peers string) (*Network, error)

New creates a Dolly network from the specified peers.json configuration.

func (*Network) Client

func (n *Network) Client(name string) (*Client, error)

Client returns a client connection for the specified replica.

func (*Network) Run

func (n *Network) Run(name string) (err error)

Run a replica or leader with the specified name.

type Replica

type Replica struct {
	PID       uint16 `json:"pid"`       // the precedence id of the peer
	Name      string `json:"name"`      // unique name of the peer
	Addr      string `json:"address"`   // the network address of the peer
	Host      string `json:"host"`      // the hostname of the peer
	IPAddr    string `json:"ipaddr"`    // the ip address of the peer
	Updates   uint16 `json:"updates"`   // the port the replica publishes updates on
	Snapshots uint16 `json:"snapshots"` // the port the replica fetches snapshots on
	Requests  uint16 `json:"requests"`  // the port the replica handles requests on
	// contains filtered or unexported fields
}

Replica defines a peer on the network that can respond to Get requests and synchronizes state by subscribing to the leader.

func (*Replica) Bind

func (r *Replica) Bind() (err error)

Bind the requests endpoint

func (*Replica) Connect

func (r *Replica) Connect(leader *Replica) (err error)

Connect the sockets to the leader.

func (*Replica) Serve

func (r *Replica) Serve(leader *Replica, ctx *zmq.Context, echan chan<- error)

Serve requests and subscribe to the leader to get updates.

func (*Replica) Snapshot

func (r *Replica) Snapshot() (err error)

Snapshot sends a snapshot request to the server to become up to date.

type Replicas

type Replicas []*Replica

Replicas represents a collection of replicas.

func (Replicas) Get

func (r Replicas) Get(name string) (*Replica, error)

Get a replica by name, returns nil if not found.

func (Replicas) Leader

func (r Replicas) Leader() (*Replica, error)

Leader returns the replica that has the lowest PID.