diff --git a/cluster.go b/cluster.go index f92b8c6e9..0e874e06e 100644 --- a/cluster.go +++ b/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). diff --git a/handler.go b/handler.go index a6309e44a..a72aa701d 100644 --- a/handler.go +++ b/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. diff --git a/server.go b/server.go index 68337049d..f7e0b232c 100644 --- a/server.go +++ b/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") } } }