add subpackages: topology, net

2026-09-12 15:51:01 +00:00 · 2021-01-06 13:21:36 -06:00 · 2021-01-06 13:21:36 -06:00 · 9d8a6ad28d
commit 9d8a6ad28d
parent fbf546f131
9 changed files with 2333 additions and 0 deletions
--- a/disco/disco.go
+++ 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
+}
--- a/etcd/cache.go
+++ 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
+}
--- a/etcd/embed.go
+++ b/etcd/embed.go
--- a/net/uri.go
+++ 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
+}
--- a/net/uri_internal_test.go
+++ 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"}
+}
--- a/topology/hasher.go
+++ 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"
+}
--- a/topology/node.go
+++ 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 }
--- a/topology/noder.go
+++ 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
+}
--- a/topology/snapshot.go
+++ 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
+}