diff --git a/disco/disco.go b/disco/disco.go
new file mode 100644
index 000000000..c2fc2145d
--- /dev/null
+++ b/disco/disco.go
@@ -0,0 +1,209 @@
+package disco
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/molecula/etcd-test/disco"
+	"github.com/pilosa/pilosa/v2/roaring"
+)
+
+var (
+	ErrTooManyResults error = fmt.Errorf("too many results")
+	ErrNoResults      error = fmt.Errorf("no results")
+	ErrKeyDeleted     error = fmt.Errorf("key deleted")
+)
+
+type Peer struct {
+	URL string
+	ID  string
+}
+
+func (p *Peer) String() string {
+	return fmt.Sprintf(`{"ID": "%s", "URL": "%s"}`, p.ID, p.URL)
+}
+
+type DisCo interface {
+	io.Closer
+
+	Start(ctx context.Context) (InitialClusterState, error)
+	IsLeader() bool
+	ID() string
+	Leader() *Peer
+	Peers() []*Peer
+	DeleteNode(ctx context.Context, id string) error
+}
+
+type (
+	InitialClusterState string
+	ClusterState        string
+)
+
+const (
+	InitialClusterStateNew      InitialClusterState = "new"
+	InitialClusterStateExisting InitialClusterState = "existing"
+
+	// ClusterState represents the state returned in the /status endpoint.
+	ClusterStateUnknown  ClusterState = "UNKNOWN"
+	ClusterStateStarting ClusterState = "STARTING"
+	ClusterStateDegraded ClusterState = "DEGRADED" // cluster is running but we've lost some # of hosts >0 but < replicaN
+	ClusterStateNormal   ClusterState = "NORMAL"
+	ClusterStateResizing ClusterState = "RESIZING" // cluster is replicating data to other nodes
+	ClusterStateDown     ClusterState = "DOWN"     // cluster is unable to serve queries
+)
+
+type NodeState string
+
+const (
+	NodeStateUnknown  NodeState = "UNKNOWN"
+	NodeStateStarting NodeState = "STARTING"
+	NodeStateStarted  NodeState = "STARTED"
+	NodeStateResizing NodeState = "RESIZING"
+)
+
+type Stator interface {
+	Started(ctx context.Context) error
+	ClusterState(context.Context) (ClusterState, error)
+	NodeState(context.Context, string) (NodeState, error)
+	NodeStates(context.Context) (map[string]NodeState, error)
+}
+
+// Index is a struct which contains the data encoded for the index as well as
+// for each of its fields.
+type Index struct {
+	Data   []byte
+	Fields map[string][]byte
+}
+
+type Schemator interface {
+	Schema(ctx context.Context) (map[string]*Index, error)
+	Index(ctx context.Context, name string) ([]byte, error)
+	CreateIndex(ctx context.Context, name string, val []byte) error
+	DeleteIndex(ctx context.Context, name string) error
+	Field(ctx context.Context, index, field string) ([]byte, error)
+	CreateField(ctx context.Context, index, field string, val []byte) error
+	DeleteField(ctx context.Context, index, field string) error
+}
+
+type Metadata interface {
+	Marshal() ([]byte, error)
+	Unmarshal([]byte) error
+}
+
+type Metadator interface {
+	Metadata(ctx context.Context, peerID string) ([]byte, error)
+	SetMetadata(ctx context.Context, metadata []byte) error
+}
+
+// Resizer triggers resizing the node and changes cluster state into RESIZING.
+// We can also return some kind of handler from Resize function (e.g. key-value)
+type Resizer interface {
+	Resize(ctx context.Context) (func([]byte) error, error)
+	DoneResize() error
+	Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error
+}
+
+// Sharder is an interface used to maintain the set of availableShards bitmaps
+// per field.
+type Sharder interface {
+	Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error)
+	AddShard(ctx context.Context, index, field string, shard uint64) error
+	AddShards(ctx context.Context, index, field string, shards *roaring.Bitmap) (*roaring.Bitmap, error)
+	RemoveShard(ctx context.Context, index, field string, shard uint64) error
+}
+
+// NopDisCo represents a DisCo that doesn't do anything.
+var NopDisCo disco.DisCo = &nopDisCo{
+	Closer: nil,
+}
+
+type nopDisCo struct {
+	io.Closer
+}
+
+// Start is a no-op implementation of the DisCo Start method.
+func (n *nopDisCo) Start(ctx context.Context) (disco.InitialClusterState, error) {
+	return disco.InitialClusterStateNew, nil
+}
+
+// ID is a no-op implementation of the DisCo ID method.
+func (n *nopDisCo) ID() string {
+	return ""
+}
+
+// IsLeader is a no-op implementation of the DisCo IsLeader method.
+func (n *nopDisCo) IsLeader() bool {
+	return false
+}
+
+// Leader is a no-op implementation of the DisCo Leader method.
+func (n *nopDisCo) Leader() *disco.Peer {
+	return nil
+}
+
+// Peers is a no-op implementation of the DisCo Peers method.
+func (n *nopDisCo) Peers() []*disco.Peer {
+	return nil
+}
+
+// DeleteNode a no-op implementation of the DisCo DeleteNode method.
+func (n *nopDisCo) DeleteNode(context.Context, string) error {
+	return nil
+}
+
+// NopStator represents a Stator that doesn't do anything.
+var NopStator disco.Stator = &nopStator{}
+
+type nopStator struct{}
+
+// ClusterState is a no-op implementation of the Stator ClusterState method.
+func (n *nopStator) ClusterState(context.Context) (disco.ClusterState, error) {
+	return "", nil
+}
+
+func (n *nopStator) Started(ctx context.Context) error {
+	return nil
+}
+
+func (n *nopStator) NodeState(context.Context, string) (disco.NodeState, error) {
+	return disco.NodeStateUnknown, nil
+}
+
+func (n *nopStator) NodeStates(context.Context) (map[string]disco.NodeState, error) {
+	return nil, nil
+}
+
+// NopResizer represents a Resizer that doesn't do anything.
+var NopResizer disco.Resizer = &nopResizer{}
+
+type nopResizer struct{}
+
+func (*nopResizer) Resize(context.Context) (func([]byte) error, error)      { return nil, nil }
+func (*nopResizer) DoneResize() error                                       { return nil }
+func (*nopResizer) Watch(context.Context, string, func([]byte) error) error { return nil }
+
+// NopSharder represents a Sharder that doesn't do anything.
+var NopSharder disco.Sharder = &nopSharder{}
+
+type nopSharder struct{}
+
+// Shards is a no-op implementation of the Sharder Shards method.
+func (n *nopSharder) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
+	return nil, nil
+}
+
+// AddShard is a no-op implementation of the Sharder AddShard method.
+func (n *nopSharder) AddShard(ctx context.Context, index, field string, shard uint64) error {
+	return nil
+}
+
+// AddShards is a no-op implementation of the Sharder AddShards method.
+func (n *nopSharder) AddShards(ctx context.Context, index, field string, shards *roaring.Bitmap) (*roaring.Bitmap, error) {
+	return nil, nil
+}
+
+// RemoveShard is a no-op implementation of the Sharder RemoveShard method.
+func (n *nopSharder) RemoveShard(ctx context.Context, index, field string, shard uint64) error {
+	return nil
+}
diff --git a/etcd/cache.go b/etcd/cache.go
new file mode 100644
index 000000000..6cc011c0a
--- /dev/null
+++ b/etcd/cache.go
@@ -0,0 +1,134 @@
+package etcd
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/molecula/etcd-test/disco"
+)
+
+// EtcdWithCache is a wrapper around the Etcd type which will return a
+// cached value when the number of requests come in below a configured
+// frequency. It also breaks the cache after a configured TTL.
+type EtcdWithCache struct {
+	*Etcd
+
+	peerMetadataMu sync.RWMutex
+	peerMetadata   map[string][]byte
+
+	stateMu sync.Mutex
+
+	nodeStates         map[string]nodeState
+	nodeStateTTL       int // seconds
+	nodeStateFrequency int // max requests per second allowed before using the cache
+
+	clusterStateVal         disco.ClusterState
+	clusterStateTTL         int // seconds
+	clusterStateFrequency   int // max requests per second allowed before using the cache
+	clusterStateLastRequest time.Time
+	clusterStateLastCache   time.Time
+}
+
+type nodeState struct {
+	val         disco.NodeState
+	lastRequest time.Time
+	lastCache   time.Time
+}
+
+// NewEtcdWithCache returns a new instance of Cache.
+func NewEtcdWithCache(opt Options, replicas int) *EtcdWithCache {
+	return &EtcdWithCache{
+		Etcd: NewEtcd(opt, replicas),
+
+		nodeStateTTL:          6,
+		nodeStateFrequency:    1,
+		clusterStateTTL:       6,
+		clusterStateFrequency: 1,
+
+		peerMetadata: make(map[string][]byte),
+		nodeStates:   make(map[string]nodeState),
+	}
+}
+
+// Metadata is a cache wrapper around the Metadator.Metadata method.
+func (c *EtcdWithCache) Metadata(ctx context.Context, peerID string) ([]byte, error) {
+	c.peerMetadataMu.RLock()
+	v, ok := c.peerMetadata[peerID]
+	c.peerMetadataMu.RUnlock()
+	if ok {
+		return v, nil
+	}
+	v, err := c.Etcd.Metadata(ctx, peerID)
+	if err == nil {
+		c.peerMetadataMu.Lock()
+		c.peerMetadata[peerID] = v
+		c.peerMetadataMu.Unlock()
+	}
+	return v, err
+}
+
+// ClusterState is a cache wrapper around the Stator.ClusterState method.
+func (c *EtcdWithCache) ClusterState(ctx context.Context) (disco.ClusterState, error) {
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+
+	now := time.Now()
+	if now.Sub(c.clusterStateLastCache) > (time.Duration(c.clusterStateTTL)*time.Second) ||
+		now.Sub(c.clusterStateLastRequest) > (time.Second/time.Duration(c.clusterStateFrequency)) {
+		v, err := c.Etcd.ClusterState(ctx)
+		if err == nil {
+			// In order to avoid NodeState() returning a cached value after
+			// cluster state has changed, we reset the node state caches to
+			// ensure that the next call to NodeState() returns the latest
+			// value. And we only need to do this if the cluster state value has
+			// actually changed.
+			if c.clusterStateVal != v {
+				for k, ns := range c.nodeStates {
+					ns.lastCache = time.Time{}
+					c.nodeStates[k] = ns
+				}
+			}
+
+			c.clusterStateVal = v
+			c.clusterStateLastCache = now
+			c.clusterStateLastRequest = now
+		}
+		return v, err
+	}
+	c.clusterStateLastRequest = now
+	return c.clusterStateVal, nil
+}
+
+// NodeState is a cache wrapper around the Stator.NodeState method.
+func (c *EtcdWithCache) NodeState(ctx context.Context, peerID string) (disco.NodeState, error) {
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+
+	ns := c.nodeStates[peerID]
+
+	now := time.Now()
+	if now.Sub(ns.lastCache) > (time.Duration(c.nodeStateTTL)*time.Second) ||
+		now.Sub(ns.lastRequest) > (time.Second/time.Duration(c.nodeStateFrequency)) {
+		v, err := c.Etcd.NodeState(ctx, peerID)
+		if err == nil {
+			// In order to avoid ClusterState() returning a cached value after a
+			// node state has changed, we reset the cluster state cache to
+			// ensure that the next call to ClusterState() returns the latest
+			// value. And we only need to do this if the node state value has
+			// actually changed.
+			if ns.val != v {
+				c.clusterStateLastCache = time.Time{}
+			}
+
+			ns.val = v
+			ns.lastCache = now
+			ns.lastRequest = now
+			c.nodeStates[peerID] = ns
+		}
+		return v, err
+	}
+	ns.lastRequest = now
+	c.nodeStates[peerID] = ns
+	return ns.val, nil
+}
diff --git a/etcd/embed.go b/etcd/embed.go
new file mode 100644
index 000000000..78f6de04c
--- /dev/null
+++ b/etcd/embed.go
@@ -0,0 +1,1054 @@
+package etcd
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"path"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/molecula/etcd-test/disco"
+	"github.com/pilosa/pilosa/v2"
+	"github.com/pilosa/pilosa/v2/roaring"
+	"github.com/pkg/errors"
+	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/clientv3/clientv3util"
+	"go.etcd.io/etcd/clientv3/concurrency"
+	"go.etcd.io/etcd/embed"
+	"go.etcd.io/etcd/mvcc/mvccpb"
+	"go.etcd.io/etcd/pkg/types"
+)
+
+type Options struct {
+	Name         string `toml:"name"`
+	Dir          string `toml:"dir"`
+	LClientURL   string `toml:"listen-client-addr"`
+	AClientURL   string `toml:"advertise-client-addr"`
+	LPeerURL     string `toml:"listen-peer-addr"`
+	APeerURL     string `toml:"advertise-peer-addr"`
+	InitCluster  string `toml:"initial-cluster"`
+	ClusterURL   string `toml:"cluster-url"`
+	ClusterName  string `toml:"cluster-name"`
+	HeartbeatTTL int64  `toml:"heartbeat-ttl"`
+}
+
+var (
+	_ disco.DisCo     = &Etcd{}
+	_ disco.Schemator = &Etcd{}
+	_ disco.Stator    = &Etcd{}
+	_ disco.Metadator = &Etcd{}
+	_ disco.Resizer   = &Etcd{}
+	_ disco.Sharder   = &Etcd{}
+
+	ErrIndexExists = errors.New("index already exists")
+	ErrFieldExists = errors.New("field already exists")
+)
+
+const (
+	heartbeatPrefix = "/heartbeat/"
+	schemaPrefix    = "/schema/"
+	resizePrefix    = "/resize/"
+	metadataPrefix  = "/metadata/"
+	shardPrefix     = "/shard/"
+	lockPrefix      = "/lock/"
+)
+
+type leaseMetadata struct {
+	started bool
+}
+
+type Etcd struct {
+	options  Options
+	replicas int
+
+	heartbeatID     clientv3.LeaseID
+	heartbeatCancel context.CancelFunc
+
+	resizeCancel context.CancelFunc
+
+	lm leaseMetadata
+
+	e *embed.Etcd
+}
+
+func NewEtcd(opt Options, replicas int) *Etcd {
+	e := &Etcd{
+		options:  opt,
+		replicas: replicas,
+	}
+	return e
+}
+
+// Close implements io.Closer
+func (e *Etcd) Close() error {
+	if e.e != nil {
+		if e.resizeCancel != nil {
+			e.resizeCancel()
+		}
+		if e.heartbeatCancel != nil {
+			e.heartbeatCancel()
+		}
+		e.e.Server.Stop()
+		e.e.Close()
+		<-e.e.Server.StopNotify()
+	}
+
+	return nil
+}
+
+func parseOptions(opt Options) *embed.Config {
+	cfg := embed.NewConfig()
+	cfg.Debug = true
+	cfg.Name = opt.Name
+	cfg.Dir = opt.Dir
+	cfg.InitialClusterToken = opt.ClusterName
+	cfg.LCUrls = types.MustNewURLs([]string{opt.LClientURL})
+	cfg.ACUrls = types.MustNewURLs([]string{opt.AClientURL})
+	cfg.LPUrls = types.MustNewURLs([]string{opt.LPeerURL})
+	cfg.APUrls = types.MustNewURLs([]string{opt.APeerURL})
+
+	if opt.InitCluster != "" {
+		cfg.InitialCluster = opt.InitCluster
+		cfg.ClusterState = embed.ClusterStateFlagNew
+	} else {
+		cfg.InitialCluster = cfg.Name + "=" + opt.APeerURL
+	}
+
+	if opt.ClusterURL != "" {
+		cfg.ClusterState = embed.ClusterStateFlagExisting
+
+		cli, err := clientv3.NewFromURL(opt.ClusterURL)
+		if err != nil {
+			panic(err)
+		}
+		defer cli.Close()
+
+		log.Println("Cluster Members:")
+		mIDs, mNames, mURLs := memberList(cli)
+		for i, id := range mIDs {
+			log.Printf("\tid: %d, name: %s, url: %s\n", id, mNames[i], mURLs[i])
+			cfg.InitialCluster += "," + mNames[i] + "=" + mURLs[i]
+		}
+
+		log.Println("Joining Cluster:")
+		id, name := memberAdd(cli, opt.APeerURL)
+		log.Printf("\tid: %d, name: %s\n", id, name)
+	}
+
+	return cfg
+}
+
+// Start starts etcd and hearbeat
+func (e *Etcd) Start(ctx context.Context) (disco.InitialClusterState, error) {
+	opts := parseOptions(e.options)
+	state := disco.InitialClusterState(opts.ClusterState)
+
+	etcd, err := embed.StartEtcd(opts)
+	if err != nil {
+		return state, errors.Wrap(err, "starting etcd")
+	}
+	e.e = etcd
+
+	select {
+	case <-ctx.Done():
+		e.e.Server.Stop()
+		return state, ctx.Err()
+
+	case err := <-e.e.Err():
+		return state, err
+
+	case <-e.e.Server.ReadyNotify():
+		return state, e.startHeartbeat()
+	}
+}
+
+func (e *Etcd) startHeartbeat() error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "startHeartbeat: creates a new client")
+	}
+	defer cli.Close()
+
+	heartbeatID, heartbeatFunc, err := e.leaseKeepAlive(e.options.HeartbeatTTL)
+	if err != nil {
+		return errors.Wrap(err, "startHeartbeat: creates a new hearbeat")
+	}
+
+	ctx, heartbeatCancel := context.WithCancel(context.Background())
+	key, value := heartbeatPrefix+e.e.Server.ID().String(), disco.ClusterStateStarting
+	if e.e.Config().ClusterState == embed.ClusterStateFlagExisting {
+		value = disco.ClusterStateResizing
+	}
+
+	if _, err := cli.Put(ctx, key, string(value), clientv3.WithLease(heartbeatID)); err != nil {
+		heartbeatCancel()
+		return errors.Wrapf(err, "startHeartbeat: puts a key-value (%s, %s) with lease (%v)", key, value, heartbeatID)
+	}
+
+	e.heartbeatID, e.heartbeatCancel = heartbeatID, heartbeatCancel
+	go heartbeatFunc(ctx, time.Second)
+
+	return nil
+}
+
+func (e *Etcd) NodeState(ctx context.Context, peerID string) (disco.NodeState, error) {
+	cli, err := e.client()
+	if err != nil {
+		return disco.NodeStateUnknown, errors.Wrap(err, "NodeState: creates a new client")
+	}
+	defer cli.Close()
+
+	return e.nodeState(ctx, cli, peerID)
+}
+
+func (e *Etcd) nodeState(ctx context.Context, cli *clientv3.Client, peerID string) (disco.NodeState, error) {
+	resp, err := cli.Get(ctx, path.Join(resizePrefix, peerID), clientv3.WithCountOnly())
+	if err != nil {
+		return disco.NodeStateUnknown, err
+	}
+	if resp.Count > 0 {
+		return disco.NodeStateResizing, nil
+	}
+
+	resp, err = cli.Get(ctx, path.Join(heartbeatPrefix, peerID))
+	if err != nil {
+		return disco.NodeStateUnknown, err
+	}
+
+	if len(resp.Kvs) > 1 {
+		return disco.NodeStateUnknown, disco.ErrTooManyResults
+	}
+
+	if len(resp.Kvs) == 0 {
+		return disco.NodeStateUnknown, disco.ErrNoResults
+	}
+
+	return disco.NodeState(resp.Kvs[0].Value), nil
+}
+
+func (e *Etcd) NodeStates(ctx context.Context) (map[string]disco.NodeState, error) {
+	out := make(map[string]disco.NodeState)
+
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "NodeStates")
+	}
+	defer cli.Close()
+
+	members := e.e.Server.Cluster().Members()
+	for _, member := range members {
+		s, err := e.nodeState(ctx, cli, member.ID.String())
+		if err != nil {
+			log.Println("NodeStates get node state", member.ID.String(), err.Error())
+		}
+
+		out[member.ID.String()] = s
+	}
+
+	return out, nil
+}
+
+func (e *Etcd) Started(ctx context.Context) error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "Started")
+	}
+	defer cli.Close()
+
+	key, value := heartbeatPrefix+e.e.Server.ID().String(), disco.NodeStateStarted
+	if _, err = cli.Put(ctx, key, string(value), clientv3.WithLease(e.heartbeatID)); err == nil {
+		e.lm.started = true
+	}
+	return err
+}
+
+func (e *Etcd) ID() string {
+	return e.e.Server.ID().String()
+}
+
+func (e *Etcd) Peers() []*disco.Peer {
+	var peers []*disco.Peer
+	for _, member := range e.e.Server.Cluster().Members() {
+		peers = append(peers, &disco.Peer{ID: member.ID.String(), URL: member.PickPeerURL()})
+	}
+	return peers
+}
+
+func (e *Etcd) IsLeader() bool {
+	return e.e.Server.Leader() == e.e.Server.ID()
+}
+
+func (e *Etcd) Leader() *disco.Peer {
+	id := e.e.Server.Leader()
+	m := e.e.Server.Cluster().Member(id)
+	return &disco.Peer{ID: id.String(), URL: m.PickPeerURL()}
+}
+
+func (e *Etcd) ClusterState(ctx context.Context) (disco.ClusterState, error) {
+	if e.e == nil {
+		return disco.ClusterStateUnknown, nil
+	}
+
+	cli, err := e.client()
+	if err != nil {
+		return disco.ClusterStateUnknown, errors.WithMessage(err, "ClusterState: creates a new client")
+	}
+	defer cli.Close()
+
+	var (
+		heartbeats int = 0
+		resize     bool
+		starting   bool
+	)
+	members := e.e.Server.Cluster().Members()
+	for _, m := range members {
+		ns, err := e.nodeState(ctx, cli, m.ID.String())
+		if err != nil {
+			log.Println("ClusterState get node state", err.Error())
+			continue
+		}
+
+		heartbeats++
+
+		if ns == disco.NodeStateStarting {
+			starting = true
+		}
+
+		if ns == disco.NodeStateResizing {
+			resize = true
+		}
+	}
+
+	if resize {
+		return disco.ClusterStateResizing, nil
+	}
+
+	if starting {
+		return disco.ClusterStateStarting, nil
+	}
+
+	if heartbeats < len(members) {
+		if len(members)-heartbeats >= e.replicas {
+			return disco.ClusterStateDown, nil
+		}
+
+		return disco.ClusterStateDegraded, nil
+	}
+
+	return disco.ClusterStateNormal, nil
+}
+
+func (e *Etcd) Resize(ctx context.Context) (func([]byte) error, error) {
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "Resize: creates a new client")
+	}
+	defer cli.Close()
+
+	resizeID, resizeFunc, err := e.leaseKeepAlive(e.options.HeartbeatTTL)
+	if err != nil {
+		return nil, errors.Wrap(err, "Resize: creates a new hearbeat")
+	}
+
+	ctx, resizeCancel := context.WithCancel(ctx)
+	// Check if key exists - maybe we are still resizing
+	key := path.Join(resizePrefix, e.e.Server.ID().String())
+	txnResp, err := cli.Txn(ctx).
+		If(clientv3util.KeyMissing(key)).
+		Then(clientv3.OpPut(key, "", clientv3.WithLease(resizeID))).
+		Commit()
+	if err != nil {
+		resizeCancel()
+		return nil, errors.Wrapf(err, "Resize: txn puts key (%s) with lease (%v)", key, resizeID)
+	}
+
+	if !txnResp.Succeeded {
+		resizeCancel()
+		return nil, errors.Errorf("Resize: key (%s) exists - maybe node (%s) is resizing", key, e.ID())
+	}
+
+	e.resizeCancel = resizeCancel
+	go resizeFunc(ctx, time.Second)
+
+	return func(value []byte) error {
+		log.Println("Update progress:", key, string(value))
+		return e.putKey(ctx, key, string(value), clientv3.WithLease(resizeID))
+	}, nil
+}
+
+func (e *Etcd) DoneResize() error {
+	if e.resizeCancel != nil {
+		e.resizeCancel()
+	}
+	return nil
+}
+
+func (e *Etcd) Watch(ctx context.Context, peerID string, onUpdate func([]byte) error) error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "Watch: creates a new client")
+	}
+	defer cli.Close()
+
+	key := path.Join(resizePrefix, peerID)
+	for resp := range cli.Watch(ctx, key) {
+		if err := resp.Err(); err != nil {
+			return errors.Wrapf(err, "Watch: key (%s) response", key)
+		}
+
+		for _, ev := range resp.Events {
+			switch ev.Type {
+			case mvccpb.PUT:
+				if onUpdate != nil && ev.Kv.Value != nil {
+					if err := onUpdate(ev.Kv.Value); err != nil {
+						return err
+					}
+				}
+
+			case mvccpb.DELETE:
+				// nothing to watch - key was deleted
+				return errors.WithMessagef(disco.ErrKeyDeleted, "Watch key %s", key)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (e *Etcd) DeleteNode(ctx context.Context, nodeID string) error {
+	id, err := types.IDFromString(nodeID)
+	if err != nil {
+		return err
+	}
+
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "DeleteNode: creates a new client")
+	}
+	defer cli.Close()
+
+	_, err = cli.MemberRemove(ctx, uint64(id))
+	if err != nil {
+		return errors.Wrap(err, "DeleteNode: removes an existing member from the cluster")
+	}
+
+	return nil
+}
+
+func (e *Etcd) Schema(ctx context.Context) (map[string]*disco.Index, error) {
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "Schema: creating client")
+	}
+	defer cli.Close()
+
+	keys, vals, err := e.getKey(ctx, cli, schemaPrefix)
+	if err != nil {
+		return nil, err
+	}
+
+	m := make(map[string]*disco.Index)
+	for i, k := range keys {
+		tokens := strings.Split(strings.Trim(k, "/"), "/")
+		// token[0] contains the schemaPrefix
+		index := tokens[1]
+		if _, ok := m[index]; !ok {
+			m[index] = &disco.Index{
+				Data:   vals[i],
+				Fields: make(map[string][]byte),
+			}
+		}
+		flds := m[index].Fields
+
+		if len(tokens) > 2 {
+			field := tokens[2]
+			flds[field] = vals[i]
+		}
+	}
+
+	return m, nil
+}
+
+func (e *Etcd) Metadata(ctx context.Context, peerID string) ([]byte, error) {
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "Metadata")
+	}
+	defer cli.Close()
+
+	resp, err := cli.Get(ctx, path.Join(metadataPrefix, peerID))
+	if err != nil {
+		return nil, err
+	}
+
+	if len(resp.Kvs) > 1 {
+		return nil, disco.ErrTooManyResults
+	}
+
+	if len(resp.Kvs) == 0 {
+		return nil, disco.ErrNoResults
+	}
+
+	return resp.Kvs[0].Value, nil
+}
+
+func (e *Etcd) SetMetadata(ctx context.Context, metadata []byte) error {
+	err := e.putKey(ctx, path.Join(metadataPrefix,
+		e.e.Server.ID().String()),
+		string(metadata),
+	)
+	if err != nil {
+		return errors.Wrap(err, "SetMetadata")
+	}
+
+	return nil
+}
+
+func (e *Etcd) CreateIndex(ctx context.Context, name string, val []byte) error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "CreateIndex: creating client")
+	}
+	defer cli.Close()
+
+	key := schemaPrefix + name
+
+	// Set up Op to write index value as bytes.
+	op := clientv3.OpPut(key, "")
+	op.WithValueBytes(val)
+
+	// Check for key existence, and execute Op within a transaction.
+	resp, err := cli.KV.Txn(ctx).
+		If(clientv3util.KeyMissing(key)).
+		Then(op).
+		Commit()
+	if err != nil {
+		return errors.Wrap(err, "executing transaction")
+	}
+
+	if !resp.Succeeded {
+		return ErrIndexExists
+	}
+
+	return nil
+}
+
+func (e *Etcd) Index(ctx context.Context, name string) ([]byte, error) {
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "Index: creating client")
+	}
+	defer cli.Close()
+
+	return e.getKeyBytes(ctx, cli, schemaPrefix+name)
+}
+
+func (e *Etcd) DeleteIndex(ctx context.Context, name string) error {
+	// Delete any fields below the index path.
+	if err := e.delKey(ctx, schemaPrefix+name+"/", true); err != nil {
+		return errors.Wrap(err, "deleting index fields")
+	}
+	// Delete the index.
+	return e.delKey(ctx, schemaPrefix+name, false)
+}
+
+func (e *Etcd) Field(ctx context.Context, indexName string, name string) ([]byte, error) {
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "GetField: creating client")
+	}
+	defer cli.Close()
+
+	key := schemaPrefix + indexName + "/" + name
+	return e.getKeyBytes(ctx, cli, key)
+}
+
+func (e *Etcd) CreateField(ctx context.Context, indexName string, name string, val []byte) error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "CreateIndex: creating client")
+	}
+	defer cli.Close()
+
+	key := schemaPrefix + indexName + "/" + name
+
+	// Set up Op to write field value as bytes.
+	op := clientv3.OpPut(key, "")
+	op.WithValueBytes(val)
+
+	// Check for key existence, and execute Op within a transaction.
+	resp, err := cli.KV.Txn(ctx).
+		If(clientv3util.KeyMissing(key)).
+		Then(op).
+		Commit()
+	if err != nil {
+		return errors.Wrap(err, "executing transaction")
+	}
+
+	if !resp.Succeeded {
+		return ErrFieldExists
+	}
+
+	return nil
+}
+
+func (e *Etcd) DeleteField(ctx context.Context, indexname string, name string) error {
+	return e.delKey(ctx, schemaPrefix+indexname+"/"+name, false)
+}
+
+func (e *Etcd) putKey(ctx context.Context, key, val string, opts ...clientv3.OpOption) error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "putKey: creates a new client")
+	}
+	defer cli.Close()
+
+	if _, err := cli.KV.Put(ctx, key, val, opts...); err != nil {
+		return errors.Wrapf(err, "putKey: Put(%s, %s)", key, val)
+	}
+
+	return nil
+}
+
+func (e *Etcd) getKeyBytes(ctx context.Context, cli *clientv3.Client, key string) ([]byte, error) {
+	// Get the current value for the key.
+	resp, err := cli.Get(ctx, key)
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO: consider returning a "key does not exist" error instead of (nil, nil)
+	if len(resp.Kvs) == 0 {
+		return nil, nil
+	}
+
+	return resp.Kvs[0].Value, nil
+}
+
+func (e *Etcd) getKey(ctx context.Context, cli *clientv3.Client, key string) ([]string, [][]byte, error) {
+	resp, err := cli.KV.Txn(ctx).
+		If(clientv3.Compare(clientv3.Version(key), ">", -1)).
+		Then(clientv3.OpGet(key, clientv3.WithPrefix())).
+		Commit()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if !resp.Succeeded {
+		return nil, nil, fmt.Errorf("key %s does not exist", key)
+	}
+
+	var (
+		keys   []string
+		values [][]byte
+	)
+
+	for _, r := range resp.Responses {
+		for _, kv := range r.GetResponseRange().Kvs {
+			keys = append(keys, string(kv.Key))
+			values = append(values, kv.Value)
+		}
+	}
+
+	return keys, values, nil
+}
+
+func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) error {
+	cli, err := clientv3.NewFromURLs(e.e.Server.Cluster().ClientURLs())
+	if err != nil {
+		return errors.Wrap(err, "delKey")
+	}
+	defer cli.Close()
+
+	var opts []clientv3.OpOption
+	if withPrefix {
+		opts = append(opts, clientv3.WithPrefix())
+	}
+
+	_, err = cli.KV.Txn(ctx).
+		If(clientv3.Compare(clientv3.Version(key), ">", -1)).
+		Then(clientv3.OpDelete(key, opts...)).
+		Commit()
+
+	return err
+}
+
+func (e *Etcd) leaseKeepAlive(ttl int64) (clientv3.LeaseID, func(context.Context, time.Duration), error) {
+	cli, err := e.client()
+	if err != nil {
+		return 0, nil, errors.Wrap(err, "leaseKeepAlive: creates a new client")
+	}
+	defer cli.Close()
+
+	leaseResp, err := cli.Grant(context.TODO(), ttl)
+	if err != nil {
+		return 0, nil, errors.Wrapf(err, "leaseKeepAlive: creates a new lease (TTL: %d)", ttl)
+	}
+
+	keepaliveFunc := func(ctx context.Context, tick time.Duration) {
+		ticker := time.NewTicker(tick)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				log.Printf("leaseKeepAlive: %v\n", ctx.Err())
+				return
+
+			case <-ticker.C:
+				if cli, err := e.client(); err != nil {
+					log.Printf("leaseKeepAlive: creates a new client: %v\n", err)
+				} else {
+					if _, err = cli.KeepAliveOnce(ctx, leaseResp.ID); err != nil {
+						log.Printf("leaseKeepAlive: renews the lease (ID: %v): %v\n", leaseResp.ID, err)
+					}
+					cli.Close()
+				}
+			}
+		}
+	}
+
+	return leaseResp.ID, keepaliveFunc, nil
+}
+
+func (e *Etcd) client() (*clientv3.Client, error) {
+	urls := e.e.Server.Cluster().ClientURLs()
+	cli, err := clientv3.NewFromURLs(urls)
+	if err != nil {
+		return nil, errors.Wrapf(err, "creates a new etcd client from URLs (%v)", urls)
+	}
+	return cli, nil
+}
+
+func memberList(cli *clientv3.Client) (ids []uint64, names []string, urls []string) {
+	ml, err := cli.MemberList(context.TODO())
+	if err != nil {
+		panic(err)
+	}
+	n := len(ml.Members)
+	ids = make([]uint64, n)
+	names = make([]string, n)
+	urls = make([]string, n)
+
+	for i, m := range ml.Members {
+		ids[i], names[i], urls[i] = m.ID, m.Name, m.PeerURLs[0]
+	}
+	return
+}
+
+func memberAdd(cli *clientv3.Client, peerURL string) (id uint64, name string) {
+	ma, err := cli.MemberAdd(context.TODO(), []string{peerURL})
+	if err != nil {
+		return 0, ""
+	}
+
+	return ma.Member.ID, ma.Member.Name
+}
+
+// Shards implements the Sharder interface.
+func (e *Etcd) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "Shards: creating client")
+	}
+	defer cli.Close()
+
+	return e.shards(ctx, cli, index, field)
+}
+
+func (e *Etcd) shards(ctx context.Context, cli *clientv3.Client, index, field string) (*roaring.Bitmap, error) {
+	key := path.Join(shardPrefix, index, field)
+
+	// Get the current shards for the field.
+	resp, err := cli.Get(ctx, key)
+	if err != nil {
+		return nil, err
+	}
+
+	bm := roaring.NewBitmap()
+
+	if len(resp.Kvs) == 0 {
+		return bm, nil
+	}
+
+	bytes := resp.Kvs[0].Value
+	if err = bm.UnmarshalBinary(bytes); err != nil {
+		return nil, errors.Wrap(err, "unmarshalling shards")
+	}
+
+	return bm, nil
+}
+
+// AddShards implements the Sharder interface.
+func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roaring.Bitmap) (*roaring.Bitmap, error) {
+	cli, err := e.client()
+	if err != nil {
+		return nil, errors.Wrap(err, "AddShards: creating client")
+	}
+	defer cli.Close()
+
+	key := path.Join(shardPrefix, index, field)
+
+	// This tended to add more overhead than it saved.
+	// // Read shards outside of a lock just to check if shard is already included.
+	// // If shard is already included, no-op.
+	// if currentShards, err := e.shards(ctx, cli, index, field); err != nil {
+	// 	return nil, errors.Wrap(err, "reading shards")
+	// } else if currentShards.Count() == currentShards.Union(shards).Count() {
+	// 	return currentShards, nil
+	// }
+
+	// Create a session to acquire a lock.
+	sess, _ := concurrency.NewSession(cli)
+	defer sess.Close()
+
+	muKey := path.Join(lockPrefix, index, field)
+	mu := concurrency.NewMutex(sess, muKey)
+
+	// Acquire lock (or wait to have it).
+	if err := mu.Lock(ctx); err != nil {
+		return nil, errors.Wrap(err, "acquiring lock")
+	}
+
+	// Read shards within lock.
+	globalShards, err := e.shards(ctx, cli, index, field)
+	if err != nil {
+		return nil, errors.Wrap(err, "reading shards")
+	}
+
+	// Union shard into shards.
+	globalShards.UnionInPlace(shards)
+
+	// Write shards to etcd.
+	var buf bytes.Buffer
+	if _, err := globalShards.WriteTo(&buf); err != nil {
+		return nil, errors.Wrap(err, "writing shards to bytes buffer")
+	}
+
+	op := clientv3.OpPut(key, "")
+	op.WithValueBytes(buf.Bytes())
+
+	if _, err := cli.Do(ctx, op); err != nil {
+		return nil, errors.Wrap(err, "doing op")
+	}
+
+	// Release lock.
+	if err := mu.Unlock(ctx); err != nil {
+		return nil, errors.Wrap(err, "releasing lock")
+	}
+
+	return globalShards, nil
+}
+
+// AddShard implements the Sharder interface.
+func (e *Etcd) AddShard(ctx context.Context, index, field string, shard uint64) error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "AddShard: creating client")
+	}
+	defer cli.Close()
+
+	key := path.Join(shardPrefix, index, field)
+
+	// Read shards outside of a lock just to check if shard is already included.
+	// If shard is already included, no-op.
+	if shards, err := e.shards(ctx, cli, index, field); err != nil {
+		return errors.Wrap(err, "reading shards")
+	} else if shards.Contains(shard) {
+		return nil
+	}
+
+	// According to the previous read, shard is not yet included in shards. So
+	// we will acquire a distributed lock, read shards again (in case it has
+	// been updated since we last read it), add shard to shards, and finally
+	// write shards to etcd.
+
+	// Create a session to acquire a lock.
+	sess, _ := concurrency.NewSession(cli)
+	defer sess.Close()
+
+	muKey := path.Join(lockPrefix, index, field)
+	mu := concurrency.NewMutex(sess, muKey)
+
+	// Acquire lock (or wait to have it).
+	if err := mu.Lock(ctx); err != nil {
+		return errors.Wrap(err, "acquiring lock")
+	}
+
+	// Read shards again (within lock).
+	shards, err := e.shards(ctx, cli, index, field)
+	if err != nil {
+		return errors.Wrap(err, "reading shards")
+	}
+
+	if shards.Contains(shard) {
+		return nil
+	}
+
+	// Union shard into shards.
+	shards.UnionInPlace(roaring.NewBitmap(shard))
+
+	// Write shards to etcd.
+	var buf bytes.Buffer
+	if _, err := shards.WriteTo(&buf); err != nil {
+		return errors.Wrap(err, "writing shards to bytes buffer")
+	}
+
+	op := clientv3.OpPut(key, "")
+	op.WithValueBytes(buf.Bytes())
+
+	if _, err := cli.Do(ctx, op); err != nil {
+		return errors.Wrap(err, "doing op")
+	}
+
+	// Release lock.
+	if err := mu.Unlock(ctx); err != nil {
+		return errors.Wrap(err, "releasing lock")
+	}
+
+	return nil
+}
+
+// RemoveShard implements the Sharder interface.
+func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint64) error {
+	cli, err := e.client()
+	if err != nil {
+		return errors.Wrap(err, "RemoveShard: creating client")
+	}
+	defer cli.Close()
+
+	key := path.Join(shardPrefix, index, field)
+
+	// Read shards outside of a lock just to check if shard is already excluded.
+	// If shard is already excluded, no-op.
+	if shards, err := e.shards(ctx, cli, index, field); err != nil {
+		return errors.Wrap(err, "reading shards")
+	} else if !shards.Contains(shard) {
+		return nil
+	}
+
+	// According to the previous read, shard is included in shards. So
+	// we will acquire a distributed lock, read shards again (in case it has
+	// been updated since we last read it), remove shard from shards, and finally
+	// write shards to etcd.
+
+	// Create a session to acquire a lock.
+	sess, _ := concurrency.NewSession(cli)
+	defer sess.Close()
+
+	muKey := path.Join(lockPrefix, index, field)
+	mu := concurrency.NewMutex(sess, muKey)
+
+	// Acquire lock (or wait to have it).
+	if err := mu.Lock(ctx); err != nil {
+		return errors.Wrap(err, "acquiring lock")
+	}
+
+	// Read shards again (within lock).
+	shards, err := e.shards(ctx, cli, index, field)
+	if err != nil {
+		return errors.Wrap(err, "reading shards")
+	}
+
+	if !shards.Contains(shard) {
+		return nil
+	}
+
+	// Remove shard from shards.
+	if _, err := shards.RemoveN(shard); err != nil {
+		return errors.Wrap(err, "removing shard")
+	}
+
+	// If this is removing the last bit from the shards bitmap, then instead of
+	// writing an empty bitmap, just delete the key.
+	if shards.Count() == 0 {
+		_, err := cli.Delete(ctx, key)
+		return err
+	}
+
+	// Write shards to etcd.
+	var buf bytes.Buffer
+	if _, err := shards.WriteTo(&buf); err != nil {
+		return errors.Wrap(err, "writing shards to bytes buffer")
+	}
+
+	op := clientv3.OpPut(key, "")
+	op.WithValueBytes(buf.Bytes())
+
+	if _, err := cli.Do(ctx, op); err != nil {
+		return errors.Wrap(err, "doing op")
+	}
+
+	// Release lock.
+	if err := mu.Unlock(ctx); err != nil {
+		return errors.Wrap(err, "releasing lock")
+	}
+
+	return nil
+}
+
+var _ pilosa.Noder = &EtcdWrapper{}
+
+// EtcdWrapper is a wrapper around the imported Etcd. Once we are no long
+// importing Etcd from etcd-test, and instead have it here in the pilosa/etcd
+// package, we can get rid of the wrapper. It's here so that we can implement
+// the Noder interface without having to do that in the etcd-test repo.
+type EtcdWrapper struct {
+	*etcd.EtcdWithCache
+}
+
+// NewEtcd returns a new instance of a wrapped Etcd.
+func NewEtcd(opt etcd.Options, replicas int) *EtcdWrapper {
+	return &EtcdWrapper{
+		EtcdWithCache: etcd.NewEtcdWithCache(opt, replicas),
+	}
+}
+
+// Nodes implements the Noder interface.
+func (e *EtcdWrapper) Nodes() []*pilosa.Node {
+	// If we have looked up nodes within a certain time, then we're going to
+	// use the cached value for now. This is temporary and will be addressed
+	// correctly in #1133.
+	peers := e.Peers()
+	nodes := make([]*pilosa.Node, len(peers))
+	for i, peer := range peers {
+		node := &pilosa.Node{}
+		if meta, err := e.Metadata(context.Background(), peer.ID); err != nil {
+			log.Println(err, "getting metadata") // TODO: handle this with a logger
+		} else if err := json.Unmarshal(meta, node); err != nil {
+			log.Println(err, "unmarshaling json metadata")
+		}
+
+		node.ID = peer.ID
+
+		nodes[i] = node
+	}
+
+	// Nodes must be sorted.
+	sort.Sort(byID(nodes))
+
+	return nodes
+}
+
+// byID implements sort.Interface for []*pilosa.Node based on
+// the ID field.
+type byID []*pilosa.Node
+
+func (h byID) Len() int           { return len(h) }
+func (h byID) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
+func (h byID) Less(i, j int) bool { return h[i].ID < h[j].ID }
+
+// SetNodes implements the Noder interface.
+func (e *EtcdWrapper) SetNodes(nodes []*pilosa.Node) {}
+
+// AppendNode implements the Noder interface.
+func (e *EtcdWrapper) AppendNode(node *pilosa.Node) {}
+
+// RemoveNode implements the Noder interface.
+func (e *EtcdWrapper) RemoveNode(nodeID string) bool {
+	return false
+}
diff --git a/net/uri.go b/net/uri.go
new file mode 100644
index 000000000..d83f3b456
--- /dev/null
+++ b/net/uri.go
@@ -0,0 +1,231 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package net
+
+import (
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/url"
+	"regexp"
+	"strconv"
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+var schemeRegexp = regexp.MustCompile("^[+a-z]+$")
+var hostRegexp = regexp.MustCompile(`^[0-9a-z.-]+$|^\[[:0-9a-fA-F]+\]$`)
+var addressRegexp = regexp.MustCompile(`^(([+a-z]+):\/\/)?([0-9a-z.-]+|\[[:0-9a-fA-F]+\])?(:([0-9]+))?$`)
+
+// URI represents a Pilosa URI.
+// A Pilosa URI consists of three parts:
+// 1) Scheme: Protocol of the URI. Default: http.
+// 2) Host: Hostname or IP URI. Default: localhost. IPv6 addresses should be written in brackets, e.g., `[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]`.
+// 3) Port: Port of the URI. Default: 10101.
+//
+// All parts of the URI are optional. The following are equivalent:
+// 	http://localhost:10101
+// 	http://localhost
+// 	http://:10101
+// 	localhost:10101
+// 	localhost
+// 	:10101
+type URI struct {
+	Scheme string `json:"scheme"`
+	Host   string `json:"host"`
+	Port   uint16 `json:"port"`
+}
+
+// URL returns a url.URL representation of the URI.
+func (u *URI) URL() url.URL {
+	return url.URL{Scheme: u.Scheme, Host: net.JoinHostPort(u.Host, strconv.Itoa(int(u.Port)))}
+}
+
+// DefaultURI creates and returns the default URI.
+func DefaultURI() *URI {
+	return defaultURI()
+}
+
+// defaultURI creates and returns the default URI.
+func defaultURI() *URI {
+	return &URI{
+		Scheme: "http",
+		Host:   "localhost",
+		Port:   10101,
+	}
+}
+
+// URIs is a convenience type representing a slice of URI.
+type URIs []URI
+
+// HostPortStrings returns a slice of host:port strings
+// based on the slice of URI.
+func (u URIs) HostPortStrings() []string {
+	s := make([]string, len(u))
+	for i, a := range u {
+		s[i] = a.HostPort()
+	}
+	return s
+}
+
+// NewURIFromHostPort returns a URI with specified host and port.
+func NewURIFromHostPort(host string, port uint16) (*URI, error) {
+	uri := defaultURI()
+	err := uri.SetHost(host)
+	if err != nil {
+		return nil, errors.Wrap(err, "setting uri host")
+	}
+	uri.SetPort(port)
+	return uri, nil
+}
+
+// NewURIFromAddress parses the passed address and returns a URI.
+func NewURIFromAddress(address string) (*URI, error) {
+	return parseAddress(address)
+}
+
+// SetScheme sets the scheme of this URI.
+func (u *URI) SetScheme(scheme string) error {
+	m := schemeRegexp.FindStringSubmatch(scheme)
+	if m == nil {
+		return errors.New("invalid scheme")
+	}
+	u.Scheme = scheme
+	return nil
+}
+
+// SetHost sets the host of this URI.
+func (u *URI) SetHost(host string) error {
+	m := hostRegexp.FindStringSubmatch(host)
+	if m == nil {
+		return errors.New("invalid host")
+	}
+	u.Host = host
+	return nil
+}
+
+// SetPort sets the port of this URI.
+func (u *URI) SetPort(port uint16) {
+	u.Port = port
+}
+
+// HostPort returns `Host:Port`
+func (u *URI) HostPort() string {
+	// XXX: The following is just to make TestHandler_Status; remove it
+	if u == nil {
+		return ""
+	}
+	s := fmt.Sprintf("%s:%d", u.Host, u.Port)
+	return s
+}
+
+// normalize returns the address in a form usable by a HTTP client.
+func (u *URI) normalize() string {
+	scheme := u.Scheme
+	index := strings.Index(scheme, "+")
+	if index >= 0 {
+		scheme = scheme[:index]
+	}
+	return fmt.Sprintf("%s://%s:%d", scheme, u.Host, u.Port)
+}
+
+// String returns the address as a string.
+func (u URI) String() string {
+	return fmt.Sprintf("%s://%s:%d", u.Scheme, u.Host, u.Port)
+}
+
+// Path returns URI with path
+func (u *URI) Path(path string) string {
+	return fmt.Sprintf("%s%s", u.normalize(), path)
+}
+
+// The following methods are required to implement pflag Value interface.
+
+// Set sets the uri value.
+func (u *URI) Set(value string) error {
+	uri, err := NewURIFromAddress(value)
+	if err != nil {
+		return err
+	}
+	*u = *uri
+	return nil
+}
+
+// Type returns the type of a uri.
+func (u URI) Type() string {
+	return "URI"
+}
+
+func parseAddress(address string) (uri *URI, err error) {
+	m := addressRegexp.FindStringSubmatch(address)
+	if m == nil {
+		return nil, errors.New("invalid address")
+	}
+	scheme := "http"
+	if m[2] != "" {
+		scheme = m[2]
+	}
+	host := "localhost"
+	if m[3] != "" {
+		host = m[3]
+	}
+	var port = 10101
+	if m[5] != "" {
+		port, err = strconv.Atoi(m[5])
+		if err != nil {
+			return nil, errors.New("converting port string to int")
+		}
+		if port > 65535 {
+			return nil, errors.New("port must be in range 0 - 65535")
+		}
+	}
+	uri = &URI{
+		Scheme: scheme,
+		Host:   host,
+		Port:   uint16(port),
+	}
+	return uri, nil
+}
+
+// MarshalJSON marshals URI into a JSON-encoded byte slice.
+func (u *URI) MarshalJSON() ([]byte, error) {
+	var output struct {
+		Scheme string `json:"scheme,omitempty"`
+		Host   string `json:"host,omitempty"`
+		Port   uint16 `json:"port,omitempty"`
+	}
+	output.Scheme = u.Scheme
+	output.Host = u.Host
+	output.Port = u.Port
+
+	return json.Marshal(output)
+}
+
+// UnmarshalJSON unmarshals a byte slice to a URI.
+func (u *URI) UnmarshalJSON(b []byte) error {
+	var input struct {
+		Scheme string `json:"scheme,omitempty"`
+		Host   string `json:"host,omitempty"`
+		Port   uint16 `json:"port,omitempty"`
+	}
+	if err := json.Unmarshal(b, &input); err != nil {
+		return err
+	}
+	u.Scheme = input.Scheme
+	u.Host = input.Host
+	u.Port = input.Port
+	return nil
+}
diff --git a/net/uri_internal_test.go b/net/uri_internal_test.go
new file mode 100644
index 000000000..3cedc30ed
--- /dev/null
+++ b/net/uri_internal_test.go
@@ -0,0 +1,176 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package net
+
+import "testing"
+
+func TestDefaultURI(t *testing.T) {
+	uri := defaultURI()
+	compare(t, uri, "http", "localhost", 10101)
+}
+
+func TestURIWithHostPort(t *testing.T) {
+	uri, err := NewURIFromHostPort("index1.pilosa.com", 3333)
+	if err != nil {
+		t.Fatal(err)
+	}
+	compare(t, uri, "http", "index1.pilosa.com", 3333)
+}
+
+func TestURIWithInvalidHostPort(t *testing.T) {
+	_, err := NewURIFromHostPort("index?.pilosa.com", 3333)
+	if err == nil {
+		t.Fatalf("should have failed")
+	}
+}
+
+func TestNewURIFromAddress(t *testing.T) {
+	for _, item := range validFixture() {
+		uri, err := NewURIFromAddress(item.address)
+		if err != nil {
+			t.Fatalf("Can't parse address: %s, %s", item.address, err)
+		}
+		compare(t, uri, item.scheme, item.host, item.port)
+	}
+}
+
+func TestNewURIFromAddressInvalidAddress(t *testing.T) {
+	for _, addr := range invalidFixture() {
+		_, err := NewURIFromAddress(addr)
+		if err == nil {
+			t.Fatalf("Invalid address should return an error: %s", addr)
+		}
+	}
+}
+
+func TestNormalizedAddress(t *testing.T) {
+	uri, err := NewURIFromAddress("http+protobuf://big-data.pilosa.com:6888")
+	if err != nil {
+		t.Fatalf("Can't parse address")
+	}
+	if uri.normalize() != "http://big-data.pilosa.com:6888" {
+		t.Fatalf("Normalized address is not normal")
+	}
+}
+
+func TestURIPath(t *testing.T) {
+	uri, err := NewURIFromAddress("http+protobuf://big-data.pilosa.com:6888")
+	if err != nil {
+		t.Fatal(err)
+	}
+	target := "http://big-data.pilosa.com:6888/index/foo"
+	if uri.Path("/index/foo") != target {
+		t.Fatalf("%s != %s", uri.Path("/index/foo"), target)
+	}
+}
+
+func TestSetScheme(t *testing.T) {
+	uri := defaultURI()
+	target := "fun"
+	err := uri.SetScheme(target)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if uri.Scheme != target {
+		t.Fatalf("%s != %s", uri.Scheme, target)
+	}
+}
+
+func TestSetHost(t *testing.T) {
+	uri := defaultURI()
+	target := "10.20.30.40"
+	err := uri.SetHost(target)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if uri.Host != target {
+		t.Fatalf("%s != %s", uri.Host, target)
+	}
+}
+
+func TestSetPort(t *testing.T) {
+	uri := defaultURI()
+	target := uint16(9999)
+	uri.SetPort(target)
+	if uri.Port != target {
+		t.Fatalf("%d != %d", uri.Port, target)
+	}
+}
+
+func TestSetInvalidScheme(t *testing.T) {
+	uri := defaultURI()
+	err := uri.SetScheme("?invalid")
+	if err == nil {
+		t.Fatalf("Should have failed")
+	}
+}
+
+func TestSetInvalidHost(t *testing.T) {
+	uri := defaultURI()
+	err := uri.SetHost("index?.pilosa.com")
+	if err == nil {
+		t.Fatalf("Should have failed")
+	}
+}
+
+func TestHostPort(t *testing.T) {
+	uri, err := NewURIFromHostPort("i.pilosa.com", 15001)
+	if err != nil {
+		t.Fatal(err)
+	}
+	target := "i.pilosa.com:15001"
+	if uri.HostPort() != target {
+		t.Fatalf("%s != %s", uri.HostPort(), target)
+	}
+}
+
+func compare(t *testing.T, uri *URI, scheme string, host string, port uint16) {
+	if uri.Scheme != scheme {
+		t.Fatalf("Scheme does not match: %s != %s", uri.Scheme, scheme)
+	}
+	if uri.Host != host {
+		t.Fatalf("Host does not match: %s != %s", uri.Host, host)
+	}
+	if uri.Port != port {
+		t.Fatalf("Port does not match: %d != %d", uri.Port, port)
+	}
+}
+
+type uriItem struct {
+	address string
+	scheme  string
+	host    string
+	port    uint16
+}
+
+func validFixture() []uriItem {
+	var test = []uriItem{
+		{"http+protobuf://index1.pilosa.com:3333", "http+protobuf", "index1.pilosa.com", 3333},
+		{"index1.pilosa.com:3333", "http", "index1.pilosa.com", 3333},
+		{"https://index1.pilosa.com", "https", "index1.pilosa.com", 10101},
+		{"index1.pilosa.com", "http", "index1.pilosa.com", 10101},
+		{"https://:3333", "https", "localhost", 3333},
+		{":3333", "http", "localhost", 3333},
+		{"[::1]", "http", "[::1]", 10101},
+		{"[::1]:3333", "http", "[::1]", 3333},
+		{"[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]:3333", "http", "[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]", 3333},
+		{"https://[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]:3333", "https", "[fd42:4201:f86b:7e09:216:3eff:fefa:ed80]", 3333},
+	}
+	return test
+}
+
+func invalidFixture() []string {
+	return []string{"foo:bar", "http://foo:", "foo:", ":bar", "http://pilosa.com:129999999999999999999999993", "fd42:4201:f86b:7e09:216:3eff:fefa:ed80", ":65536"}
+}
diff --git a/topology/hasher.go b/topology/hasher.go
new file mode 100644
index 000000000..a5c3f5964
--- /dev/null
+++ b/topology/hasher.go
@@ -0,0 +1,41 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package topology
+
+// Hasher represents an interface to hash integers into buckets.
+type Hasher interface {
+	// Hashes the key into a number between [0,N).
+	Hash(key uint64, n int) int
+	Name() string
+}
+
+// Jmphasher represents an implementation of jmphash. Implements Hasher.
+type Jmphasher struct{}
+
+// Hash returns the integer hash for the given key.
+func (h *Jmphasher) Hash(key uint64, n int) int {
+	b, j := int64(-1), int64(0)
+	for j < int64(n) {
+		b = j
+		key = key*uint64(2862933555777941757) + 1
+		j = int64(float64(b+1) * (float64(int64(1)<<31) / float64((key>>33)+1)))
+	}
+	return int(b)
+}
+
+// Name returns the name of this hash.
+func (h *Jmphasher) Name() string {
+	return "jump-hash"
+}
diff --git a/topology/node.go b/topology/node.go
new file mode 100644
index 000000000..dd46f6207
--- /dev/null
+++ b/topology/node.go
@@ -0,0 +1,142 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package topology
+
+import (
+	"fmt"
+
+	"github.com/pilosa/pilosa/v2/net"
+)
+
+// Node represents a node in the cluster.
+type Node struct {
+	ID            string  `json:"id"`
+	URI           net.URI `json:"uri"`
+	GRPCURI       net.URI `json:"grpc-uri"`
+	IsCoordinator bool    `json:"isCoordinator"`
+	State         string  `json:"state"`
+}
+
+func (n *Node) Clone() *Node {
+	if n == nil {
+		return nil
+	}
+	other := *n
+	return &other
+}
+
+func (n Node) String() string {
+	return fmt.Sprintf("Node:%s:%s:%s", n.URI, n.State, n.ID)
+}
+
+// Nodes represents a list of nodes.
+type Nodes []*Node
+
+// Contains returns true if a node exists in the list.
+func (a Nodes) Contains(n *Node) bool {
+	for i := range a {
+		if a[i] == n {
+			return true
+		}
+	}
+	return false
+}
+
+// ContainsID returns true if host matches one of the node's id.
+func (a Nodes) ContainsID(id string) bool {
+	for _, n := range a {
+		if n.ID == id {
+			return true
+		}
+	}
+	return false
+}
+
+// NodeByID returns the node for an ID. If the ID is not found,
+// it returns nil.
+func (a Nodes) NodeByID(id string) *Node {
+	for _, n := range a {
+		if n.ID == id {
+			return n
+		}
+	}
+	return nil
+}
+
+// Filter returns a new list of nodes with node removed.
+func (a Nodes) Filter(n *Node) []*Node {
+	other := make([]*Node, 0, len(a))
+	for i := range a {
+		if a[i] != n {
+			other = append(other, a[i])
+		}
+	}
+	return other
+}
+
+// FilterID returns a new list of nodes with ID removed.
+func (a Nodes) FilterID(id string) []*Node {
+	other := make([]*Node, 0, len(a))
+	for _, node := range a {
+		if node.ID != id {
+			other = append(other, node)
+		}
+	}
+	return other
+}
+
+// FilterURI returns a new list of nodes with URI removed.
+func (a Nodes) FilterURI(uri net.URI) []*Node {
+	other := make([]*Node, 0, len(a))
+	for _, node := range a {
+		if node.URI != uri {
+			other = append(other, node)
+		}
+	}
+	return other
+}
+
+// IDs returns a list of all node IDs.
+func (a Nodes) IDs() []string {
+	ids := make([]string, len(a))
+	for i, n := range a {
+		ids[i] = n.ID
+	}
+	return ids
+}
+
+// URIs returns a list of all uris.
+func (a Nodes) URIs() []net.URI {
+	uris := make([]net.URI, len(a))
+	for i, n := range a {
+		uris[i] = n.URI
+	}
+	return uris
+}
+
+// Clone returns a shallow copy of nodes.
+func (a Nodes) Clone() []*Node {
+	other := make([]*Node, len(a))
+	copy(other, a)
+	return other
+}
+
+// ByID implements sort.Interface for []Node based on
+// the ID field.
+type ByID []*Node
+
+func (h ByID) Len() int           { return len(h) }
+func (h ByID) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
+func (h ByID) Less(i, j int) bool { return h[i].ID < h[j].ID }
diff --git a/topology/noder.go b/topology/noder.go
new file mode 100644
index 000000000..d6dff517a
--- /dev/null
+++ b/topology/noder.go
@@ -0,0 +1,74 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package topology
+
+import (
+	"sort"
+)
+
+// Noder is an interface which abstracts the Node slice so that the list of
+// nodes in a cluster can be maintained outside of the cluster struct.
+type Noder interface {
+	Nodes() []*Node // Remember: this has to be sorted correctly!!
+	SetNodes([]*Node)
+	AppendNode(*Node)
+	RemoveNode(nodeID string) bool
+}
+
+// localNoder is a simple implementation of the Noder interface
+// which maintains an instance of the `nodes` slice.
+type localNoder struct {
+	nodes []*Node
+}
+
+// NewLocalNoder is a helper function for wrapping an existing slice of Nodes
+// with something which implements Noder.
+func NewLocalNoder(nodes []*Node) *localNoder {
+	return &localNoder{
+		nodes: nodes,
+	}
+}
+
+// Nodes implements the Noder interface.
+func (n *localNoder) Nodes() []*Node {
+	return n.nodes
+}
+
+// SetNodes implements the Noder interface.
+func (n *localNoder) SetNodes(nodes []*Node) {
+	n.nodes = nodes
+}
+
+// AppendNode implements the Noder interface.
+func (n *localNoder) AppendNode(node *Node) {
+	n.nodes = append(n.nodes, node)
+
+	// All hosts must be merged in the same order on all nodes in the cluster.
+	sort.Sort(ByID(n.nodes))
+}
+
+// RemoveNode implements the Noder interface.
+func (n *localNoder) RemoveNode(nodeID string) bool {
+	i := NodePositionByID(n.nodes, nodeID)
+	if i < 0 {
+		return false
+	}
+
+	copy(n.nodes[i:], n.nodes[i+1:])
+	n.nodes[len(n.nodes)-1] = nil
+	n.nodes = n.nodes[:len(n.nodes)-1]
+
+	return true
+}
diff --git a/topology/snapshot.go b/topology/snapshot.go
new file mode 100644
index 000000000..2eaa7b0b9
--- /dev/null
+++ b/topology/snapshot.go
@@ -0,0 +1,272 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package topology
+
+import (
+	"encoding/binary"
+	"hash/fnv"
+
+	"github.com/pilosa/pilosa/v2/roaring"
+	"github.com/pilosa/pilosa/v2/shardwidth"
+)
+
+const (
+	// DefaultPartitionN is the default number of partitions in a cluster.
+	DefaultPartitionN = 256
+
+	// ShardWidth is the number of column IDs in a shard. It must be a power of 2 greater than or equal to 16.
+	// shardWidthExponent = 20 // set in shardwidthNN.go files
+	ShardWidth = 1 << shardwidth.Exponent
+)
+
+// ClusterSnapshot is a static representation of a cluster and its nodes. It is
+// used to calculate things like partition location and data distribution.
+type ClusterSnapshot struct {
+	Nodes []*Node
+
+	// Hashing algorithm used to assign partitions to nodes.
+	Hasher Hasher
+
+	// The number of partitions in the cluster.
+	PartitionN int
+
+	// The number of replicas a partition has.
+	ReplicaN int
+}
+
+// NewClusterSnapshot returns a new instance of ClusterSnapshot.
+func NewClusterSnapshot(noder Noder, hasher Hasher, replicas int) *ClusterSnapshot {
+	nodes := noder.Nodes()
+
+	// Make sure replica count doesn't exceed the number of nodes.
+	nodeN := len(nodes)
+	if replicas > nodeN {
+		replicas = nodeN
+	} else if replicas == 0 {
+		replicas = 1
+	}
+
+	return &ClusterSnapshot{
+		Nodes:      nodes,
+		Hasher:     hasher,
+		PartitionN: DefaultPartitionN,
+		ReplicaN:   replicas,
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////////
+
+// shardToShardPartition returns the shard-partition that the given shard
+// belongs to. NOTE: This is DIFFERENT from the key-partition.
+func (c *ClusterSnapshot) shardToShardPartition(index string, shard uint64) int {
+	return dedupShardToShardPartition(index, shard, c.PartitionN)
+}
+
+// dedupShardToShardParition would ideally be called `shardToShardPartition`, but since
+// we can't put this into it's own package yet (see the TODO below about import loops),
+// that name conflicts with a function that already exists in the `pilosa` package.
+func dedupShardToShardPartition(index string, shard uint64, partitionN int) int {
+	var buf [8]byte
+	binary.BigEndian.PutUint64(buf[:], shard)
+
+	// Hash the bytes and mod by partition count.
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(index))
+	_, _ = h.Write(buf[:])
+	return int(h.Sum64() % uint64(partitionN))
+}
+
+// keyToKeyPartition returns the key-partition that the given key belongs to.
+// NOTE: The key-partition is DIFFERENT from the shard-partition.
+func (c *ClusterSnapshot) keyToKeyPartition(index, key string) int {
+	// Hash the bytes and mod by partition count.
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(index))
+	_, _ = h.Write([]byte(key))
+	return int(h.Sum64() % uint64(c.PartitionN))
+}
+
+// ShardNodes returns a list of nodes that own a shard.
+func (c *ClusterSnapshot) ShardNodes(index string, shard uint64) []*Node {
+	return c.PartitionNodes(c.shardToShardPartition(index, shard))
+}
+
+// KeyNodes returns a list of nodes that own a key.
+func (c *ClusterSnapshot) KeyNodes(index, key string) []*Node {
+	return c.PartitionNodes(c.keyToKeyPartition(index, key))
+}
+
+// PartitionNodes returns a list of nodes that own the given partition.
+func (c *ClusterSnapshot) PartitionNodes(partitionID int) []*Node {
+	// Determine primary owner node.
+	nodeIndex := c.PrimaryNodeIndex(partitionID)
+	if nodeIndex < 0 {
+		// no nodes anyway
+		return nil
+	}
+	// Collect nodes around the ring.
+	nodes := make([]*Node, 0, c.ReplicaN)
+	for i := 0; i < c.ReplicaN; i++ {
+		nodes = append(nodes, c.Nodes[(nodeIndex+i)%len(c.Nodes)])
+	}
+
+	return nodes
+}
+
+// PrimaryFieldTranslationNode is the primary node responsible for translating
+// field keys. The primary could be any node in the cluster, but we arbitrarily
+// define it to be the node responsible for partition 0.
+func (c *ClusterSnapshot) PrimaryFieldTranslationNode() *Node {
+	return c.PrimaryPartitionNode(0)
+}
+
+// IsPrimaryFieldTranslationNode returns true if nodeID represents the primary
+// node responsible for field translation.
+func (c *ClusterSnapshot) IsPrimaryFieldTranslationNode(nodeID string) bool {
+	return c.PrimaryFieldTranslationNode().ID == nodeID
+}
+
+// PrimaryPartitionNode returns the primary node of the given partition.
+func (c *ClusterSnapshot) PrimaryPartitionNode(partition int) *Node {
+	if nodes := c.PartitionNodes(partition); len(nodes) > 0 {
+		return nodes[0]
+	}
+	return nil
+}
+
+// IsPrimary returns true if the given node is the primary for the given
+// partition.
+func (c *ClusterSnapshot) IsPrimary(nodeID string, partition int) bool {
+	primary := c.PrimaryNodeIndex(partition)
+	return nodeID == c.Nodes[primary].ID
+}
+
+// PrimaryNodeIndex returns the index (position in the cluster) of the primary
+// node for the given partition.
+func (c *ClusterSnapshot) PrimaryNodeIndex(partition int) int {
+	return c.Hasher.Hash(uint64(partition), len(c.Nodes))
+}
+
+// NonPrimaryReplicas returns the list of node IDs which are replicas for the
+// given partition.
+func (c *ClusterSnapshot) NonPrimaryReplicas(partition int) (nonPrimaryReplicas []string) {
+	primary := c.PrimaryNodeIndex(partition)
+	nodeN := len(c.Nodes)
+
+	// Collect nodes around the ring.
+	for i := 1; i < nodeN; i++ {
+		node := c.Nodes[(primary+i)%nodeN]
+		if i < c.ReplicaN {
+			nonPrimaryReplicas = append(nonPrimaryReplicas, node.ID)
+		}
+	}
+	return
+}
+
+// ReplicasForPrimary returns the map replicaNodeIDs[nodeID] which will have a
+// true value for the primary nodeID, and false for others.
+func (c *ClusterSnapshot) ReplicasForPrimary(primary int) (replicaNodeIDs, nonReplicas map[string]bool) {
+	if primary < 0 {
+		// no nodes anyway
+		return
+	}
+	replicaNodeIDs = make(map[string]bool)
+	nonReplicas = make(map[string]bool)
+
+	nodeN := len(c.Nodes)
+
+	// Collect nodes around the ring.
+	for i := 0; i < nodeN; i++ {
+		node := c.Nodes[(primary+i)%nodeN]
+		if i < c.ReplicaN {
+			// mark true if primary
+			replicaNodeIDs[node.ID] = (i == 0)
+		} else {
+			nonReplicas[node.ID] = false
+		}
+	}
+	return
+}
+
+// ContainsShards is like OwnsShards, but it includes replicas.
+func (c *ClusterSnapshot) ContainsShards(index string, availableShards *roaring.Bitmap, node *Node) []uint64 {
+	var shards []uint64
+	_ = availableShards.ForEach(func(i uint64) error {
+		p := c.shardToShardPartition(index, i)
+		// Determine the nodes for partition.
+		nodes := c.PartitionNodes(p)
+		for _, n := range nodes {
+			if n.ID == node.ID {
+				shards = append(shards, i)
+			}
+		}
+		return nil
+	})
+	return shards
+}
+
+// TODO: update this comment
+// The boltdb key translation stores are partitioned, designated by partitionIDs. These
+// are shared between replicas, and one node is the primary for
+// replication. So with 4 nodes and 3-way replication, each node has 3/4 of
+// the translation stores on it.
+func (c *ClusterSnapshot) PrimaryForColKeyTranslation(index, key string) (primary int) {
+	partitionID := c.keyToKeyPartition(index, key)
+	return c.PrimaryNodeIndex(partitionID)
+}
+
+// TODO: update this comment
+// should match cluster.go:1033 cluster.ownsShard(nodeID, index, shard)
+// 	return Nodes(c.shardNodes(index, shard)).ContainsID(nodeID)
+func (c *ClusterSnapshot) PrimaryForShardReplication(index string, shard uint64) int {
+	n := len(c.Nodes)
+	if n == 0 {
+		return -1
+	}
+	partition := uint64(dedupShardToShardPartition(index, shard, c.PartitionN))
+	nodeIndex := c.Hasher.Hash(partition, n)
+	return nodeIndex
+}
+
+// PrimaryReplicaNode returns the node listed before the current node in Nodes().
+// This is different than "previous node" as the first node always returns nil.
+func (c *ClusterSnapshot) PrimaryReplicaNode(nodeID string) *Node {
+	pos := c.nodePositionByID(nodeID)
+	if pos <= 0 {
+		return nil
+	}
+	return c.Nodes[pos-1]
+}
+
+// nodePositionByID returns the position of the node in slice c.Nodes.
+func (c *ClusterSnapshot) nodePositionByID(nodeID string) int {
+	return NodePositionByID(c.Nodes, nodeID)
+}
+
+// NodePositionByID returns the position of the node in slice nodes.
+// TODO: this is exported because it's used in noder.go. Because that's the same
+// package, it doesn't need to be exported, but ideally we could put this
+// snapshot code into its own package. I tried to do that (by putting it into a
+// package called `topology`), but that created an import loop. So what we
+// really need to do is do a better job of creating sub-packages under pilosa
+// (for things like `Noder` and `Nodes`).
+func NodePositionByID(nodes []*Node, nodeID string) int {
+	for i, n := range nodes {
+		if n.ID == nodeID {
+			return i
+		}
+	}
+	return -1
+}