Store the node state information in Cluster

Access the overall cluster state info via the /status endpoint

cluster.go

@@ -3,6 +3,9 @@ package pilosa
 import (
 	"encoding/binary"
 	"hash/fnv"
+	"sync"
+
+	"github.com/pilosa/pilosa/internal"
 )
 
 const (
@@ -86,6 +89,7 @@ func (a Nodes) Clone() []*Node {
 
 // Cluster represents a collection of nodes.
 type Cluster struct {
+	mu      sync.Mutex
 	Nodes   []*Node
 	NodeSet NodeSet
 
@@ -97,6 +101,9 @@ type Cluster struct {
 
 	// The number of replicas a partition has.
 	ReplicaN int
+
+	// Current state of nodes in the cluster
+	NodeState map[string]*internal.NodeState
 }
 
 // NewCluster returns a new instance of Cluster with defaults.
@@ -105,6 +112,7 @@ func NewCluster() *Cluster {
 		Hasher:     &jmphasher{},
 		PartitionN: DefaultPartitionN,
 		ReplicaN:   DefaultReplicaN,
+		NodeState:  make(map[string]*internal.NodeState),
 	}
 }
 
@@ -190,6 +198,37 @@ func (c *Cluster) PartitionNodes(partitionID int) []*Node {
 	return nodes
 }
 
+// OwnsSlices find the set of slices owned by the node per DB
+func (c *Cluster) OwnsSlices(db string, maxSlice uint64, host string) []uint64 {
+	var slices []uint64
+	for i := uint64(0); i <= maxSlice; i++ {
+		p := c.Partition(db, i)
+		// Determine primary owner node.
+		index := c.Hasher.Hash(uint64(p), len(c.Nodes))
+		if c.Nodes[index].Host == host {
+			slices = append(slices, i)
+		}
+
+	}
+	return slices
+}
+
+// SetNodeState stores the remote node states transmitted through gossip
+func (c *Cluster) SetNodeState(state *internal.NodeState) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.NodeState[state.Host] = state
+}
+
+// GetNodeState stores the remote node states transmitted through gossip
+func (c *Cluster) GetNodeState(host string) *internal.NodeState {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.NodeState[host]
+}
+
 // Hasher represents an interface to hash integers into buckets.
 type Hasher interface {
 	// Hashes the key into a number between [0,N).

handler.go

@@ -23,10 +23,18 @@ import (
 	"github.com/pilosa/pilosa/pql"
 )
 
+// ServerHandler a method to update the local node's state information
+// this is used to handle the cluster status request and append the
+// local node's state with the cluster state gathered via Gossip
+type ServerHandler interface {
+	HandleStateRequest() error
+}
+
 // Handler represents an HTTP handler.
 type Handler struct {
-	Index       *Index
-	Broadcaster Broadcaster
+	Index         *Index
+	Broadcaster   Broadcaster
+	ServerHandler ServerHandler
 
 	// Local hostname & cluster configuration.
 	Host    string
@@ -83,6 +91,7 @@ func NewRouter(handler *Handler) *mux.Router {
 	router.HandleFunc("/nodes", handler.handleGetNodes).Methods("GET")
 	router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET")
 	router.HandleFunc("/slices/max", handler.handleGetSliceMax).Methods("GET")
+	router.HandleFunc("/status", handler.handleGetStatus).Methods("GET")
 	router.HandleFunc("/version", handler.handleGetVersion).Methods("GET")
 
 	// TODO: Apply MethodNotAllowed statuses to all endpoints.
@@ -112,12 +121,17 @@ 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) {
+	// Compute my local state
+	fmt.Println("Call interface")
+	h.ServerHandler.HandleStateRequest()
+
 	if err := json.NewEncoder(w).Encode(getStatusResponse{
-		Health: h.Cluster.Health(),
+		Health:   h.Cluster.NodeState,
+		Version:  h.Version,
+		Replicas: h.Cluster.ReplicaN,
 	}); err != nil {
-		h.logger().Printf("write status response error: %s", err)
+		h.logger().Printf("Node State Error: %s", err)
 	}
 }
 
@@ -126,7 +140,9 @@ type getSchemaResponse struct {
 }
 
 type getStatusResponse struct {
-	Health map[string]string `json:"health"`
+	Health   map[string]*internal.NodeState `json:"health"`
+	Version  string                         `json:"version"`
+	Replicas int
 }
 
 // handlePostQuery handles /query requests.

server.go

@@ -71,6 +71,7 @@ func NewServer() *Server {
 	}
 
 	s.Handler.Index = s.Index
+	s.Handler.ServerHandler = s
 
 	return s
 }
@@ -241,6 +242,11 @@ func (s *Server) monitorMaxSlices() {
 	}
 }
 
+func (s *Server) HandleStateRequest() error {
+	_, err := s.LocalState()
+	return err
+}
+
 // 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.
@@ -248,11 +254,22 @@ func (s *Server) LocalState() (proto.Message, error) {
 	if s.Index == nil {
 		return nil, errors.New("Server.Index is nil.")
 	}
-	return &internal.NodeState{
+
+	// Get Node DB Slices
+	for _, db := range s.Index.DBs() {
+		maxSlice := db.MaxSlice()
+		slices := s.Cluster.OwnsSlices(db.name, maxSlice, s.Host)
+		fmt.Println("Slices ", slices)
+	}
+
+	ns := internal.NodeState{
 		Host:  s.Host,
 		State: "OK", // TODO: make this work, pull from s.Cluster.Node
 		DBs:   encodeDBs(s.Index.DBs()),
-	}, nil
+	}
+
+	s.Cluster.SetNodeState(&ns)
+	return &ns, nil
 }
 
 func (s *Server) ReceiveMessage(pb proto.Message) error {
@@ -296,7 +313,8 @@ func (s *Server) HandleRemoteState(pb proto.Message) error {
 }
 
 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)
+	// store this node's state in the cluster node map
+	s.Cluster.SetNodeState(ns)
 
 	// Create databases that don't exist.
 	for _, db := range ns.DBs {
@@ -392,13 +410,8 @@ func (s *Server) monitorRuntime() {
 			case <-ticker.C:
 			}
 
-			s.logger().Printf("runtime stats  beginning")
-
-			// TODO
+			// Record the number of go routines
 			s.Index.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()))
-
-			// Record successful sync in log.
-			s.logger().Printf("runtime stats  complete")
 		}
 	}
 }