Merge branch 'master' into cluster-resize

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

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
 }

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

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

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.

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

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