diff --git a/broadcast.go b/broadcast.go index 0785a21cc..38eb53b96 100644 --- a/broadcast.go +++ b/broadcast.go @@ -65,6 +65,7 @@ type Broadcaster interface { func init() { NopBroadcaster = &nopBroadcaster{} + NopGossiper = &nopGossiper{} } // NopBroadcaster represents a Broadcaster that doesn't do anything. @@ -73,12 +74,12 @@ var NopBroadcaster Broadcaster type nopBroadcaster struct{} // SendSync A no-op implemenetation of Broadcaster SendSync method. -func (c *nopBroadcaster) SendSync(pb proto.Message) error { +func (n *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 { +func (n *nopBroadcaster) SendAsync(pb proto.Message) error { return nil } @@ -106,6 +107,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 e44d9c5de..f71cf57d4 100644 --- a/client.go +++ b/client.go @@ -1044,6 +1044,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 { @@ -1226,4 +1261,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 e413410d8..d296f7b83 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -22,14 +22,17 @@ import ( "strings" "time" - "golang.org/x/sync/errgroup" - "github.com/gogo/protobuf/proto" "github.com/hashicorp/memberlist" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/internal" ) +// Ensure GossipNodeSet implements interfaces. +var _ pilosa.BroadcastReceiver = &GossipNodeSet{} +var _ pilosa.Gossiper = &GossipNodeSet{} +var _ memberlist.Delegate = &GossipNodeSet{} + // 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 @@ -234,35 +237,7 @@ func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed return g, nil } -// 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. +// SendAsync implementation of the Gossiper interface. func (g *GossipNodeSet) SendAsync(pb proto.Message) error { msg, err := pilosa.MarshalMessage(pb) if err != nil { diff --git a/handler.go b/handler.go index ad9e02e2e..7a094bdc9 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 @@ -136,6 +137,7 @@ func NewRouter(handler *Handler) *mux.Router { router.HandleFunc("/status", handler.handleGetStatus).Methods("GET") router.HandleFunc("/version", handler.handleGetVersion).Methods("GET") 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: @@ -483,6 +485,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. @@ -1987,3 +1991,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 208d0cd6f..68e9c6a74 100644 --- a/server.go +++ b/server.go @@ -36,6 +36,7 @@ import ( "github.com/pilosa/pilosa/diagnostics" "github.com/pilosa/pilosa/internal" "golang.org/x/net/context" + "golang.org/x/sync/errgroup" ) // Default server settings. @@ -45,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 @@ -58,6 +64,7 @@ type Server struct { Handler *Handler Broadcaster Broadcaster BroadcastReceiver BroadcastReceiver + Gossiper Gossiper RemoteClient *http.Client // Cluster configuration. @@ -181,6 +188,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 @@ -404,6 +412,34 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { return nil } +// SendSync represents an implementation of Broadcaster. +func (s *Server) SendSync(pb proto.Message) error { + var eg errgroup.Group + for _, node := range s.Cluster.Nodes { + uri, err := node.URI() + if err != nil { + return err + } + + // Don't forward the message to ourselves. + if *s.URI == *uri { + continue + } + + ctx := context.WithValue(context.Background(), "uri", 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) +} + // LocalStatus returns the state of the local node as well as the // holder (indexes/frames) according to the local node. // In a gossip implementation, memberlist.Delegate.LocalState() uses this. diff --git a/server/server.go b/server/server.go index 93b0f184f..22bfc592a 100644 --- a/server/server.go +++ b/server/server.go @@ -226,12 +226,14 @@ func (m *Command) SetupServer() error { return err } m.Server.Cluster.NodeSet = gossipNodeSet - m.Server.Broadcaster = gossipNodeSet + m.Server.Broadcaster = m.Server m.Server.BroadcastReceiver = gossipNodeSet + m.Server.Gossiper = gossipNodeSet case pilosa.ClusterStatic, pilosa.ClusterNone: m.Server.Broadcaster = pilosa.NopBroadcaster m.Server.Cluster.NodeSet = pilosa.NewStaticNodeSet() m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver + m.Server.Gossiper = pilosa.NopGossiper err := m.Server.Cluster.NodeSet.(*pilosa.StaticNodeSet).Join(m.Server.Cluster.Nodes) if err != nil { return err diff --git a/server/server_test.go b/server/server_test.go index ca8ac7f09..f9f63581a 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -412,7 +412,8 @@ func TestMain_SendReceiveMessage(t *testing.T) { t.Fatal(err) } m0.Server.Cluster.NodeSet = gossipNodeSet0 - m0.Server.Broadcaster = gossipNodeSet0 + m0.Server.Broadcaster = m0.Server + m0.Server.Gossiper = gossipNodeSet0 m0.Server.Handler.Broadcaster = m0.Server.Broadcaster m0.Server.Holder.Broadcaster = m0.Server.Broadcaster m0.Server.BroadcastReceiver = gossipNodeSet0 @@ -437,7 +438,8 @@ func TestMain_SendReceiveMessage(t *testing.T) { t.Fatal(err) } m1.Server.Cluster.NodeSet = gossipNodeSet1 - m1.Server.Broadcaster = gossipNodeSet1 + m1.Server.Broadcaster = m1.Server + m1.Server.Gossiper = gossipNodeSet1 m1.Server.Handler.Broadcaster = m1.Server.Broadcaster m1.Server.Holder.Broadcaster = m1.Server.Broadcaster m1.Server.BroadcastReceiver = gossipNodeSet1