diff --git a/broadcast.go b/broadcast.go index ec0e53c8d..0a9b671ec 100644 --- a/broadcast.go +++ b/broadcast.go @@ -66,6 +66,7 @@ type Broadcaster interface { func init() { NopBroadcaster = &nopBroadcaster{} + NopGossiper = &nopGossiper{} } // NopBroadcaster represents a Broadcaster that doesn't do anything. @@ -73,13 +74,13 @@ var NopBroadcaster Broadcaster type nopBroadcaster struct{} -// SendSync is a no-op implemenetation of Broadcaster SendSync method. -func (c *nopBroadcaster) SendSync(pb proto.Message) error { +// SendSync A no-op implemenetation of Broadcaster SendSync method. +func (n *nopBroadcaster) SendSync(pb proto.Message) error { return nil } -// SendAsync is a no-op implemenetation of Broadcaster SendAsync method. -func (c *nopBroadcaster) SendAsync(pb proto.Message) error { +// SendAsync A no-op implemenetation of Broadcaster SendAsync method. +func (n *nopBroadcaster) SendAsync(pb proto.Message) error { return nil } @@ -112,6 +113,21 @@ func (n *nopBroadcastReceiver) Start(b BroadcastHandler) error { return nil } // NopBroadcastReceiver is a no-op implementation of the BroadcastReceiver. var NopBroadcastReceiver = &nopBroadcastReceiver{} +// Gossiper is an interface for sharing messages via gossip. +type Gossiper interface { + SendAsync(pb proto.Message) error +} + +// NopBroadcaster represents a Broadcaster that doesn't do anything. +var NopGossiper Gossiper + +type nopGossiper struct{} + +// SendAsync A no-op implemenetation of Gossiper SendAsync method. +func (n *nopGossiper) SendAsync(pb proto.Message) error { + return nil +} + // Broadcast message types. const ( MessageTypeCreateSlice = 1 diff --git a/client.go b/client.go index 1515892c7..cc9d71ce5 100644 --- a/client.go +++ b/client.go @@ -1051,6 +1051,41 @@ func (c *InternalHTTPClient) RowAttrDiff(ctx context.Context, index, frame strin return rsp.Attrs, nil } +// SendMessage posts a message synchronously. +func (c *InternalHTTPClient) SendMessage(ctx context.Context, pb proto.Message) error { + msg, err := MarshalMessage(pb) + if err != nil { + return fmt.Errorf("marshaling message: %v", err) + } + + u := uriPathToURL(ctx.Value("uri").(*URI), "/cluster/message") + req, err := http.NewRequest("POST", u.String(), bytes.NewReader(msg)) + req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("User-Agent", "pilosa/"+Version) + + // Execute request. + resp, err := c.HTTPClient.Do(req.WithContext(ctx)) + if err != nil { + return fmt.Errorf("executing http request: %v", err) + } + defer resp.Body.Close() + + // Read body. + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("reading response body: %v", err) + } + + // Return error if status is not OK. + switch resp.StatusCode { + case http.StatusOK: // ok + default: + return fmt.Errorf("unexpected response status code: %d: %s", resp.StatusCode, body) + } + + return nil +} + func (c *InternalHTTPClient) clientURI(ctx context.Context) *URI { clientURI := c.defaultURI if contextURI, ok := ctx.Value("uri").(*URI); ok { @@ -1235,4 +1270,5 @@ type InternalClient interface { BlockData(ctx context.Context, index, frame, view string, slice uint64, block int) ([]uint64, []uint64, error) ColumnAttrDiff(ctx context.Context, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) RowAttrDiff(ctx context.Context, index, frame string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) + SendMessage(ctx context.Context, pb proto.Message) error } diff --git a/gossip/gossip.go b/gossip/gossip.go index 5300d2e1e..28b530e93 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -30,9 +30,12 @@ import ( "github.com/pilosa/pilosa/internal" ) -// GossipMemberSet represents a gossip implementation of MemberSet using memberlist -// GossipMemberSet also represents a gossip implementation of pilosa.Broadcaster -// GossipMemberSet also represents an implementation of memberlist.Delegate +// Ensure GossipMemberSet implements interfaces. +var _ pilosa.BroadcastReceiver = &GossipMemberSet{} +var _ pilosa.Gossiper = &GossipMemberSet{} +var _ memberlist.Delegate = &GossipMemberSet{} + +// GossipMemberSet represents a gossip implementation of MemberSet using memberlist. type GossipMemberSet struct { memberlist *memberlist.Memberlist handler pilosa.BroadcastHandler @@ -275,7 +278,7 @@ func (g *GossipMemberSet) SendSync(pb proto.Message) error { return eg.Wait() } -// SendAsync implementation of the Broadcaster interface. +// SendAsync implementation of the Gossiper interface. func (g *GossipMemberSet) SendAsync(pb proto.Message) error { msg, err := pilosa.MarshalMessage(pb) if err != nil { @@ -290,25 +293,6 @@ func (g *GossipMemberSet) SendAsync(pb proto.Message) error { return nil } -// SendTo implementation of the Broadcaster interface. -func (g *GossipMemberSet) SendTo(to *pilosa.Node, pb proto.Message) error { - msg, err := pilosa.MarshalMessage(pb) - if err != nil { - return err - } - - mlist := g.memberlist - - // Get the memberlist.Node from the pilosa.Node. - for _, node := range mlist.Members() { - if node.Name == to.URI.String() { - return mlist.SendToTCP(node, msg) - } - } - - return nil -} - // NodeMeta implementation of the memberlist.Delegate interface. func (g *GossipMemberSet) NodeMeta(limit int) []byte { return []byte{} diff --git a/handler.go b/handler.go index f9b21b4e1..36e6adb34 100644 --- a/handler.go +++ b/handler.go @@ -51,9 +51,10 @@ import ( // Handler represents an HTTP handler. type Handler struct { - Holder *Holder - Broadcaster Broadcaster - StatusHandler StatusHandler + Holder *Holder + Broadcaster Broadcaster + BroadcastHandler BroadcastHandler + StatusHandler StatusHandler // Local hostname & cluster configuration. URI URI @@ -174,6 +175,7 @@ func loadNormal(router *mux.Router, handler *Handler) { router.HandleFunc("/index/{index}/query", handler.handlePostQuery).Methods("POST") router.HandleFunc("/index/{index}/time-quantum", handler.handlePatchIndexTimeQuantum).Methods("PATCH") router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST") + router.HandleFunc("/cluster/message", handler.handlePostClusterMessage).Methods("POST") // TODO: Apply MethodNotAllowed statuses to all endpoints. // Ideally this would be automatic, as described in this (wontfix) ticket: @@ -516,6 +518,8 @@ func (h *Handler) handlePostIndex(w http.ResponseWriter, r *http.Request) { }) if err != nil { h.logger().Printf("problem sending CreateIndex message: %s", err) + http.Error(w, err.Error(), http.StatusInternalServerError) + return } // Encode response. @@ -2158,3 +2162,39 @@ func GetTimeStamp(data map[string]interface{}, timeField string) (int64, error) return v.Unix(), nil } + +func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Request) { + // Verify that request is only communicating over protobufs. + if r.Header.Get("Content-Type") != "application/x-protobuf" { + fmt.Println("**unsupported media type**") + 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 + } + + // Marshal into request object. + pb, err := UnmarshalMessage(body) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + // Forward the error message. + err = h.BroadcastHandler.ReceiveMessage(pb) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if err := json.NewEncoder(w).Encode(defaultClusterMessageResponse{}); err != nil { + h.logger().Printf("response encoding error: %s", err) + } +} + +type defaultClusterMessageResponse struct{} diff --git a/server.go b/server.go index 038ceab4b..94c91f2ae 100644 --- a/server.go +++ b/server.go @@ -34,6 +34,9 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/diagnostics" "github.com/pilosa/pilosa/internal" + + "golang.org/x/net/context" + "golang.org/x/sync/errgroup" ) // Default server settings. @@ -43,6 +46,11 @@ const ( DefaultDiagnosticServer = "https://diagnostics.pilosa.com/v0/diagnostics" ) +// Ensure Server implements interfaces. +var _ Broadcaster = &Server{} +var _ BroadcastHandler = &Server{} +var _ StatusHandler = &Server{} + // Server represents a holder wrapped by a running HTTP server. type Server struct { ln net.Listener @@ -56,6 +64,7 @@ type Server struct { Handler *Handler Broadcaster Broadcaster BroadcastReceiver BroadcastReceiver + Gossiper Gossiper RemoteClient *http.Client // Cluster configuration. @@ -165,6 +174,7 @@ func (s *Server) Open() error { // Initialize HTTP handler. s.Handler.Broadcaster = s.Broadcaster + s.Handler.BroadcastHandler = s s.Handler.StatusHandler = s s.Handler.URI = s.URI s.Handler.Cluster = s.Cluster @@ -395,9 +405,33 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { return nil } -// State returns the cluster state according to this node. -func (s *Server) State() string { - return s.Cluster.State +// SendSync represents an implementation of Broadcaster. +func (s *Server) SendSync(pb proto.Message) error { + var eg errgroup.Group + for _, node := range s.Cluster.Nodes { + // Don't forward the message to ourselves. + if s.URI == node.URI { + continue + } + + ctx := context.WithValue(context.Background(), "uri", &node.URI) + eg.Go(func() error { + return s.defaultClient.SendMessage(ctx, pb) + }) + } + + return eg.Wait() +} + +// SendAsync represents an implementation of Broadcaster. +func (s *Server) SendAsync(pb proto.Message) error { + return s.Gossiper.SendAsync(pb) +} + +// SendTo represents an implementation of Broadcaster. +func (s *Server) SendTo(to *Node, pb proto.Message) error { + ctx := context.WithValue(context.Background(), "uri", to.URI) + return s.defaultClient.SendMessage(ctx, pb) } // Server implements StatusHandler. diff --git a/server/server.go b/server/server.go index bd51631f6..dbc85e87a 100644 --- a/server/server.go +++ b/server/server.go @@ -229,8 +229,9 @@ func (m *Command) SetupServer() error { return err } m.Server.Cluster.MemberSet = gossipMemberSet - m.Server.Broadcaster = gossipMemberSet + m.Server.Broadcaster = m.Server m.Server.BroadcastReceiver = gossipMemberSet + m.Server.Gossiper = gossipMemberSet case pilosa.ClusterStatic, pilosa.ClusterNone: m.Server.Cluster.Static = true @@ -247,6 +248,7 @@ func (m *Command) SetupServer() error { m.Server.Broadcaster = pilosa.NopBroadcaster m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet() m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver + m.Server.Gossiper = pilosa.NopGossiper err := m.Server.Cluster.MemberSet.(*pilosa.StaticMemberSet).Join(m.Server.Cluster.Nodes) if err != nil { return err diff --git a/server/server_test.go b/server/server_test.go index d4d346716..9992dab6f 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -418,7 +418,8 @@ func TestMain_SendReceiveMessage(t *testing.T) { t.Fatal(err) } m0.Server.Cluster.MemberSet = gossipMemberSet0 - m0.Server.Broadcaster = gossipMemberSet0 + m0.Server.Broadcaster = m0.Server + m0.Server.Gossiper = gossipMemberSet0 m0.Server.Handler.Broadcaster = m0.Server.Broadcaster m0.Server.Holder.Broadcaster = m0.Server.Broadcaster m0.Server.BroadcastReceiver = gossipMemberSet0 @@ -446,7 +447,8 @@ func TestMain_SendReceiveMessage(t *testing.T) { t.Fatal(err) } m1.Server.Cluster.MemberSet = gossipMemberSet1 - m1.Server.Broadcaster = gossipMemberSet1 + m1.Server.Broadcaster = m1.Server + m1.Server.Gossiper = gossipMemberSet1 m1.Server.Handler.Broadcaster = m1.Server.Broadcaster m1.Server.Holder.Broadcaster = m1.Server.Broadcaster m1.Server.BroadcastReceiver = gossipMemberSet1