Merge pull request #435 from travisturner/messenger-rebase

Messenger rebase
2026-08-28 10:54:59 +00:00 · 2017-04-21 14:04:27 -05:00 · 2017-04-21 14:04:27 -05:00 · ea663b980f
commit ea663b980f
parent 49269a5d90 93f36d0e9a
28 changed files with 3052 additions and 285 deletions
--- a/broadcast.go
+++ b/broadcast.go
@ -0,0 +1,139 @@
+package pilosa
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/pilosa/pilosa/internal"
+)
+
+// NodeSet represents an interface for Node membership and inter-node communication.
+type NodeSet interface {
+	// Returns a list of all Nodes in the cluster
+	Nodes() []*Node
+
+	// Open starts any network activity implemented by the NodeSet
+	Open() error
+}
+
+// StaticNodeSet represents a basic NodeSet for testing
+type StaticNodeSet struct {
+	nodes []*Node
+}
+
+func NewStaticNodeSet() *StaticNodeSet {
+	return &StaticNodeSet{}
+}
+
+func (s *StaticNodeSet) Nodes() []*Node {
+	return s.nodes
+}
+
+func (s *StaticNodeSet) Open() error {
+	return nil
+}
+
+// Broadcaster is an interface for broadcasting messages.
+type Broadcaster interface {
+	SendSync(pb proto.Message) error
+	SendAsync(pb proto.Message) error
+}
+
+func init() {
+	NopBroadcaster = &nopBroadcaster{}
+}
+
+var NopBroadcaster Broadcaster
+
+// nopBroadcaster represents a Broadcaster that doesn't do anything.
+type nopBroadcaster struct{}
+
+// SendSync A no-op implemenetation of Broadcaster SendSync method.
+func (c *nopBroadcaster) SendSync(pb proto.Message) error {
+	return nil
+}
+
+// SendAsync A no-op implemenetation of Broadcaster SendAsync method.
+func (c *nopBroadcaster) SendAsync(pb proto.Message) error {
+	return nil
+}
+
+// BroadcastHandler is the interface for the pilosa object which knows how to
+// handle broadcast messages. (Hint: this is implemented by pilosa.Server)
+type BroadcastHandler interface {
+	ReceiveMessage(pb proto.Message) error
+}
+
+// BroadcastReceiver is the interface for the object which will listen for and
+// decode broadcast messages before passing them to pilosa to handle. The
+// implementation of this could be an http server which listens for messages,
+// gets the protobuf payload, and then passes it to
+// BroadcastHandler.ReceiveMessage.
+type BroadcastReceiver interface {
+	// Start starts listening for broadcast messages - it should return
+	// immediately, spawning a goroutine if necessary.
+	Start(BroadcastHandler) error
+}
+
+type nopBroadcastReceiver struct{}
+
+func (n *nopBroadcastReceiver) Start(b BroadcastHandler) error { return nil }
+
+var NopBroadcastReceiver = &nopBroadcastReceiver{}
+
+const (
+	MessageTypeCreateSlice = 1
+	MessageTypeCreateDB    = 2
+	MessageTypeDeleteDB    = 3
+	MessageTypeCreateFrame = 4
+	MessageTypeDeleteFrame = 5
+)
+
+func MarshalMessage(m proto.Message) ([]byte, error) {
+	var typ uint8
+	switch obj := m.(type) {
+	case *internal.CreateSliceMessage:
+		typ = MessageTypeCreateSlice
+	case *internal.CreateDBMessage:
+		typ = MessageTypeCreateDB
+	case *internal.DeleteDBMessage:
+		typ = MessageTypeDeleteDB
+	case *internal.CreateFrameMessage:
+		typ = MessageTypeCreateFrame
+	case *internal.DeleteFrameMessage:
+		typ = MessageTypeDeleteFrame
+	default:
+		return nil, fmt.Errorf("message type not implemented for marshalling: %s", reflect.TypeOf(obj))
+	}
+	buf, err := proto.Marshal(m)
+	if err != nil {
+		return nil, err
+	}
+	return append([]byte{typ}, buf...), nil
+}
+
+func UnmarshalMessage(buf []byte) (proto.Message, error) {
+	typ, buf := buf[0], buf[1:]
+
+	var m proto.Message
+	switch typ {
+	case MessageTypeCreateSlice:
+		m = &internal.CreateSliceMessage{}
+	case MessageTypeCreateDB:
+		m = &internal.CreateDBMessage{}
+	case MessageTypeDeleteDB:
+		m = &internal.DeleteDBMessage{}
+	case MessageTypeCreateFrame:
+		m = &internal.CreateFrameMessage{}
+	case MessageTypeDeleteFrame:
+		m = &internal.DeleteFrameMessage{}
+	default:
+		return nil, fmt.Errorf("invalid message type: %d", typ)
+	}
+
+	if err := proto.Unmarshal(buf, m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
--- a/broadcast_test.go
+++ b/broadcast_test.go
@ -0,0 +1,91 @@
+package pilosa_test
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/pilosa/pilosa"
+	"github.com/pilosa/pilosa/internal"
+)
+
+// Ensure a message can be marshaled and unmarshaled.
+func TestMessage_Marshal(t *testing.T) {
+
+	testMessageMarshal(t, &internal.CreateSliceMessage{
+		DB:    "d",
+		Slice: 8,
+	})
+
+	testMessageMarshal(t, &internal.DeleteDBMessage{
+		DB: "d",
+	})
+}
+
+func testMessageMarshal(t *testing.T, m proto.Message) {
+	marshalled, err := pilosa.MarshalMessage(m)
+	if err != nil {
+		t.Fatal(err)
+	}
+	unmarshalled, err := pilosa.UnmarshalMessage(marshalled)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(unmarshalled, m) {
+		t.Fatalf("unexpected message marshalling: %s", unmarshalled)
+	}
+}
+
+// Ensure that BroadcastReceiver can register a BroadcastHandler.
+func TestBroadcast_BroadcastReceiver(t *testing.T) {
+
+	s := pilosa.NewServer()
+
+	sbr := NewSimpleBroadcastReceiver()
+	sbh := NewSimpleBroadcastHandler()
+
+	s.BroadcastReceiver = sbr
+	s.BroadcastReceiver.Start(sbh)
+
+	msg := &internal.DeleteDBMessage{
+		DB: "d",
+	}
+
+	s.BroadcastReceiver.(*SimpleBroadcastReceiver).Receive(msg)
+
+	// Make sure the message received is what was sentd
+	if !reflect.DeepEqual(sbh.receivedMessage, msg) {
+		t.Fatalf("unexpected message: %s", sbh.receivedMessage)
+	}
+}
+
+type SimpleBroadcastReceiver struct {
+	broadcastHandler pilosa.BroadcastHandler
+}
+
+func NewSimpleBroadcastReceiver() *SimpleBroadcastReceiver {
+	return &SimpleBroadcastReceiver{}
+}
+
+func (r *SimpleBroadcastReceiver) Start(h pilosa.BroadcastHandler) error {
+	r.broadcastHandler = h
+	return nil
+}
+
+func (r *SimpleBroadcastReceiver) Receive(pb proto.Message) error {
+	r.broadcastHandler.ReceiveMessage(pb)
+	return nil
+}
+
+type SimpleBroadcastHandler struct {
+	receivedMessage proto.Message
+}
+
+func NewSimpleBroadcastHandler() *SimpleBroadcastHandler {
+	return &SimpleBroadcastHandler{}
+}
+
+func (h *SimpleBroadcastHandler) ReceiveMessage(pb proto.Message) error {
+	h.receivedMessage = pb.(proto.Message)
+	return nil
+}
--- a/cache.go
+++ b/cache.go
@ -12,6 +12,11 @@ import (
 	"github.com/pilosa/pilosa/internal"
 )

+const (
+	// ThresholdFactor is used to calculate the threshold for new items entering the cache
+	ThresholdFactor = 1.1
+)
+
 // Cache represents a cache for bitmap counts.
 type Cache interface {
 	Add(bitmapID uint64, n uint64)
@ -39,9 +44,9 @@ type LRUCache struct {
 }

 // NewLRUCache returns a new instance of LRUCache.
-func NewLRUCache(maxEntries int) *LRUCache {
+func NewLRUCache(maxEntries uint32) *LRUCache {
 	c := &LRUCache{
-		cache:  lru.New(maxEntries),
+		cache:  lru.New(int(maxEntries)),
 		counts: make(map[uint64]uint64),
 	}
 	c.cache.OnEvicted = c.onEvicted
@ -111,15 +116,23 @@ type RankCache struct {
 	updateN    int
 	updateTime time.Time

-	ThresholdLength int
-	ThresholdIndex  int
-	ThresholdValue  uint64
+	// maxEntries is the user defined size of the cache
+	maxEntries uint32
+
+	// thresholdBuffer is used the calculate the lowest cached threshold value
+	// This threshold determines what new items are added to the cache
+	thresholdBuffer int
+
+	// thresholdValue is the value of the last item in the cache
+	thresholdValue uint64
 }

 // NewRankCache returns a new instance of RankCache.
-func NewRankCache() *RankCache {
+func NewRankCache(maxEntries uint32) *RankCache {
 	return &RankCache{
-		entries: make(map[uint64]uint64),
+		maxEntries:      maxEntries,
+		thresholdBuffer: int(ThresholdFactor * float64(maxEntries)),
+		entries:         make(map[uint64]uint64),
 	}
 }

@ -128,7 +141,7 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	// Ignore if the bit count on the bitmap is below the threshold.
-	if n < c.ThresholdValue {
+	if n < c.thresholdValue {
 		return
 	}

@ -141,7 +154,7 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) {
 func (c *RankCache) BulkAdd(bitmapID uint64, n uint64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	if n < c.ThresholdValue {
+	if n < c.thresholdValue {
 		return
 	}

@ -209,19 +222,20 @@ func (c *RankCache) recalculate() {

 	// Store the count of the item at the threshold index.
 	c.rankings = rankings
-	if len(c.rankings) > c.ThresholdIndex {
-		c.ThresholdValue = rankings[c.ThresholdIndex].Count
+	if len(c.rankings) > int(c.maxEntries) {
+		c.thresholdValue = rankings[c.maxEntries].Count
+		c.rankings = c.rankings[0:c.maxEntries]
 	} else {
-		c.ThresholdValue = 1
+		c.thresholdValue = 1
 	}

 	// Reset counters.
 	c.updateTime, c.updateN = time.Now(), 0

 	// If size is larger than the threshold then trim it.
-	if len(c.entries) > c.ThresholdLength {
+	if len(c.entries) > c.thresholdBuffer {
 		for id, cnt := range c.entries {
-			if cnt <= c.ThresholdValue {
+			if cnt <= c.thresholdValue {
 				delete(c.entries, id)
 			}
 		}
--- a/cluster.go
+++ b/cluster.go
@ -11,11 +11,16 @@ const (

 	// DefaultReplicaN is the default number of replicas per partition.
 	DefaultReplicaN = 1
+
+	// HealthStatus is the return value of the /health endpoint for a node in the cluster.
+	HealthStatusUp   = "UP"
+	HealthStatusDown = "DOWN"
 )

 // Node represents a node in the cluster.
 type Node struct {
-	Host string `json:"host"`
+	Host         string `json:"host"`
+	InternalHost string `json:"internalHost"`
 }

 // Nodes represents a list of nodes.
@ -81,7 +86,8 @@ func (a Nodes) Clone() []*Node {

 // Cluster represents a collection of nodes.
 type Cluster struct {
-	Nodes []*Node
+	Nodes   []*Node
+	NodeSet NodeSet

 	// Hashing algorithm used to assign partitions to nodes.
 	Hasher Hasher
@ -102,6 +108,33 @@ func NewCluster() *Cluster {
 	}
 }

+// NodeSetHosts returns the list of host strings for NodeSet members
+func (c *Cluster) NodeSetHosts() []string {
+	if c.NodeSet == nil {
+		return []string{}
+	}
+	a := make([]string, 0, len(c.NodeSet.Nodes()))
+	for _, m := range c.NodeSet.Nodes() {
+		a = append(a, m.Host)
+	}
+	return a
+}
+
+// Health returns a map of nodes in the cluster with each node's state (UP/DOWN) as the value.
+func (c *Cluster) Health() map[string]string {
+	h := make(map[string]string)
+	for _, n := range c.Nodes {
+		h[n.Host] = HealthStatusDown
+	}
+	// we are assuming that NodeSetHosts is a subset of c.Nodes
+	for _, m := range c.NodeSetHosts() {
+		if _, ok := h[m]; ok {
+			h[m] = HealthStatusUp
+		}
+	}
+	return h
+}
+
 // NodeByHost returns a node reference by host.
 func (c *Cluster) NodeByHost(host string) *Node {
 	for _, n := range c.Nodes {
--- a/cluster_test.go
+++ b/cluster_test.go
@ -9,6 +9,7 @@ import (

 	"github.com/davecgh/go-spew/spew"
 	"github.com/pilosa/pilosa"
+	"github.com/pilosa/pilosa/httpbroadcast"
 )

 // Ensure the cluster can fairly distribute partitions across the nodes.
@ -77,6 +78,46 @@ func TestHasher(t *testing.T) {
 	}
 }

+// Ensure that an empty cluster returns a valid (empty) NodeSet
+func TestCluster_NodeSetHosts(t *testing.T) {
+
+	c := pilosa.Cluster{}
+
+	if h := c.NodeSetHosts(); !reflect.DeepEqual(h, []string{}) {
+		t.Fatalf("unexpected slice of hosts: %s", h)
+	}
+}
+
+// Ensure cluster can compare its Nodes and Members
+func TestCluster_Health(t *testing.T) {
+	c := pilosa.Cluster{
+		Nodes: []*pilosa.Node{
+			{Host: "serverA:1000"},
+			{Host: "serverB:1000"},
+			{Host: "serverC:1000"},
+		},
+		NodeSet: &httpbroadcast.HTTPNodeSet{},
+	}
+
+	err := c.NodeSet.(*httpbroadcast.HTTPNodeSet).Join([]*pilosa.Node{
+		&pilosa.Node{Host: "serverA:1000"},
+		&pilosa.Node{Host: "serverC:1000"},
+		&pilosa.Node{Host: "serverD:1000"},
+	})
+	if err != nil {
+		t.Fatalf("unexpected gossiper nodes: %s", err)
+	}
+
+	// Verify a DOWN node is reported, and extraneous nodes are ignored
+	if a := c.Health(); !reflect.DeepEqual(a, map[string]string{
+		"serverA:1000": pilosa.HealthStatusUp,
+		"serverB:1000": pilosa.HealthStatusDown,
+		"serverC:1000": pilosa.HealthStatusUp,
+	}) {
+		t.Fatalf("unexpected health: %s", spew.Sdump(a))
+	}
+}
+
 // NewCluster returns a cluster with n nodes and uses a mod-based hasher.
 func NewCluster(n int) *pilosa.Cluster {
 	c := pilosa.NewCluster()
--- a/cmd/server.go
+++ b/cmd/server.go
@ -75,14 +75,18 @@ on the configured port.`,

 	flags.StringVarP(&Server.Config.DataDir, "data-dir", "d", "~/.pilosa", "Directory to store pilosa data files.")
 	flags.StringVarP(&Server.Config.Host, "bind", "b", ":10101", "Default URI on which pilosa should listen.")
-	flags.IntVarP(&Server.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number hosts each piece of data should be stored on.")
-	flags.StringSliceVarP(&Server.Config.Cluster.Nodes, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.")
+	flags.IntVarP(&Server.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.")
+	flags.StringSliceVarP(&Server.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.")
+	flags.StringSliceVarP(&Server.Config.Cluster.InternalHosts, "cluster.internal-hosts", "", []string{}, "Comma separated list of hosts in cluster used for internal communication.")
 	flags.DurationVarP((*time.Duration)(&Server.Config.Cluster.PollingInterval), "cluster.poll-interval", "", time.Minute, "Polling interval for cluster.") // TODO what actually is this?
 	flags.StringVarP(&Server.Config.Plugins.Path, "plugins.path", "", "", "Path to plugin directory.")
 	flags.StringVar(&Server.Config.LogPath, "log-path", "", "Log path")
 	flags.DurationVarP((*time.Duration)(&Server.Config.AntiEntropy.Interval), "anti-entropy.interval", "", time.Minute*10, "Interval at which to run anti-entropy routine.")
 	flags.StringVarP(&Server.CPUProfile, "profile.cpu", "", "", "Where to store CPU profile.")
 	flags.DurationVarP(&Server.CPUTime, "profile.cpu-time", "", 30*time.Second, "CPU profile duration.")
+	flags.StringVarP(&Server.Config.Cluster.Type, "cluster.type", "", "static", "Determine how the cluster handles membership and state sharing. Choose from [static, http, gossip]")
+	flags.StringVarP(&Server.Config.Cluster.GossipSeed, "cluster.gossip-seed", "", "", "Host with which to seed the gossip membership.")
+	flags.StringVarP(&Server.Config.Cluster.InternalPort, "cluster.internal-port", "", "", "Port to which pilosa should bind for internal state sharing.")

 	return serveCmd
 }
--- a/cmd/server_test.go
+++ b/cmd/server_test.go
@ -47,7 +47,7 @@ bind = "localhost:0"
 				v.Check(cmd.Server.Config.DataDir, actualDataDir)
 				v.Check(cmd.Server.Config.Host, "localhost:0")
 				v.Check(cmd.Server.Config.Cluster.ReplicaN, 2)
-				v.Check(cmd.Server.Config.Cluster.Nodes, []string{"example.com:10101", "example.com:10110"})
+				v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:10101", "example.com:10110"})
 				v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Second*182))
 				return v.Error()
 			},
@ -68,7 +68,7 @@ data-dir = "` + actualDataDir + `"
 `,
 			validation: func() error {
 				v := validator{}
-				v.Check(cmd.Server.Config.Cluster.Nodes, []string{"example.com:1110", "example.com:1111"})
+				v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:1110", "example.com:1111"})
 				v.Check(cmd.Server.Config.Plugins.Path, "/var/sloth")
 				v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*9))
 				return v.Error()
@ -94,7 +94,7 @@ data-dir = "` + actualDataDir + `"
 `,
 			validation: func() error {
 				v := validator{}
-				v.Check(cmd.Server.Config.Cluster.Nodes, []string{"localhost:19444"})
+				v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:19444"})
 				v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Minute*2))
 				v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*11))
 				v.Check(cmd.Server.CPUProfile, profFile.Name())
@ -106,6 +106,8 @@ data-dir = "` + actualDataDir + `"
 				// confirm log file was written
 				info, err := logFile.Stat()
 				if err != nil || info.Size() == 0 {
+					// NOTE: this test assumes that something is being written to the log
+					// currently, that is relying on log: "index sync monitor initializing"
 					return errors.New("Log file was not written!")
 				}
 				return nil
--- a/config.go
+++ b/config.go
@ -4,8 +4,10 @@ import "time"

 const (
 	// DefaultHost is the default hostname and port to use.
-	DefaultHost = "localhost"
-	DefaultPort = "10101"
+	DefaultHost         = "localhost"
+	DefaultPort         = "10101"
+	DefaultClusterType  = "static"
+	DefaultInternalPort = "14000"
 )

 // Config represents the configuration for the command.
@ -15,8 +17,12 @@ type Config struct {

 	Cluster struct {
 		ReplicaN        int      `toml:"replicas"`
-		Nodes           []string `toml:"hosts"`
+		Type            string   `toml:"type"`
+		Hosts           []string `toml:"hosts"`
+		InternalHosts   []string `toml:"internal-hosts"`
 		PollingInterval Duration `toml:"polling-interval"`
+		InternalPort    string   `toml:"internal-port"`
+		GossipSeed      string   `toml:"gossip-seed"`
 	} `toml:"cluster"`

 	Plugins struct {
@ -36,34 +42,14 @@ func NewConfig() *Config {
 		Host: DefaultHost + ":" + DefaultPort,
 	}
 	c.Cluster.ReplicaN = DefaultReplicaN
+	c.Cluster.Type = DefaultClusterType
 	c.Cluster.PollingInterval = Duration(DefaultPollingInterval)
-	c.Cluster.Nodes = []string{}
+	c.Cluster.Hosts = []string{}
+	c.Cluster.InternalHosts = []string{}
 	c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval)
 	return c
 }

-// NewConfigForHosts returns a Config object with Config.Cluster.Nodes already
-// set up.
-func NewConfigForHosts(hosts []string) *Config {
-	conf := NewConfig()
-	for _, hostport := range hosts {
-		conf.Cluster.Nodes = append(conf.Cluster.Nodes, hostport)
-	}
-	return conf
-}
-
-// PilosaCluster returns a new instance of Cluster based on the config.
-func (c *Config) PilosaCluster() *Cluster {
-	cluster := NewCluster()
-	cluster.ReplicaN = c.Cluster.ReplicaN
-
-	for _, hostport := range c.Cluster.Nodes {
-		cluster.Nodes = append(cluster.Nodes, &Node{Host: hostport})
-	}
-
-	return cluster
-}
-
 // Duration is a TOML wrapper type for time.Duration.
 type Duration time.Duration

--- a/ctl/import.go
+++ b/ctl/import.go
@ -27,7 +27,7 @@ type ImportCommand struct {
 	Paths []string `json:"paths"`

 	// Size of buffer used to chunk import.
-	BufferSize int `json:"buffer-size"`
+	BufferSize int `json:"bufferSize"`

 	// Reusable client.
 	Client *pilosa.Client `json:"-"`
--- a/db.go
+++ b/db.go
@ -43,7 +43,8 @@ type DB struct {
 	// Profile attribute storage and cache
 	profileAttrStore *AttrStore

-	stats StatsClient
+	broadcaster Broadcaster
+	stats       StatsClient

 	LogOutput io.Writer
 }
@ -171,7 +172,7 @@ func (db *DB) openFrames() error {

 // loadMeta reads meta data for the database, if any.
 func (db *DB) loadMeta() error {
-	var pb internal.DB
+	var pb internal.DBMeta

 	// Read data from meta file.
 	buf, err := ioutil.ReadFile(filepath.Join(db.path, ".meta"))
@ -197,7 +198,7 @@ func (db *DB) loadMeta() error {
 // saveMeta writes meta data for the database.
 func (db *DB) saveMeta() error {
 	// Marshal metadata.
-	buf, err := proto.Marshal(&internal.DB{
+	buf, err := proto.Marshal(&internal.DBMeta{
 		TimeQuantum: string(db.timeQuantum),
 		ColumnLabel: db.columnLabel,
 	})
@ -249,10 +250,10 @@ func (db *DB) MaxSlice() uint64 {
 	return max
 }

-func (db *DB) SetRemoteMaxSlice(v uint64) {
+func (db *DB) SetRemoteMaxSlice(newmax uint64) {
 	db.mu.Lock()
 	defer db.mu.Unlock()
-	db.remoteMaxSlice = v
+	db.remoteMaxSlice = newmax
 }

 // MaxInverseSlice returns the max inverse slice in the database according to this node.
@ -391,6 +392,10 @@ func (db *DB) createFrame(name string, opt FrameOptions) (*Frame, error) {
 	if opt.RowLabel != "" {
 		f.rowLabel = opt.RowLabel
 	}
+	if opt.CacheSize != 0 {
+		f.cacheSize = opt.CacheSize
+	}
+
 	f.inverseEnabled = opt.InverseEnabled
 	if err := f.saveMeta(); err != nil {
 		f.Close()
@ -412,6 +417,7 @@ func (db *DB) newFrame(path, name string) (*Frame, error) {
 	}
 	f.LogOutput = db.LogOutput
 	f.stats = db.stats.WithTags(fmt.Sprintf("frame:%s", name))
+	f.broadcaster = db.broadcaster
 	return f, nil
 }

@ -502,9 +508,40 @@ func MergeSchemas(a, b []*DBInfo) []*DBInfo {
 	return dbs
 }

+// encodeDBs converts a into its internal representation.
+func encodeDBs(a []*DB) []*internal.DB {
+	other := make([]*internal.DB, len(a))
+	for i := range a {
+		other[i] = encodeDB(a[i])
+	}
+	return other
+}
+
+// encodeDB converts d into its internal representation.
+func encodeDB(d *DB) *internal.DB {
+	return &internal.DB{
+		Name: d.name,
+		Meta: &internal.DBMeta{
+			ColumnLabel: d.columnLabel,
+			TimeQuantum: string(d.timeQuantum),
+		},
+		MaxSlice: d.remoteMaxSlice,
+		Frames:   encodeFrames(d.Frames()),
+	}
+}
+
 // DBOptions represents options to set when initializing a db.
 type DBOptions struct {
-	ColumnLabel string `json:"columnLabel,omitempty"`
+	ColumnLabel string      `json:"columnLabel,omitempty"`
+	TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"`
+}
+
+// Encode converts o into its internal representation.
+func (o *DBOptions) Encode() *internal.DBMeta {
+	return &internal.DBMeta{
+		ColumnLabel: o.ColumnLabel,
+		TimeQuantum: string(o.TimeQuantum),
+	}
 }

 // hasTime returns true if a contains a non-nil time.
--- a/executor_test.go
+++ b/executor_test.go
@ -470,7 +470,9 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) {
 			t.Fatalf("unexpected db: %s", db)
 		} else if query.String() != `Bitmap(frame="f", id=10)` {
 			t.Fatalf("unexpected query: %s", query.String())
-		} else if !reflect.DeepEqual(slices, []uint64{0}) { //TODO: this is incorrect because the calling node doesn't know about slice 2
+			// NOTE: while the following is technically incorrect (it should be {0, 2}) because the calling node doesn't know about slice 2 yet,
+			// we are ok with this and assuming that the calling node will become aware of slice 2 via inter-node messaging
+		} else if !reflect.DeepEqual(slices, []uint64{0}) {
 			t.Fatalf("unexpected slices: %+v", slices)
 		}

--- a/fragment.go
+++ b/fragment.go
@ -70,6 +70,7 @@ type Fragment struct {
 	// Cache for bitmap counts.
 	cacheType string // passed in by frame
 	cache     Cache
+	cacheSize uint32

 	// Cache containing full bitmaps (not just counts).
 	bitmapCache BitmapCache
@ -101,6 +102,7 @@ func NewFragment(path, db, frame, view string, slice uint64) *Fragment {
 		view:      view,
 		slice:     slice,
 		cacheType: DefaultCacheType,
+		cacheSize: DefaultCacheSize,

 		LogOutput: ioutil.Discard,
 		MaxOpN:    DefaultFragmentMaxOpN,
@ -222,12 +224,9 @@ func (f *Fragment) openCache() error {
 	// Determine cache type from frame name.
 	switch f.cacheType {
 	case CacheTypeRanked:
-		c := NewRankCache()
-		c.ThresholdLength = 50000
-		c.ThresholdIndex = 45000
-		f.cache = c
+		f.cache = NewRankCache(f.cacheSize)
 	case CacheTypeLRU:
-		f.cache = NewLRUCache(50000)
+		f.cache = NewLRUCache(f.cacheSize)
 	default:
 		return ErrInvalidCacheType
 	}
--- a/fragment_test.go
+++ b/fragment_test.go
@ -277,6 +277,73 @@ func TestFragment_TopN_BitmapIDs(t *testing.T) {
 	}
 }

+// Ensure the fragment cache limit works
+func TestFragment_TopN_CacheSize(t *testing.T) {
+	slice := uint64(0)
+	cacheSize := uint32(3)
+
+	// Create DB.
+	db := MustOpenDB()
+	defer db.Close()
+
+	// Create frame.
+	frame, err := db.CreateFrameIfNotExists("f", pilosa.FrameOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: cacheSize})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create view.
+	view, err := frame.CreateViewIfNotExists(pilosa.ViewStandard)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create fragment.
+	frag, err := view.CreateFragmentIfNotExists(slice)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Close the storage so we can re-open it without encountering a flock.
+	frag.Close()
+
+	f := &Fragment{
+		Fragment:        frag,
+		BitmapAttrStore: MustOpenAttrStore(),
+	}
+	f.Fragment.BitmapAttrStore = f.BitmapAttrStore.AttrStore
+	if err := f.Open(); err != nil {
+		panic(err)
+	}
+	defer f.Close()
+
+	// Set bits on various bitmaps.
+	f.MustSetBits(100, 1, 2, 3)
+	f.MustSetBits(101, 4, 5, 6, 7)
+	f.MustSetBits(102, 8, 9, 10, 11, 12)
+	f.MustSetBits(103, 8, 9, 10, 11, 12, 13)
+	f.MustSetBits(104, 8, 9, 10, 11, 12, 13, 14)
+	f.MustSetBits(105, 10, 11)
+
+	f.RecalculateCache()
+
+	p := []pilosa.Pair{
+		{ID: 104, Count: 7},
+		{ID: 103, Count: 6},
+		{ID: 102, Count: 5},
+	}
+
+	// Retrieve top bitmaps.
+	if pairs, err := f.Top(pilosa.TopOptions{N: 5}); err != nil {
+		t.Fatal(err)
+	} else if len(pairs) > int(cacheSize) {
+		t.Fatalf("TopN count cannot exceed cache size: %d", cacheSize)
+	} else if pairs[0] != (pilosa.Pair{ID: 104, Count: 7}) {
+		t.Fatalf("unexpected pair(0): %v", pairs)
+	} else if !reflect.DeepEqual(pairs, p) {
+		t.Fatalf("Invalid TopN result set: %s", spew.Sdump(pairs))
+	}
+}
+
 // Ensure fragment can return a checksum for its blocks.
 func TestFragment_Checksum(t *testing.T) {
 	f := MustOpenFragment("d", "f", pilosa.ViewStandard, 0)
--- a/frame.go
+++ b/frame.go
@ -20,6 +20,9 @@ const (
 	DefaultRowLabel       = "id"
 	DefaultCacheType      = CacheTypeLRU
 	DefaultInverseEnabled = false
+
+	// Default ranked frame cache
+	DefaultCacheSize = 50000
 )

 // Frame represents a container for views.
@ -35,13 +38,17 @@ type Frame struct {
 	// Bitmap attribute storage and cache
 	bitmapAttrStore *AttrStore

-	stats StatsClient
+	broadcaster Broadcaster
+	stats       StatsClient

 	// Frame settings.
 	rowLabel       string
 	cacheType      string
 	inverseEnabled bool

+	// Cache size for ranked frames
+	cacheSize uint32
+
 	LogOutput io.Writer
 }

@ -63,8 +70,9 @@ func NewFrame(path, db, name string) (*Frame, error) {
 		stats: NopStatsClient,

 		rowLabel:       DefaultRowLabel,
-		cacheType:      DefaultCacheType,
 		inverseEnabled: DefaultInverseEnabled,
+		cacheType:      DefaultCacheType,
+		cacheSize:      DefaultCacheSize,

 		LogOutput: ioutil.Discard,
 	}, nil
@ -149,13 +157,43 @@ func (f *Frame) InverseEnabled() bool {
 	return f.inverseEnabled
 }

+// SetCacheSize sets the cache size for ranked fames. Persists to meta file on update.
+// defaults to DefaultCacheSize 50000
+func (f *Frame) SetCacheSize(v uint32) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	// Ignore if no change occurred.
+	if v == 0 || f.cacheSize == v {
+		return nil
+	}
+
+	// Persist meta data to disk on change.
+	f.cacheSize = v
+	if err := f.saveMeta(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// CacheSize returns the ranked frame cache size.
+func (f *Frame) CacheSize() uint32 {
+	f.mu.Lock()
+	v := f.cacheSize
+	f.mu.Unlock()
+	return v
+}
+
 // Options returns all options for this frame.
 func (f *Frame) Options() FrameOptions {
 	f.mu.Lock()
 	opt := FrameOptions{
 		RowLabel:       f.rowLabel,
-		CacheType:      f.cacheType,
 		InverseEnabled: f.inverseEnabled,
+		CacheType:      f.cacheType,
+		CacheSize:      f.cacheSize,
+		TimeQuantum:    f.timeQuantum,
 	}
 	f.mu.Unlock()
 	return opt
@ -226,7 +264,7 @@ func (f *Frame) openViews() error {

 // loadMeta reads meta data for the frame, if any.
 func (f *Frame) loadMeta() error {
-	var pb internal.Frame
+	var pb internal.FrameMeta

 	// Read data from meta file.
 	buf, err := ioutil.ReadFile(filepath.Join(f.path, ".meta"))
@ -235,6 +273,7 @@ func (f *Frame) loadMeta() error {
 		f.rowLabel = DefaultRowLabel
 		f.cacheType = DefaultCacheType
 		f.inverseEnabled = DefaultInverseEnabled
+		f.cacheSize = DefaultCacheSize
 		return nil
 	} else if err != nil {
 		return err
@ -248,6 +287,7 @@ func (f *Frame) loadMeta() error {
 	f.timeQuantum = TimeQuantum(pb.TimeQuantum)
 	f.rowLabel = pb.RowLabel
 	f.inverseEnabled = pb.InverseEnabled
+	f.cacheSize = pb.CacheSize

 	// Copy cache type.
 	f.cacheType = pb.CacheType
@ -261,11 +301,12 @@ func (f *Frame) loadMeta() error {
 // saveMeta writes meta data for the frame.
 func (f *Frame) saveMeta() error {
 	// Marshal metadata.
-	buf, err := proto.Marshal(&internal.Frame{
-		TimeQuantum:    string(f.timeQuantum),
+	buf, err := proto.Marshal(&internal.FrameMeta{
 		RowLabel:       f.rowLabel,
-		CacheType:      f.cacheType,
 		InverseEnabled: f.inverseEnabled,
+		CacheType:      f.cacheType,
+		CacheSize:      f.cacheSize,
+		TimeQuantum:    string(f.timeQuantum),
 	})
 	if err != nil {
 		return err
@ -377,7 +418,7 @@ func (f *Frame) CreateViewIfNotExists(name string) (*View, error) {
 }

 func (f *Frame) newView(path, name string) *View {
-	view := NewView(path, f.db, f.name, name)
+	view := NewView(path, f.db, f.name, name, f.cacheSize)
 	view.cacheType = f.cacheType
 	view.LogOutput = f.LogOutput
 	view.BitmapAttrStore = f.bitmapAttrStore
@ -545,6 +586,29 @@ func (f *Frame) Import(bitmapIDs, profileIDs []uint64, timestamps []*time.Time)
 	return nil
 }

+// encodeFrames converts a into its internal representation.
+func encodeFrames(a []*Frame) []*internal.Frame {
+	other := make([]*internal.Frame, len(a))
+	for i := range a {
+		other[i] = encodeFrame(a[i])
+	}
+	return other
+}
+
+// encodeFrame converts f into its internal representation.
+func encodeFrame(f *Frame) *internal.Frame {
+	return &internal.Frame{
+		Name: f.name,
+		Meta: &internal.FrameMeta{
+			RowLabel:       f.rowLabel,
+			InverseEnabled: f.inverseEnabled,
+			CacheType:      f.cacheType,
+			CacheSize:      f.cacheSize,
+			TimeQuantum:    string(f.timeQuantum),
+		},
+	}
+}
+
 type frameSlice []*Frame

 func (p frameSlice) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
@ -565,9 +629,22 @@ func (p frameInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name }

 // FrameOptions represents options to set when initializing a frame.
 type FrameOptions struct {
-	RowLabel       string `json:"rowLabel,omitempty"`
-	CacheType      string `json:"cacheType,omitempty"`
-	InverseEnabled bool   `json:"inverseEnabled,omitempty"`
+	RowLabel       string      `json:"rowLabel,omitempty"`
+	InverseEnabled bool        `json:"inverseEnabled,omitempty"`
+	CacheType      string      `json:"cacheType,omitempty"`
+	CacheSize      uint32      `json:"cacheSize,omitempty"`
+	TimeQuantum    TimeQuantum `json:"timeQuantum,omitempty"`
+}
+
+// Encode converts o into its internal representation.
+func (o *FrameOptions) Encode() *internal.FrameMeta {
+	return &internal.FrameMeta{
+		RowLabel:       o.RowLabel,
+		InverseEnabled: o.InverseEnabled,
+		CacheType:      o.CacheType,
+		CacheSize:      o.CacheSize,
+		TimeQuantum:    string(o.TimeQuantum),
+	}
 }

 // importBitSet represents slices of row and column ids.
--- a/frame_test.go
+++ b/frame_test.go
@ -126,3 +126,24 @@ func (f *Frame) MustSetBit(view string, bitmapID, profileID uint64, t *time.Time
 	}
 	return changed
 }
+
+// Ensure frame can set its cache
+func TestFrame_SetCacheSize(t *testing.T) {
+	f := MustOpenFrame()
+	defer f.Close()
+	cacheSize := uint32(100)
+
+	// Set & retrieve frame cache size.
+	if err := f.SetCacheSize(cacheSize); err != nil {
+		t.Fatal(err)
+	} else if q := f.CacheSize(); q != cacheSize {
+		t.Fatalf("unexpected frame cache size: %d", q)
+	}
+
+	// Reload frame and verify that it is persisted.
+	if err := f.Reopen(); err != nil {
+		t.Fatal(err)
+	} else if q := f.CacheSize(); q != cacheSize {
+		t.Fatalf("unexpected frame cache size (reopen): %d", q)
+	}
+}
--- a/glide.lock
+++ b/glide.lock
@ -1,6 +1,8 @@
-hash: 743e8f978eb4ad8f80a2ab71b05caebbf50b6769b71aa457bc4f144fef8c6595
-updated: 2017-04-18T15:33:39.035615802-05:00
+hash: 4bdea17c62dcd469584382515052e7a246fae6deefc2d62da70bf768e18e1f0c
+updated: 2017-04-19T10:51:10.409094081-05:00
 imports:
+- name: github.com/armon/go-metrics
+  version: 97c69685293dce4c0a2d0b19535179bbc976e4d2
 - name: github.com/boltdb/bolt
  version: 4b1ebc1869ad66568b313d0dc410e2be72670dda
 - name: github.com/BurntSushi/toml
@ -24,13 +26,21 @@ imports:
  subpackages:
  - lru
 - name: github.com/golang/protobuf
-  version: 888eb0692c857ec880338addf316bd662d5e630e
+  version: 8ee79997227bf9b34611aee7946ae64735e6fd93
  subpackages:
  - proto
 - name: github.com/gorilla/context
  version: 08b5f424b9271eedf6f9f0ce86cb9396ed337a42
 - name: github.com/gorilla/mux
  version: 392c28fe23e1c45ddba891b0320b3b5df220beea
+- name: github.com/hashicorp/errwrap
+  version: 7554cd9344cec97297fa6649b055a8c98c2a1e55
+- name: github.com/hashicorp/go-msgpack
+  version: fa3f63826f7c23912c15263591e65d54d080b458
+  subpackages:
+  - codec
+- name: github.com/hashicorp/go-multierror
+  version: ed905158d87462226a13fe39ddf685ea65f1c11f
 - name: github.com/hashicorp/hcl
  version: 630949a3c5fa3c613328e1b8256052cbc2327c9b
  subpackages:
@ -42,10 +52,14 @@ imports:
  - json/parser
  - json/scanner
  - json/token
+- name: github.com/hashicorp/memberlist
+  version: 9800c50ab79c002353852a9b1095e9591b161513
 - name: github.com/inconshreveable/mousetrap
  version: 76626ae9c91c4f2a10f34cad8ce83ea42c93bb75
 - name: github.com/magiconair/properties
  version: b3b15ef068fd0b17ddf408a23669f20811d194d2
+- name: github.com/miekg/dns
+  version: ca336a1f95a6b89be9c250df26c7a41742eb4a6f
 - name: github.com/mitchellh/mapstructure
  version: db1efb556f84b25a0a13a04aad883943538ad2e0
 - name: github.com/pelletier/go-buffruneio
@ -68,6 +82,14 @@ imports:
  version: 9ff6c6923cfffbcd502984b8e0c80539a94968b7
 - name: github.com/spf13/viper
  version: 7538d73b4eb9511d85a9f1dfef202eeb8ac260f4
+- name: golang.org/x/net
+  version: 60c41d1de8da134c05b7b40154a9a82bf5b7edb9
+  subpackages:
+  - context
+- name: golang.org/x/sync
+  version: 450f422ab23cf9881c94e2db30cac0eb1b7cf80c
+  subpackages:
+  - errgroup
 - name: golang.org/x/sys
  version: c200b10b5d5e122be351b67af224adc6128af5bf
  subpackages:
--- a/glide.yaml
+++ b/glide.yaml
@ -31,3 +31,5 @@ import:
 - package: github.com/spf13/viper
 - package: github.com/gorilla/mux
  version: ^1.3.0
+- package: github.com/hashicorp/memberlist
+- package: golang.org/x/sync
--- a/gossip/gossip.go
+++ b/gossip/gossip.go
@ -0,0 +1,225 @@
+package gossip
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"os"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/hashicorp/memberlist"
+	"github.com/pilosa/pilosa"
+	"github.com/pilosa/pilosa/internal"
+)
+
+// StateHandler specifies two methods which an object must implement to share
+// state in the cluster. These are used by the GossipNodeSet to implement the
+// LocalState and MergeRemoteState methods of memberlist.Delegate
+type StateHandler interface {
+	LocalState() (proto.Message, error)
+	HandleRemoteState(proto.Message) error
+}
+
+// GossipNodeSet represents a gossip implementation of NodeSet using memberlist
+// GossipNodeSet also represents a gossip implementation of pilosa.Broadcaster
+// GossipNodeSet also represents an implementation of memberlist.Delegate
+type GossipNodeSet struct {
+	memberlist *memberlist.Memberlist
+	handler    pilosa.BroadcastHandler
+
+	broadcasts *memberlist.TransmitLimitedQueue
+
+	stateHandler StateHandler
+	config       *GossipConfig
+
+	// The writer for any logging.
+	LogOutput io.Writer
+}
+
+func (g *GossipNodeSet) Nodes() []*pilosa.Node {
+	a := make([]*pilosa.Node, 0, g.memberlist.NumMembers())
+	for _, n := range g.memberlist.Members() {
+		a = append(a, &pilosa.Node{Host: n.Name})
+	}
+	return a
+}
+
+func (g *GossipNodeSet) Start(h pilosa.BroadcastHandler) error {
+	g.handler = h
+	return nil
+}
+
+func (g *GossipNodeSet) Open() error {
+	if g.handler == nil {
+		return fmt.Errorf("opening GossipNodeSet: you must call Start(pilosa.BroadcastHandler) before calling Open()")
+	}
+	ml, err := memberlist.Create(g.config.memberlistConfig)
+	if err != nil {
+		return err
+	}
+	g.memberlist = ml
+
+	// attach to gossip seed node
+	nodes := []*pilosa.Node{&pilosa.Node{Host: g.config.gossipSeed}} //TODO: support a list of seeds
+	_, err = g.memberlist.Join(pilosa.Nodes(nodes).Hosts())
+	if err != nil {
+		return err
+	}
+	g.broadcasts = &memberlist.TransmitLimitedQueue{
+		NumNodes: func() int {
+			return ml.NumMembers()
+		},
+		RetransmitMult: 3,
+	}
+	return nil
+}
+
+// logger returns a logger for the GossipNodeSet.
+func (g *GossipNodeSet) logger() *log.Logger {
+	return log.New(g.LogOutput, "", log.LstdFlags)
+}
+
+////////////////////////////////////////////////////////////////
+
+type GossipConfig struct {
+	gossipSeed       string
+	memberlistConfig *memberlist.Config
+}
+
+// NewGossipNodeSet returns a new instance of GossipNodeSet.
+func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed string, sh StateHandler) *GossipNodeSet {
+	g := &GossipNodeSet{
+		LogOutput: os.Stderr,
+	}
+
+	//TODO: pull memberlist config from pilosa.cfg file
+	g.config = &GossipConfig{
+		memberlistConfig: memberlist.DefaultLocalConfig(),
+		gossipSeed:       gossipSeed,
+	}
+	g.config.memberlistConfig.Name = name
+	g.config.memberlistConfig.BindAddr = gossipHost
+	g.config.memberlistConfig.BindPort = gossipPort
+	g.config.memberlistConfig.AdvertiseAddr = gossipHost
+	g.config.memberlistConfig.AdvertisePort = gossipPort
+	g.config.memberlistConfig.Delegate = g
+
+	g.stateHandler = sh
+
+	return g
+}
+
+// SendSync implementation of the Broadcaster interface
+func (g *GossipNodeSet) SendSync(pb proto.Message) error {
+	msg, err := pilosa.MarshalMessage(pb)
+	if err != nil {
+		return err
+	}
+
+	mlist := g.memberlist
+
+	// Direct sends the message directly to every node.
+	// An error from any node raises an error on the entire operation.
+	//
+	// Gossip uses the gossip protocol to eventually deliver the message
+	// to every node.
+	var eg errgroup.Group
+	for _, n := range mlist.Members() {
+		// Don't send the message to the local node.
+		if n == mlist.LocalNode() {
+			continue
+		}
+		node := n
+		eg.Go(func() error {
+			return mlist.SendToTCP(node, msg)
+		})
+	}
+	return eg.Wait()
+}
+
+// SendAsync implementation of the Broadcaster interface
+func (g *GossipNodeSet) SendAsync(pb proto.Message) error {
+	msg, err := pilosa.MarshalMessage(pb)
+	if err != nil {
+		return err
+	}
+
+	b := &broadcast{
+		msg:    msg,
+		notify: nil,
+	}
+	g.broadcasts.QueueBroadcast(b)
+	return nil
+}
+
+// implementation of the memberlist.Delegate interface
+func (g *GossipNodeSet) NodeMeta(limit int) []byte {
+	return []byte{}
+}
+
+func (g *GossipNodeSet) NotifyMsg(b []byte) {
+	m, err := pilosa.UnmarshalMessage(b)
+	if err != nil {
+		g.logger().Printf("unmarshal message error: %s", err)
+		return
+	}
+	if err := g.handler.ReceiveMessage(m); err != nil {
+		g.logger().Printf("receive message error: %s", err)
+		return
+	}
+}
+
+func (g *GossipNodeSet) GetBroadcasts(overhead, limit int) [][]byte {
+	return g.broadcasts.GetBroadcasts(overhead, limit)
+}
+
+func (g *GossipNodeSet) LocalState(join bool) []byte {
+	pb, err := g.stateHandler.LocalState()
+	if err != nil {
+		g.logger().Printf("error getting local state, err=%s", err)
+		return []byte{}
+	}
+
+	// Marshal nodestate data to bytes.
+	buf, err := proto.Marshal(pb)
+	if err != nil {
+		g.logger().Printf("error marshalling nodestate data, err=%s", err)
+		return []byte{}
+	}
+	return buf
+}
+
+func (g *GossipNodeSet) MergeRemoteState(buf []byte, join bool) {
+	// Unmarshal nodestate data.
+	var pb internal.NodeState
+	if err := proto.Unmarshal(buf, &pb); err != nil {
+		g.logger().Printf("error unmarshalling nodestate data, err=%s", err)
+		return
+	}
+	err := g.stateHandler.HandleRemoteState(&pb)
+	if err != nil {
+		g.logger().Printf("merge state error: %s", err)
+	}
+}
+
+// broadcast represents an implementation of memberlist.Broadcast
+type broadcast struct {
+	msg    []byte
+	notify chan<- struct{}
+}
+
+func (b *broadcast) Invalidates(other memberlist.Broadcast) bool {
+	return false
+}
+
+func (b *broadcast) Message() []byte {
+	return b.msg
+}
+
+func (b *broadcast) Finished() {
+	if b.notify != nil {
+		close(b.notify)
+	}
+}
--- a/handler.go
+++ b/handler.go
@ -26,7 +26,8 @@ import (

 // Handler represents an HTTP handler.
 type Handler struct {
-	Index *Index
+	Index       *Index
+	Broadcaster Broadcaster

 	// Local hostname & cluster configuration.
 	Host    string
@ -112,10 +113,23 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
 	}
 }

+// handleGetStatus handles GET /status requests.
+func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
+	if err := json.NewEncoder(w).Encode(getStatusResponse{
+		Health: h.Cluster.Health(),
+	}); err != nil {
+		h.logger().Printf("write status response error: %s", err)
+	}
+}
+
 type getSchemaResponse struct {
 	DBs []*DBInfo `json:"dbs"`
 }

+type getStatusResponse struct {
+	Health map[string]string `json:"health"`
+}
+
 // handlePostQuery handles /query requests.
 func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
 	dbName := mux.Vars(r)["db"]
@ -201,7 +215,7 @@ func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) {
 }

 type sliceMaxResponse struct {
-	MaxSlices map[string]uint64 `json:"MaxSlices"`
+	MaxSlices map[string]uint64 `json:"maxSlices"`
 }

 // handleGetDBs handles GET /db request.
@ -303,6 +317,15 @@ func (h *Handler) handleDeleteDB(w http.ResponseWriter, r *http.Request) {
 		return
 	}

+	// Send the delete database message to all nodes.
+	err := h.Broadcaster.SendSync(
+		&internal.DeleteDBMessage{
+			DB: dbName,
+		})
+	if err != nil {
+		h.logger().Printf("problem sending DeleteDB message: %s", err)
+	}
+
 	// Encode response.
 	if err := json.NewEncoder(w).Encode(deleteDBResponse{}); err != nil {
 		h.logger().Printf("response encoding error: %s", err)
@ -317,13 +340,17 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) {

 	// Decode request.
 	var req postDBRequest
-	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+	err := json.NewDecoder(r.Body).Decode(&req)
+	if err == io.EOF {
+		// If no data was provided (EOF), we still create the database
+		// with default values.
+	} else if err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}

 	// Create database.
-	_, err := h.Index.CreateDB(dbName, req.Options)
+	_, err = h.Index.CreateDB(dbName, req.Options)
 	if err == ErrDatabaseExists {
 		http.Error(w, err.Error(), http.StatusConflict)
 		return
@ -332,6 +359,16 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) {
 		return
 	}

+	// Send the create database message to all nodes.
+	err = h.Broadcaster.SendSync(
+		&internal.CreateDBMessage{
+			DB:   dbName,
+			Meta: req.Options.Encode(),
+		})
+	if err != nil {
+		h.logger().Printf("problem sending CreateDB message: %s", err)
+	}
+
 	// Encode response.
 	if err := json.NewEncoder(w).Encode(postDBResponse{}); err != nil {
 		h.logger().Printf("response encoding error: %s", err)
@ -376,7 +413,7 @@ func (h *Handler) handlePatchDBTimeQuantum(w http.ResponseWriter, r *http.Reques
 }

 type patchDBTimeQuantumRequest struct {
-	TimeQuantum string `json:"time_quantum"`
+	TimeQuantum string `json:"timeQuantum"`
 }

 type patchDBTimeQuantumResponse struct{}
@ -445,7 +482,11 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) {

 	// Decode request.
 	var req postFrameRequest
-	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+	err := json.NewDecoder(r.Body).Decode(&req)
+	if err == io.EOF {
+		// If no data was provided (EOF), we still create the frame
+		// with default values.
+	} else if err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
@ -458,7 +499,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) {
 	}

 	// Create frame.
-	_, err := db.CreateFrame(frameName, req.Options)
+	_, err = db.CreateFrame(frameName, req.Options)
 	if err == ErrFrameExists {
 		http.Error(w, err.Error(), http.StatusConflict)
 		return
@ -467,6 +508,17 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) {
 		return
 	}

+	// Send the create frame message to all nodes.
+	err = h.Broadcaster.SendSync(
+		&internal.CreateFrameMessage{
+			DB:    dbName,
+			Frame: frameName,
+			Meta:  req.Options.Encode(),
+		})
+	if err != nil {
+		h.logger().Printf("problem sending CreateFrame message: %s", err)
+	}
+
 	// Encode response.
 	if err := json.NewEncoder(w).Encode(postFrameResponse{}); err != nil {
 		h.logger().Printf("response encoding error: %s", err)
@ -538,6 +590,16 @@ func (h *Handler) handleDeleteFrame(w http.ResponseWriter, r *http.Request) {
 		return
 	}

+	// Send the delete frame message to all nodes.
+	err := h.Broadcaster.SendSync(
+		&internal.DeleteFrameMessage{
+			DB:    dbName,
+			Frame: frameName,
+		})
+	if err != nil {
+		h.logger().Printf("problem sending DeleteFrame message: %s", err)
+	}
+
 	// Encode response.
 	if err := json.NewEncoder(w).Encode(deleteFrameResponse{}); err != nil {
 		h.logger().Printf("response encoding error: %s", err)
@ -585,7 +647,7 @@ func (h *Handler) handlePatchFrameTimeQuantum(w http.ResponseWriter, r *http.Req
 }

 type patchFrameTimeQuantumRequest struct {
-	TimeQuantum string `json:"time_quantum"`
+	TimeQuantum string `json:"timeQuantum"`
 }

 type patchFrameTimeQuantumResponse struct{}
--- a/handler_test.go
+++ b/handler_test.go
@ -83,7 +83,7 @@ func TestHandler_MaxSlices(t *testing.T) {
 	h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max", nil))
 	if w.Code != http.StatusOK {
 		t.Fatalf("unexpected status code: %d", w.Code)
-	} else if body := w.Body.String(); body != `{"MaxSlices":{"d0":3,"d1":0}}`+"\n" {
+	} else if body := w.Body.String(); body != `{"maxSlices":{"d0":3,"d1":0}}`+"\n" {
 		t.Fatalf("unexpected body: %s", body)
 	}
 }
@ -123,7 +123,7 @@ func TestHandler_MaxSlices_Inverse(t *testing.T) {
 	h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max?inverse=true", nil))
 	if w.Code != http.StatusOK {
 		t.Fatalf("unexpected status code: %d", w.Code)
-	} else if body := w.Body.String(); body != `{"MaxSlices":{"d0":3,"d1":0}}`+"\n" {
+	} else if body := w.Body.String(); body != `{"maxSlices":{"d0":3,"d1":0}}`+"\n" {
 		t.Fatalf("unexpected body: %s", body)
 	}
 }
@ -556,7 +556,7 @@ func TestHandler_SetDBTimeQuantum(t *testing.T) {
 	h := NewHandler()
 	h.Index = idx.Index
 	w := httptest.NewRecorder()
-	h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/time-quantum", strings.NewReader(`{"time_quantum":"ymdh"}`)))
+	h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`)))
 	if w.Code != http.StatusOK {
 		t.Fatalf("unexpected status code: %d", w.Code)
 	} else if body := w.Body.String(); body != `{}`+"\n" {
@ -579,7 +579,7 @@ func TestHandler_SetFrameTimeQuantum(t *testing.T) {
 	h := NewHandler()
 	h.Index = idx.Index
 	w := httptest.NewRecorder()
-	h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/frame/f1/time-quantum", strings.NewReader(`{"time_quantum":"ymdh"}`)))
+	h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/frame/f1/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`)))
 	if w.Code != http.StatusOK {
 		t.Fatalf("unexpected status code: %d", w.Code)
 	} else if body := w.Body.String(); body != `{}`+"\n" {
@ -763,7 +763,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) {
 	h.ServeHTTP(w, r)
 	if w.Code != http.StatusOK {
 		t.Fatalf("unexpected status code: %d", w.Code)
-	} else if w.Body.String() != `[{"host":"host1"},{"host":"host2"}]`+"\n" {
+	} else if w.Body.String() != `[{"host":"host1","internalHost":""},{"host":"host2","internalHost":""}]`+"\n" {
 		t.Fatalf("unexpected body: %q", w.Body.String())
 	}
 }
@ -792,6 +792,10 @@ func NewHandler() *Handler {
 	}
 	h.Handler.Executor = &h.Executor
 	h.Handler.LogOutput = ioutil.Discard
+
+	// Handler test messages can no-op.
+	h.Broadcaster = pilosa.NopBroadcaster
+
 	return h
 }

@ -823,6 +827,8 @@ func NewServer() *Server {
 	// Update handler to use hostname.
 	s.Handler.Host = s.Host()

+	// Handler test messages can no-op.
+	s.Handler.Broadcaster = pilosa.NopBroadcaster
 	// Create a default cluster on the handler
 	s.Handler.Cluster = NewCluster(1)
 	s.Handler.Cluster.Nodes[0].Host = s.Host()
--- a/httpbroadcast/messenger.go
+++ b/httpbroadcast/messenger.go
@ -0,0 +1,180 @@
+package httpbroadcast
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/pilosa/pilosa"
+)
+
+// HTTPBroadcaster represents a NodeSet that broadcasts messages over HTTP.
+type HTTPBroadcaster struct {
+	server       *pilosa.Server
+	internalPort string
+}
+
+// NewHTTPBroadcaster returns a new instance of HTTPBroadcaster.
+func NewHTTPBroadcaster(s *pilosa.Server, internalPort string) *HTTPBroadcaster {
+	return &HTTPBroadcaster{server: s, internalPort: internalPort}
+}
+
+// SendSync sends a protobuf message to all nodes simultaneously.
+// It waits for all nodes to respond before the function returns (and returns any errors).
+func (h *HTTPBroadcaster) SendSync(pb proto.Message) error {
+	// Marshal the pb to []byte
+	buf, err := pilosa.MarshalMessage(pb)
+	if err != nil {
+		return err
+	}
+
+	nodes, err := h.nodes()
+	if err != nil {
+		return err
+	}
+
+	var g errgroup.Group
+	for _, n := range nodes {
+		// Don't send the message to the local node.
+		if n.Host == h.server.Host {
+			continue
+		}
+		node := n
+		g.Go(func() error {
+			return h.sendNodeMessage(node, buf)
+		})
+	}
+	return g.Wait()
+}
+
+// SendAsync exists to implement the Broadcaster interface, but just calls
+// SendSync.
+func (h *HTTPBroadcaster) SendAsync(pb proto.Message) error {
+	return h.SendSync(pb)
+}
+
+func (h *HTTPBroadcaster) nodes() ([]*pilosa.Node, error) {
+	if h.server == nil {
+		return nil, errors.New("HTTPBroadcaster has no reference to Server.")
+	}
+	nodeset, ok := h.server.Cluster.NodeSet.(*HTTPNodeSet)
+	if !ok {
+		return nil, errors.New("NodeSet cannot be caste to HTTPNodeSet.")
+	}
+	return nodeset.Nodes(), nil
+}
+
+func (h *HTTPBroadcaster) sendNodeMessage(node *pilosa.Node, msg []byte) error {
+	var client *http.Client
+	client = http.DefaultClient
+
+	// Create HTTP request.
+	req, err := http.NewRequest("POST", (&url.URL{
+		Scheme: "http",
+		Host:   node.InternalHost,
+	}).String(), bytes.NewReader(msg))
+
+	// Require protobuf encoding.
+	req.Header.Set("Content-Type", "application/x-protobuf")
+
+	// Send request to remote node.
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	// Read response into buffer.
+	body, err := ioutil.ReadAll(resp.Body)
+
+	if err != nil {
+		return err
+	}
+
+	// Check status code.
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body)
+	}
+
+	return nil
+}
+
+type HTTPBroadcastReceiver struct {
+	port      string
+	handler   pilosa.BroadcastHandler
+	logOutput io.Writer
+}
+
+func NewHTTPBroadcastReceiver(port string, logOutput io.Writer) *HTTPBroadcastReceiver {
+	return &HTTPBroadcastReceiver{
+		port:      port,
+		logOutput: logOutput,
+	}
+}
+
+func (rec *HTTPBroadcastReceiver) Start(b pilosa.BroadcastHandler) error {
+	rec.handler = b
+	go func() {
+		err := http.ListenAndServe(":"+rec.port, rec)
+		if err != nil {
+			fmt.Fprintf(rec.logOutput, "Error listening on %v for HTTPBroadcastReceiver: %v\n", ":"+rec.port, err)
+		}
+	}()
+	return nil
+}
+
+func (rec *HTTPBroadcastReceiver) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if r.Header.Get("Content-Type") != "application/x-protobuf" {
+		http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
+		return
+	}
+
+	// Read entire body.
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	// Unmarshal message to specific proto type.
+	m, err := pilosa.UnmarshalMessage(body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	if err := rec.handler.ReceiveMessage(m); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+}
+
+// HTTPNodeSet represents a NodeSet that broadcasts messages over HTTP.
+type HTTPNodeSet struct {
+	nodes []*pilosa.Node
+}
+
+// NewHTTPNodeSet returns a new instance of HTTPNodeSet.
+func NewHTTPNodeSet() *HTTPNodeSet {
+	return &HTTPNodeSet{}
+}
+
+func (h *HTTPNodeSet) Nodes() []*pilosa.Node {
+	return h.nodes
+}
+
+func (h *HTTPNodeSet) Open() error {
+	return nil
+}
+
+func (h *HTTPNodeSet) Join(nodes []*pilosa.Node) error {
+	h.nodes = nodes
+	return nil
+}
--- a/index.go
+++ b/index.go
@ -23,6 +23,7 @@ type Index struct {
 	// Databases by name.
 	dbs map[string]*DB

+	Broadcaster Broadcaster
 	// Close management
 	wg      sync.WaitGroup
 	closing chan struct{}
@ -181,6 +182,7 @@ func (i *Index) DBs() []*DB {
 }

 // CreateDB creates a database.
+// An error is returned if the database already exists.
 func (i *Index) CreateDB(name string, opt DBOptions) (*DB, error) {
 	i.mu.Lock()
 	defer i.mu.Unlock()
@ -198,7 +200,7 @@ func (i *Index) CreateDBIfNotExists(name string, opt DBOptions) (*DB, error) {
 	i.mu.Lock()
 	defer i.mu.Unlock()

-	// Find frame in cache first.
+	// Find database in cache first.
 	if db := i.dbs[name]; db != nil {
 		return db, nil
 	}
@ -228,6 +230,7 @@ func (i *Index) createDB(name string, opt DBOptions) (*DB, error) {

 	// Update options.
 	db.SetColumnLabel(opt.ColumnLabel)
+	db.SetTimeQuantum(opt.TimeQuantum)

 	i.dbs[db.Name()] = db

@ -243,6 +246,7 @@ func (i *Index) newDB(path, name string) (*DB, error) {
 	}
 	db.LogOutput = i.LogOutput
 	db.stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name()))
+	db.broadcaster = i.Broadcaster
 	return db, nil
 }

--- a/internal/private.pb.go
+++ b/internal/private.pb.go
--- a/internal/private.proto
+++ b/internal/private.proto
@ -2,16 +2,17 @@ syntax = "proto3";

 package internal;

-message DB {
-	string TimeQuantum = 1;
-	string ColumnLabel = 2;
+message DBMeta {
+	string ColumnLabel = 1;
+	string TimeQuantum = 2;
 }

-message Frame {
-	string TimeQuantum = 1;
-	string RowLabel = 2;
-	bool InverseEnabled = 3;
-	string CacheType = 4;
+message FrameMeta {
+	string RowLabel = 1;
+	bool InverseEnabled = 2;
+	string CacheType = 3;
+	uint32 CacheSize = 4;
+	string TimeQuantum = 5;
 }

 message ImportResponse {
@ -39,3 +40,45 @@ message MaxSlicesResponse {
    map<string, uint64> MaxSlices = 1;
 }

+message CreateSliceMessage {
+    string DB = 1;
+    uint64 Slice = 2;
+}
+
+message DeleteDBMessage {
+    string DB = 1;
+}
+
+message CreateDBMessage {
+    string DB = 1;
+    DBMeta Meta = 2;
+}
+
+message CreateFrameMessage {
+    string DB = 1;
+    string Frame = 2;
+    FrameMeta Meta = 3;
+}
+
+message DeleteFrameMessage {
+    string DB = 1;
+    string Frame = 2;
+}
+
+message Frame {
+    string Name = 1;
+    FrameMeta Meta = 2;
+}
+
+message DB {
+    string Name = 1;
+    DBMeta Meta = 2;
+    uint64 MaxSlice = 3;
+    repeated Frame Frames = 4;
+}
+
+message NodeState {
+    string Host = 1;
+    string State = 2;
+    repeated DB DBs = 3;
+}
--- a/server.go
+++ b/server.go
@ -1,6 +1,7 @@
 package pilosa

 import (
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@ -32,8 +33,10 @@ type Server struct {
 	closing chan struct{}

 	// Data storage and HTTP interface.
-	Index   *Index
-	Handler *Handler
+	Index             *Index
+	Handler           *Handler
+	Broadcaster       Broadcaster
+	BroadcastReceiver BroadcastReceiver

 	// Cluster configuration.
 	// Host is replaced with actual host after opening if port is ":0".
@ -52,8 +55,10 @@ func NewServer() *Server {
 	s := &Server{
 		closing: make(chan struct{}),

-		Index:   NewIndex(),
-		Handler: NewHandler(),
+		Index:             NewIndex(),
+		Handler:           NewHandler(),
+		Broadcaster:       NopBroadcaster,
+		BroadcastReceiver: NopBroadcastReceiver,

 		AntiEntropyInterval: DefaultAntiEntropyInterval,
 		PollingInterval:     DefaultPollingInterval,
@ -96,6 +101,15 @@ func (s *Server) Open() error {
 		return err
 	}

+	if err := s.BroadcastReceiver.Start(s); err != nil {
+		return err
+	}
+
+	// Open NodeSet communication
+	if err := s.Cluster.NodeSet.Open(); err != nil {
+		return err
+	}
+
 	// Create executor for executing queries.
 	e := NewExecutor()
 	e.Index = s.Index
@ -103,10 +117,14 @@ func (s *Server) Open() error {
 	e.Cluster = s.Cluster

 	// Initialize HTTP handler.
+	s.Handler.Broadcaster = s.Broadcaster
 	s.Handler.Host = s.Host
 	s.Handler.Cluster = s.Cluster
 	s.Handler.Executor = e
 	s.Handler.LogOutput = s.LogOutput
+
+	// Initialize Index.
+	s.Index.Broadcaster = s.Broadcaster
 	s.Index.LogOutput = s.LogOutput

 	// Serve HTTP.
@ -202,21 +220,15 @@ func (s *Server) monitorMaxSlices() {
 			if s.Host != node.Host {
 				maxSlices, _ := checkMaxSlices(node.Host)
 				for db, newmax := range maxSlices {
-					// if we don't know about a db locally, create it
-					// so that the /schema endpoint can report it
+					// if we don't know about a db locally, log an error because
+					// db's should be created and synced prior to slice creation
 					if localdb := s.Index.DB(db); localdb != nil {
 						if newmax > oldmaxslices[db] {
 							oldmaxslices[db] = newmax
 							localdb.SetRemoteMaxSlice(newmax)
 						}
 					} else {
-						d := s.Index.DB(db)
-						if d == nil {
-							s.logger().Printf("Local DB not found: %s", db)
-							return
-						}
-						oldmaxslices[db] = newmax
-						d.SetRemoteMaxSlice(newmax)
+						s.logger().Printf("Local DB not found: %s", db)
 					}
 				}
 			}
@ -224,6 +236,91 @@ func (s *Server) monitorMaxSlices() {
 	}
 }

+// ReceiveMessage represents an implementation of BroadcastHandler.
+func (s *Server) ReceiveMessage(pb proto.Message) error {
+	switch obj := pb.(type) {
+	case *internal.CreateSliceMessage:
+		d := s.Index.DB(obj.DB)
+		if d == nil {
+			return fmt.Errorf("Local DB not found: %s", obj.DB)
+		}
+		d.SetRemoteMaxSlice(obj.Slice)
+	case *internal.CreateDBMessage:
+		opt := DBOptions{ColumnLabel: obj.Meta.ColumnLabel}
+		_, err := s.Index.CreateDB(obj.DB, opt)
+		if err != nil {
+			return err
+		}
+	case *internal.DeleteDBMessage:
+		if err := s.Index.DeleteDB(obj.DB); err != nil {
+			return err
+		}
+	case *internal.CreateFrameMessage:
+		db := s.Index.DB(obj.DB)
+		opt := FrameOptions{RowLabel: obj.Meta.RowLabel}
+		_, err := db.CreateFrame(obj.Frame, opt)
+		if err != nil {
+			return err
+		}
+	case *internal.DeleteFrameMessage:
+		db := s.Index.DB(obj.DB)
+		if err := db.DeleteFrame(obj.Frame); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Server implements gossip.StateHandler.
+// LocalState returns the state of the local node as well as the
+// index (dbs/frames) according to the local node.
+// In a gossip implementation, memberlist.Delegate.LocalState() uses this.
+func (s *Server) LocalState() (proto.Message, error) {
+	if s.Index == nil {
+		return nil, errors.New("Server.Index is nil.")
+	}
+	return &internal.NodeState{
+		Host:  s.Host,
+		State: "OK", // TODO: make this work, pull from s.Cluster.Node
+		DBs:   encodeDBs(s.Index.DBs()),
+	}, nil
+}
+
+// HandleRemoteState receives incoming NodeState from remote nodes.
+func (s *Server) HandleRemoteState(pb proto.Message) error {
+	return s.mergeRemoteState(pb.(*internal.NodeState))
+}
+
+func (s *Server) mergeRemoteState(ns *internal.NodeState) error {
+	// TODO: update some node state value in the cluster (it should be in cluster.node i guess)
+
+	// Create databases that don't exist.
+	for _, db := range ns.DBs {
+		opt := DBOptions{
+			ColumnLabel: db.Meta.ColumnLabel,
+			TimeQuantum: TimeQuantum(db.Meta.TimeQuantum),
+		}
+		d, err := s.Index.CreateDBIfNotExists(db.Name, opt)
+		if err != nil {
+			return err
+		}
+		// Create frames that don't exist.
+		for _, f := range db.Frames {
+			opt := FrameOptions{
+				RowLabel:    f.Meta.RowLabel,
+				TimeQuantum: TimeQuantum(f.Meta.TimeQuantum),
+				CacheSize:   f.Meta.CacheSize,
+			}
+			_, err := d.CreateFrameIfNotExists(f.Name, opt)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
 func checkMaxSlices(hostport string) (map[string]uint64, error) {
 	// Create HTTP request.
 	req, err := http.NewRequest("GET", (&url.URL{
--- a/server/server.go
+++ b/server/server.go
@ -9,12 +9,16 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
+	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"

 	"github.com/pilosa/pilosa"
+	"github.com/pilosa/pilosa/gossip"
+	"github.com/pilosa/pilosa/httpbroadcast"
 )

 func init() {
@ -71,6 +75,35 @@ func (m *Command) Run(args ...string) (err error) {
 		m.Config.DataDir = filepath.Join(HomeDir, strings.TrimPrefix(m.Config.DataDir, prefix))
 	}

+	// SetupServer
+	err = m.SetupServer()
+	if err != nil {
+		return err
+	}
+
+	// Initialize server.
+	if err = m.Server.Open(); err != nil {
+		return fmt.Errorf("server.Open: %v", err)
+	}
+	fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Server.Host)
+	return nil
+}
+
+func (m *Command) SetupServer() error {
+	cluster := pilosa.NewCluster()
+	cluster.ReplicaN = m.Config.Cluster.ReplicaN
+
+	for _, hostport := range m.Config.Cluster.Hosts {
+		cluster.Nodes = append(cluster.Nodes, &pilosa.Node{Host: hostport})
+	}
+	// TODO: if InternalHosts is not provided then pilosa.Node.InternalHost is empty.
+	// This will throw an error when trying to Broadcast messages over HTTP.
+	// One option may be to fall back to using host from hostport + config.InternalPort.
+	for i, internalhostport := range m.Config.Cluster.InternalHosts {
+		cluster.Nodes[i].InternalHost = internalhostport
+	}
+	m.Server.Cluster = cluster
+
 	// Setup logging output.
 	if m.Config.LogPath == "" {
 		m.Server.LogOutput = m.Stderr
@ -87,21 +120,55 @@ func (m *Command) Run(args ...string) (err error) {
 	m.Server.Index.Path = m.Config.DataDir
 	m.Server.Index.Stats = pilosa.NewExpvarStatsClient()

-	// Build cluster from config file.
+	var err error
 	m.Server.Host, err = normalizeHost(m.Config.Host)
 	if err != nil {
 		return err
 	}
-	m.Server.Cluster = m.Config.PilosaCluster()
+
+	// Set internal port (string).
+	internalPortStr := pilosa.DefaultInternalPort
+	if m.Config.Cluster.InternalPort != "" {
+		internalPortStr = m.Config.Cluster.InternalPort
+	}
+
+	switch m.Config.Cluster.Type {
+	case "http":
+		m.Server.Broadcaster = httpbroadcast.NewHTTPBroadcaster(m.Server, internalPortStr)
+		m.Server.BroadcastReceiver = httpbroadcast.NewHTTPBroadcastReceiver(internalPortStr, m.Stderr)
+		m.Server.Cluster.NodeSet = httpbroadcast.NewHTTPNodeSet()
+		err := m.Server.Cluster.NodeSet.(*httpbroadcast.HTTPNodeSet).Join(m.Server.Cluster.Nodes)
+		if err != nil {
+			return err
+		}
+	case "gossip":
+		gossipPort, err := strconv.Atoi(internalPortStr)
+		if err != nil {
+			return err
+		}
+		gossipSeed := pilosa.DefaultHost
+		if m.Config.Cluster.GossipSeed != "" {
+			gossipSeed = m.Config.Cluster.GossipSeed
+		}
+		// get the host portion of addr to use for binding
+		gossipHost, _, err := net.SplitHostPort(m.Config.Host)
+		if err != nil {
+			gossipHost = m.Config.Host
+		}
+		gossipNodeSet := gossip.NewGossipNodeSet(m.Config.Host, gossipHost, gossipPort, gossipSeed, m.Server)
+		m.Server.Cluster.NodeSet = gossipNodeSet
+		m.Server.Broadcaster = gossipNodeSet
+		m.Server.BroadcastReceiver = gossipNodeSet
+	case "static", "":
+		m.Server.Broadcaster = pilosa.NopBroadcaster
+		m.Server.Cluster.NodeSet = pilosa.NewStaticNodeSet()
+		m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver
+	default:
+		return fmt.Errorf("'%v' is not a supported value for broadcaster type.", m.Config.Cluster.Type)
+	}

 	// Set configuration options.
 	m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
-
-	// Initialize server.
-	if err = m.Server.Open(); err != nil {
-		return fmt.Errorf("server.Open: %v", err)
-	}
-	fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Server.Host)
 	return nil
 }

--- a/view.go
+++ b/view.go
@ -30,6 +30,8 @@ type View struct {
 	frame string
 	name  string

+	cacheSize uint32
+
 	// Fragments by slice.
 	cacheType string // passed in by frame
 	fragments map[uint64]*Fragment
@ -41,12 +43,13 @@ type View struct {
 }

 // NewView returns a new instance of View.
-func NewView(path, db, frame, name string) *View {
+func NewView(path, db, frame, name string, cacheSize uint32) *View {
 	return &View{
-		path:  path,
-		db:    db,
-		frame: frame,
-		name:  name,
+		path:      path,
+		db:        db,
+		frame:     frame,
+		name:      name,
+		cacheSize: cacheSize,

 		cacheType: DefaultCacheType,
 		fragments: make(map[uint64]*Fragment),
@ -215,6 +218,7 @@ func (v *View) createFragmentIfNotExists(slice uint64) (*Fragment, error) {
 func (v *View) newFragment(path string, slice uint64) *Fragment {
 	frag := NewFragment(path, v.db, v.frame, v.name, slice)
 	frag.cacheType = v.cacheType
+	frag.cacheSize = v.cacheSize
 	frag.LogOutput = v.LogOutput
 	frag.stats = v.stats.WithTags(fmt.Sprintf("slice:%d", slice))
 	return frag
--- a/view_test.go
+++ b/view_test.go
@ -22,7 +22,7 @@ func NewView(db, frame, name string) *View {
 	file.Close()

 	v := &View{
-		View:            pilosa.NewView(file.Name(), db, frame, name),
+		View:            pilosa.NewView(file.Name(), db, frame, name, pilosa.DefaultCacheSize),
 		BitmapAttrStore: MustOpenAttrStore(),
 	}
 	v.View.BitmapAttrStore = v.BitmapAttrStore.AttrStore
@ -52,7 +52,7 @@ func (v *View) Reopen() error {
 		return err
 	}

-	v.View = pilosa.NewView(path, v.DB(), v.Frame(), v.Name())
+	v.View = pilosa.NewView(path, v.DB(), v.Frame(), v.Name(), pilosa.DefaultCacheSize)
 	v.View.BitmapAttrStore = v.BitmapAttrStore.AttrStore
 	if err := v.Open(); err != nil {
 		return err