Revert "Change Serf communication from local to http calls (#1330)"

This reverts commit fcff141.

api/http.go

@@ -8,6 +8,7 @@ import (
 	"github.com/golang/glog"
 	"github.com/julienschmidt/httprouter"
 	"github.com/livepeer/catalyst-api/balancer"
+	"github.com/livepeer/catalyst-api/cluster"
 	"github.com/livepeer/catalyst-api/config"
 	"github.com/livepeer/catalyst-api/handlers"
 	"github.com/livepeer/catalyst-api/handlers/analytics"
@@ -20,8 +21,8 @@ import (
 	"github.com/livepeer/go-api-client"
 )
 
-func ListenAndServe(ctx context.Context, cli config.Cli, vodEngine *pipeline.Coordinator, bal balancer.Balancer, mapic mistapiconnector.IMac) error {
-	router := NewCatalystAPIRouter(cli, vodEngine, bal, mapic)
+func ListenAndServe(ctx context.Context, cli config.Cli, vodEngine *pipeline.Coordinator, bal balancer.Balancer, c cluster.Cluster, mapic mistapiconnector.IMac) error {
+	router := NewCatalystAPIRouter(cli, vodEngine, bal, c, mapic)
 	server := http.Server{Addr: cli.HTTPAddress, Handler: router}
 	ctx, cancel := context.WithCancel(ctx)
 
@@ -47,7 +48,7 @@ func ListenAndServe(ctx context.Context, cli config.Cli, vodEngine *pipeline.Coo
 	return server.Shutdown(ctx)
 }
 
-func NewCatalystAPIRouter(cli config.Cli, vodEngine *pipeline.Coordinator, bal balancer.Balancer, mapic mistapiconnector.IMac) *httprouter.Router {
+func NewCatalystAPIRouter(cli config.Cli, vodEngine *pipeline.Coordinator, bal balancer.Balancer, c cluster.Cluster, mapic mistapiconnector.IMac) *httprouter.Router {
 	router := httprouter.New()
 	withLogging := middleware.LogRequest()
 	withCORS := middleware.AllowCORS()
@@ -58,7 +59,7 @@ func NewCatalystAPIRouter(cli config.Cli, vodEngine *pipeline.Coordinator, bal b
 		AccessToken: cli.APIToken,
 	})
 	catalystApiHandlers := &handlers.CatalystAPIHandlersCollection{VODEngine: vodEngine}
-	geoHandlers := geolocation.NewGeolocationHandlersCollection(bal, cli, lapi)
+	geoHandlers := geolocation.NewGeolocationHandlersCollection(bal, c, cli, lapi)
 
 	router.GET("/ok", withLogging(catalystApiHandlers.Ok()))
 	router.GET("/healthcheck", withLogging(catalystApiHandlers.Healthcheck()))

api/http_internal.go

@@ -68,12 +68,12 @@ func NewCatalystAPIRouterInternal(cli config.Cli, vodEngine *pipeline.Coordinato
 		Server:      cli.APIServer,
 		AccessToken: cli.APIToken,
 	})
-	geoHandlers := geolocation.NewGeolocationHandlersCollection(bal, cli, lapi)
+	geoHandlers := geolocation.NewGeolocationHandlersCollection(bal, c, cli, lapi)
 
 	spkiPublicKey, _ := crypto.ConvertToSpki(cli.VodDecryptPublicKey)
 
 	catalystApiHandlers := &handlers.CatalystAPIHandlersCollection{VODEngine: vodEngine}
-	eventsHandler := handlers.NewEventsHandlersCollection(c, mapic, bal)
+	eventsHandler := &handlers.EventsHandlersCollection{Cluster: c}
 	ffmpegSegmentingHandlers := &ffmpeg.HandlersCollection{VODEngine: vodEngine}
 	accessControlHandlers := accesscontrol.NewAccessControlHandlersCollection(cli, mapic)
 	analyticsHandlers := analytics.NewAnalyticsHandler(metricsDB)
@@ -109,14 +109,9 @@ func NewCatalystAPIRouterInternal(cli config.Cli, vodEngine *pipeline.Coordinato
 		),
 	)
 
-	// Handler to get members Catalyst API => Catalyst
-	router.GET("/api/serf/members", withLogging(adminHandlers.MembersHandler()))
-	// Public handler to propagate an event to all Catalyst nodes, execute from Studio API => Catalyst
+	// Public handler to propagate an event to all Catalyst nodes
 	router.POST("/api/events", withLogging(eventsHandler.Events()))
 
-	// Handler to forward the user event from Catalyst => Catalyst API
-	router.POST("/api/serf/receiveUserEvent", withLogging(eventsHandler.ReceiveUserEvent()))
-
 	// Public GET handler to retrieve the public key for vod encryption
 	router.GET("/api/pubkey", withLogging(encryptionHandlers.PublicKeyHandler()))
 
@@ -135,6 +130,9 @@ func NewCatalystAPIRouterInternal(cli config.Cli, vodEngine *pipeline.Coordinato
 	// Endpoint to receive segments and manifests that ffmpeg produces
 	router.POST("/api/ffmpeg/:id/:filename", withLogging(ffmpegSegmentingHandlers.NewFile()))
 
+	// Temporary endpoint for admin queries
+	router.GET("/admin/members", withLogging(adminHandlers.MembersHandler()))
+
 	return router
 }
 

cluster/cluster.go

@@ -182,25 +182,10 @@ func (c *ClusterImpl) retryJoin(ctx context.Context) {
 }
 
 func (c *ClusterImpl) MembersFiltered(filter map[string]string, status, name string) ([]Member, error) {
-	return FilterMembers(toClusterMembers(c.serf.Members()), filter, status, name)
-}
-
-func toClusterMembers(members []serf.Member) []Member {
-	var nodes []Member
-	for _, member := range members {
-		nodes = append(nodes, Member{
-			Name:   member.Name,
-			Tags:   member.Tags,
-			Status: member.Status.String(),
-		})
-	}
-	return nodes
-}
-
-func FilterMembers(all []Member, filter map[string]string, status string, name string) ([]Member, error) {
-	var nodes []Member
+	all := c.serf.Members()
+	nodes := []Member{}
 	for _, member := range all {
-		if status != "" && status != member.Status {
+		if status != "" && status != member.Status.String() {
 			continue
 		}
 		if name != "" && name != member.Name {
@@ -215,7 +200,11 @@ func FilterMembers(all []Member, filter map[string]string, status string, name s
 			}
 		}
 		if matches {
-			nodes = append(nodes, member)
+			nodes = append(nodes, Member{
+				Name:   member.Name,
+				Tags:   member.Tags,
+				Status: member.Status.String(),
+			})
 		}
 	}
 	return nodes, nil

config/cli.go

@@ -71,7 +71,6 @@ type Cli struct {
 	KafkaUser                 string
 	KafkaPassword             string
 	AnalyticsKafkaTopic       string
-	SerfMembersEndpoint       string
 
 	// mapping playbackId to value between 0.0 to 100.0
 	CdnRedirectPlaybackPct             map[string]float64

handlers/events.go

@@ -3,43 +3,25 @@ package handlers
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/golang/glog"
 	"github.com/hashicorp/serf/serf"
 	"github.com/julienschmidt/httprouter"
-	"github.com/livepeer/catalyst-api/balancer"
 	"github.com/livepeer/catalyst-api/cluster"
 	"github.com/livepeer/catalyst-api/errors"
-	"github.com/livepeer/catalyst-api/events"
-	mistapiconnector "github.com/livepeer/catalyst-api/mapic"
 	"github.com/xeipuuv/gojsonschema"
 	"io"
 	"net/http"
-	"strings"
 )
 
 type EventsHandlersCollection struct {
-	cluster cluster.Cluster
-
-	mapic mistapiconnector.IMac
-	bal   balancer.Balancer
-}
-
-type Event struct {
-	Resource   string `json:"resource"`
-	PlaybackID string `json:"playback_id"`
+	Cluster cluster.Cluster
 }
 
-func NewEventsHandlersCollection(cluster cluster.Cluster, mapic mistapiconnector.IMac, bal balancer.Balancer) *EventsHandlersCollection {
-	return &EventsHandlersCollection{
-		cluster: cluster,
-		mapic:   mapic,
-		bal:     bal,
+func (d *EventsHandlersCollection) Events() httprouter.Handle {
+	type Event struct {
+		Resource   string `json:"resource"`
+		PlaybackID string `json:"playback_id"`
 	}
-}
 
-// Events is a handler called by Studio API to send an event, e.g., to refresh a stream or nuke a stream.
-// This event is then propagated to all Serf nodes and then forwarded to catalyst-api and handled by ReceiveUserEvent().
-func (d *EventsHandlersCollection) Events() httprouter.Handle {
 	schema := inputSchemasCompiled["Event"]
 	return func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
 		payload, err := io.ReadAll(req.Body)
@@ -62,7 +44,7 @@ func (d *EventsHandlersCollection) Events() httprouter.Handle {
 			return
 		}
 
-		err = d.cluster.BroadcastEvent(serf.UserEvent{
+		err = d.Cluster.BroadcastEvent(serf.UserEvent{
 			Name:     fmt.Sprintf("%s-%s", event.Resource, event.PlaybackID),
 			Payload:  payload,
 			Coalesce: true,
@@ -74,51 +56,3 @@ func (d *EventsHandlersCollection) Events() httprouter.Handle {
 		}
 	}
 }
-
-// ReceiveUserEvent is a handler to receive Serf events from Catalyst.
-// The idea is that:
-// 1. Studio API sends an event to Catalyst (received by Events() handler)
-// 2. Events() handler propagates the event to all Serf nodes
-// 3. Each Serf node sends tne event to its corresponding catalyst-api instance (to the ReceiveUserEvent() handler)
-// 4. ReceiveUserEvent() handler processes the event
-func (c *EventsHandlersCollection) ReceiveUserEvent() httprouter.Handle {
-	return func(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
-		userEventPayload, err := io.ReadAll(r.Body)
-		if err != nil {
-			glog.Errorf("cannot read payload: %s", err)
-			return
-		}
-		e, err := events.Unmarshal(userEventPayload)
-		if err != nil {
-			glog.Errorf("cannot unmarshal received serf event %v: %s", userEventPayload, err)
-			return
-		}
-		switch event := e.(type) {
-		case *events.StreamEvent:
-			glog.V(5).Infof("received serf StreamEvent: %v", event.PlaybackID)
-			c.mapic.RefreshStreamIfNeeded(event.PlaybackID)
-		case *events.NukeEvent:
-			glog.V(5).Infof("received serf NukeEvent: %v", event.PlaybackID)
-			c.mapic.NukeStream(event.PlaybackID)
-			return
-		case *events.StopSessionsEvent:
-			glog.V(5).Infof("received serf StopSessionsEvent: %v", event.PlaybackID)
-			c.mapic.StopSessions(event.PlaybackID)
-			return
-		case *events.NodeUpdateEvent:
-			if glog.V(5) {
-				glog.Infof("received serf NodeUpdateEvent. Node: %s. Length: %d bytes. Ingest Streams: %v. Non-Ingest Streams: %v", event.NodeID, len(userEventPayload), strings.Join(event.GetIngestStreams(), ","), strings.Join(event.GetStreams(), ","))
-			}
-
-			c.bal.UpdateNodes(event.NodeID, event.NodeMetrics)
-			for _, stream := range event.GetStreams() {
-				c.bal.UpdateStreams(event.NodeID, stream, false)
-			}
-			for _, stream := range event.GetIngestStreams() {
-				c.bal.UpdateStreams(event.NodeID, stream, true)
-			}
-		default:
-			glog.Errorf("unsupported serf event: %v", e)
-		}
-	}
-}

handlers/events_test.go

@@ -5,7 +5,6 @@ import (
 	"github.com/hashicorp/serf/serf"
 	"github.com/julienschmidt/httprouter"
 	mockcluster "github.com/livepeer/catalyst-api/mocks/cluster"
-	mock_mistapiconnector "github.com/livepeer/catalyst-api/mocks/mistapiconnector"
 	"github.com/stretchr/testify/require"
 	"net/http"
 	"net/http/httptest"
@@ -63,7 +62,7 @@ func TestEventHandler(t *testing.T) {
 		return nil
 	}).AnyTimes()
 
-	catalystApiHandlers := NewEventsHandlersCollection(mc, nil, nil)
+	catalystApiHandlers := EventsHandlersCollection{Cluster: mc}
 	router := httprouter.New()
 	router.POST("/events", catalystApiHandlers.Events())
 
@@ -75,65 +74,3 @@ func TestEventHandler(t *testing.T) {
 		require.Equal(rr.Result().StatusCode, tt.wantHttpCode)
 	}
 }
-
-func TestReceiveUserEventHandler(t *testing.T) {
-	require := require.New(t)
-	playbackId := "123456789"
-
-	tests := []struct {
-		name           string
-		requestBody    string
-		functionCalled string
-	}{
-		{
-			name: "Refresh Stream",
-			requestBody: `{
-				"resource": "stream",
-				"playback_id": "123456789"
-			}`,
-			functionCalled: "RefreshStreamIfNeeded",
-		},
-		{
-			name: "Nuke Stream",
-			requestBody: `{
-				"resource": "nuke",
-				"playback_id": "123456789"
-			}`,
-			functionCalled: "NukeStream",
-		},
-		{
-			name: "Stop Sessions",
-			requestBody: `{
-				"resource": "stopSessions",
-				"playback_id": "123456789"
-			}`,
-			functionCalled: "StopSessions",
-		},
-	}
-
-	ctrl := gomock.NewController(t)
-	mac := mock_mistapiconnector.NewMockIMac(ctrl)
-
-	catalystApiHandlers := NewEventsHandlersCollection(nil, mac, nil)
-	router := httprouter.New()
-	router.POST("/receiveUserEvent", catalystApiHandlers.ReceiveUserEvent())
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			switch tt.functionCalled {
-			case "RefreshStreamIfNeeded":
-				mac.EXPECT().RefreshStreamIfNeeded(playbackId).Times(1)
-			case "NukeStream":
-				mac.EXPECT().NukeStream(playbackId).Times(1)
-			case "StopSessions":
-				mac.EXPECT().StopSessions(playbackId).Times(1)
-			}
-
-			req, _ := http.NewRequest("POST", "/receiveUserEvent", strings.NewReader(tt.requestBody))
-			rr := httptest.NewRecorder()
-			router.ServeHTTP(rr, req)
-
-			require.Equal(rr.Result().StatusCode, 200)
-		})
-	}
-}

handlers/geolocation/geolocation.go

@@ -67,14 +67,16 @@ func (l *streamPullRateLimit) mark(playbackID string) {
 
 type GeolocationHandlersCollection struct {
 	Balancer            balancer.Balancer
+	Cluster             cluster.Cluster
 	Config              config.Cli
 	Lapi                *api.Client
 	streamPullRateLimit *streamPullRateLimit
 }
 
-func NewGeolocationHandlersCollection(balancer balancer.Balancer, config config.Cli, lapi *api.Client) *GeolocationHandlersCollection {
+func NewGeolocationHandlersCollection(balancer balancer.Balancer, cluster cluster.Cluster, config config.Cli, lapi *api.Client) *GeolocationHandlersCollection {
 	return &GeolocationHandlersCollection{
 		Balancer:            balancer,
+		Cluster:             cluster,
 		Config:              config,
 		Lapi:                lapi,
 		streamPullRateLimit: newStreamPullRateLimit(streamSourceRetryInterval),
@@ -242,7 +244,7 @@ func (c *GeolocationHandlersCollection) resolveNodeURL(streamURL string) (string
 }
 
 func (c *GeolocationHandlersCollection) clusterMember(filter map[string]string, status, name string) (cluster.Member, error) {
-	members, err := c.membersFiltered(filter, "", name)
+	members, err := c.Cluster.MembersFiltered(filter, "", name)
 	if err != nil {
 		return cluster.Member{}, err
 	}
@@ -402,7 +404,7 @@ func (c *GeolocationHandlersCollection) getStreamPull(playbackID string, retryCo
 }
 
 func (c *GeolocationHandlersCollection) sendPlaybackRequestAsync(playbackID string, region string) {
-	members, err := c.membersFiltered(map[string]string{"region": region}, "", "")
+	members, err := c.Cluster.MembersFiltered(map[string]string{"region": region}, "", "")
 	if err != nil || len(members) == 0 {
 		glog.Errorf("Error fetching member list: %v", err)
 		return
@@ -420,23 +422,6 @@ func (c *GeolocationHandlersCollection) sendPlaybackRequestAsync(playbackID stri
 	}()
 }
 
-func (c *GeolocationHandlersCollection) membersFiltered(filter map[string]string, status, name string) ([]cluster.Member, error) {
-	resp, err := http.Get(c.Config.SerfMembersEndpoint)
-	if err != nil {
-		return []cluster.Member{}, err
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusOK {
-		return []cluster.Member{}, fmt.Errorf("failed to get members: %s", resp.Status)
-	}
-	var members []cluster.Member
-	if err := json.NewDecoder(resp.Body).Decode(&members); err != nil {
-		return []cluster.Member{}, err
-	}
-
-	return cluster.FilterMembers(members, filter, status, name)
-}
-
 func parsePlus(plusString string) (string, string) {
 	slice := strings.Split(plusString, "+")
 	prefix := ""