Merge pull request #1001 from travisturner/973-sendsync

973 sendsync

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

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
 }

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 {

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{}

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.

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

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