Add option for running catalyst-api as a stateless service

.github/workflows/docker.yaml

@@ -0,0 +1,76 @@
+name: Build and publish catalyst-api docker images
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+      - dev
+    tags:
+      - "v*"
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  docker:
+    name: Build and publish docker image
+    runs-on: ubuntu-latest
+    permissions:
+      packages: write
+      contents: read
+
+    steps:
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v2
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v2
+
+      - name: Login to DockerHub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.CI_DOCKERHUB_USERNAME }}
+          password: ${{ secrets.CI_DOCKERHUB_TOKEN }}
+
+      - name: Log in to the Container registry
+        uses: docker/login-action@v2
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ github.token }}
+
+      - name: Extract metadata (tags, labels) for Docker
+        id: meta
+        uses: docker/metadata-action@v4
+        with:
+          images: |
+            livepeerci/${{ github.event.repository.name }}
+            ghcr.io/${{ github.repository }}
+          tags: |
+            type=sha
+            type=ref,event=pr
+            type=ref,event=tag
+            type=sha,format=long
+            type=ref,event=branch
+            type=semver,pattern={{version}},prefix=v
+            type=semver,pattern={{major}}.{{minor}},prefix=v
+            type=raw,value=latest,enable={{is_default_branch}}
+            type=raw,value=${{ github.event.pull_request.head.ref }}
+            type=raw,value=stable,enable=${{ startsWith(github.event.ref, 'refs/tags/v') }}
+
+      - name: Build and push
+        uses: docker/build-push-action@v4
+        with:
+          platforms: linux/amd64, linux/arm64
+          push: true
+          build-args: |
+            GIT_VERSION=${{ github.ref_type == 'tag' && github.ref_name || github.event.pull_request.head.sha || github.sha }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
+
+      - name: Notify new build upload
+        run: curl -X POST https://holy-bread-207a.livepeer.workers.dev

Dockerfile

@@ -0,0 +1,44 @@
+FROM	golang:1-bullseye	as	gobuild
+
+ARG TARGETARCH
+
+# Download c2patool needed to sign our C2PA manifest
+# We download it from any of our previous builds, because building c2patool from source is very slow with QEMU
+RUN	apt update && apt install -yqq \
+	curl \
+	ca-certificates \
+	&& curl https://build.livepeer.live/c2patool/0.6.2/c2patool-linux-${TARGETARCH}.tar.gz -o /c2patool.tar.gz \
+	&& tar xzf /c2patool.tar.gz
+
+WORKDIR	/src
+
+ADD	go.mod go.sum	./
+RUN	go mod download
+
+ADD	.	.
+RUN make build
+
+ARG	GIT_VERSION
+ENV	GIT_VERSION="${GIT_VERSION}"
+
+FROM	ubuntu:22.04	AS	catalyst
+
+ENV	DEBIAN_FRONTEND=noninteractive
+
+LABEL	maintainer="Amritanshu Varshney <[email protected]>"
+
+ARG	BUILD_TARGET
+
+RUN	apt update && apt install -yqq wget software-properties-common
+
+RUN	apt update && apt install -yqq \
+	curl \
+	ca-certificates \
+	procps \
+	vnstat \
+	&& rm -rf /var/lib/apt/lists/*
+
+COPY --from=gobuild		/src/build/catalyst-api /bin/catalyst-api
+COPY --from=gobuild		/go/c2patool /bin/
+
+CMD ["/bin/catalyst-api"]

api/http.go

@@ -8,7 +8,6 @@ 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"
@@ -21,8 +20,8 @@ import (
 	"github.com/livepeer/go-api-client"
 )
 
-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)
+func ListenAndServe(ctx context.Context, cli config.Cli, vodEngine *pipeline.Coordinator, bal balancer.Balancer, mapic mistapiconnector.IMac) error {
+	router := NewCatalystAPIRouter(cli, vodEngine, bal, mapic)
 	server := http.Server{Addr: cli.HTTPAddress, Handler: router}
 	ctx, cancel := context.WithCancel(ctx)
 
@@ -48,7 +47,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, c cluster.Cluster, mapic mistapiconnector.IMac) *httprouter.Router {
+func NewCatalystAPIRouter(cli config.Cli, vodEngine *pipeline.Coordinator, bal balancer.Balancer, mapic mistapiconnector.IMac) *httprouter.Router {
 	router := httprouter.New()
 	withLogging := middleware.LogRequest()
 	withCORS := middleware.AllowCORS()
@@ -59,7 +58,7 @@ func NewCatalystAPIRouter(cli config.Cli, vodEngine *pipeline.Coordinator, bal b
 		AccessToken: cli.APIToken,
 	})
 	catalystApiHandlers := &handlers.CatalystAPIHandlersCollection{VODEngine: vodEngine}
-	geoHandlers := geolocation.NewGeolocationHandlersCollection(bal, c, cli, lapi)
+	geoHandlers := geolocation.NewGeolocationHandlersCollection(bal, 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, c, cli, lapi)
+	geoHandlers := geolocation.NewGeolocationHandlersCollection(bal, cli, lapi)
 
 	spkiPublicKey, _ := crypto.ConvertToSpki(cli.VodDecryptPublicKey)
 
 	catalystApiHandlers := &handlers.CatalystAPIHandlersCollection{VODEngine: vodEngine}
-	eventsHandler := &handlers.EventsHandlersCollection{Cluster: c}
+	eventsHandler := handlers.NewEventsHandlersCollection(c, mapic, bal)
 	ffmpegSegmentingHandlers := &ffmpeg.HandlersCollection{VODEngine: vodEngine}
 	accessControlHandlers := accesscontrol.NewAccessControlHandlersCollection(cli, mapic)
 	analyticsHandlers := analytics.NewAnalyticsHandler(metricsDB)
@@ -84,54 +84,59 @@ func NewCatalystAPIRouterInternal(cli config.Cli, vodEngine *pipeline.Coordinato
 	// Simple endpoint for healthchecks
 	router.GET("/ok", withLogging(catalystApiHandlers.Ok()))
 
-	var metricsHandlers []http.Handler
-	if cli.ShouldMapic() {
-		metricsHandlers = append(metricsHandlers, mapic.MetricsHandler())
+	if cli.IsClusterMode() {
+		router.GET("/api/serf/members", withLogging(adminHandlers.MembersHandler()))
+		// Public handler to propagate an event to all Catalyst nodes
+		router.POST("/api/events", withLogging(eventsHandler.Events()))
 	}
-	if cli.MistPrometheus != "" {
-		// Enable Mist metrics enrichment
-		metricsHandlers = append(metricsHandlers, mapic.MistMetricsHandler())
-	}
-	metricsHandlers = append(metricsHandlers, promhttp.Handler())
-	// Hacky combined metrics handler. To be refactored away with mapic.
-	router.GET("/metrics", concatHandlers(metricsHandlers...))
-
-	// Public Catalyst API
-	router.POST("/api/vod",
-		withLogging(
-			withAuth(
-				cli.APIToken,
-				withCapacityChecking(
-					vodEngine,
-					catalystApiHandlers.UploadVOD(),
+
+	if cli.IsApiMode() {
+		var metricsHandlers []http.Handler
+		if cli.ShouldMapic() {
+			metricsHandlers = append(metricsHandlers, mapic.MetricsHandler())
+		}
+		if cli.MistPrometheus != "" {
+			// Enable Mist metrics enrichment
+			metricsHandlers = append(metricsHandlers, mapic.MistMetricsHandler())
+		}
+		metricsHandlers = append(metricsHandlers, promhttp.Handler())
+		// Hacky combined metrics handler. To be refactored away with mapic.
+		router.GET("/metrics", concatHandlers(metricsHandlers...))
+
+		// Public Catalyst API
+		router.POST("/api/vod",
+			withLogging(
+				withAuth(
+					cli.APIToken,
+					withCapacityChecking(
+						vodEngine,
+						catalystApiHandlers.UploadVOD(),
+					),
 				),
 			),
-		),
-	)
-
-	// Public handler to propagate an event to all Catalyst nodes
-	router.POST("/api/events", withLogging(eventsHandler.Events()))
+		)
 
-	// Public GET handler to retrieve the public key for vod encryption
-	router.GET("/api/pubkey", withLogging(encryptionHandlers.PublicKeyHandler()))
+		// Handlers for remote serf interaction
+		router.POST("/api/serf/receiveUserEvent", withLogging(eventsHandler.ReceiveUserEvent()))
 
-	// Endpoint to receive "Triggers" (callbacks) from Mist
-	router.POST("/api/mist/trigger", withLogging(mistCallbackHandlers.Trigger()))
+		// Public GET handler to retrieve the public key for vod encryption
+		router.GET("/api/pubkey", withLogging(encryptionHandlers.PublicKeyHandler()))
 
-	// Handler for STREAM_SOURCE triggers
-	broker.OnStreamSource(geoHandlers.HandleStreamSource)
+		// Endpoint to receive "Triggers" (callbacks) from Mist
+		router.POST("/api/mist/trigger", withLogging(mistCallbackHandlers.Trigger()))
 
-	// Handler for USER_NEW triggers
-	broker.OnUserNew(accessControlHandlers.HandleUserNew)
+		// Handler for STREAM_SOURCE triggers
+		broker.OnStreamSource(geoHandlers.HandleStreamSource)
 
-	// Handler for USER_END triggers.
-	broker.OnUserEnd(analyticsHandlers.HandleUserEnd)
+		// Handler for USER_NEW triggers
+		broker.OnUserNew(accessControlHandlers.HandleUserNew)
 
-	// Endpoint to receive segments and manifests that ffmpeg produces
-	router.POST("/api/ffmpeg/:id/:filename", withLogging(ffmpegSegmentingHandlers.NewFile()))
+		// Handler for USER_END triggers.
+		broker.OnUserEnd(analyticsHandlers.HandleUserEnd)
 
-	// Temporary endpoint for admin queries
-	router.GET("/admin/members", withLogging(adminHandlers.MembersHandler()))
+		// Endpoint to receive segments and manifests that ffmpeg produces
+		router.POST("/api/ffmpeg/:id/:filename", withLogging(ffmpegSegmentingHandlers.NewFile()))
+	}
 
 	return router
 }

balancer/mist/mist_balancer.go

@@ -26,27 +26,32 @@ var mistUtilLoadLoopTimeout = 2 * time.Minute
 
 type MistBalancer struct {
 	config   *balancer.Config
-	cmd      *exec.Cmd
 	endpoint string
 	// Blocks until initial startup
 	startupOnce  sync.Once
 	startupError error
 }
 
 // create a new load balancer instance
-func NewBalancer(config *balancer.Config) balancer.Balancer {
+func NewLocalBalancer(config *balancer.Config) balancer.Balancer {
 	_, err := exec.LookPath("MistUtilLoad")
 	if err != nil {
 		glog.Warning("MistUtilLoad not found, not doing meaningful balancing")
 		return &balancer.BalancerStub{}
 	}
 	return &MistBalancer{
 		config:   config,
-		cmd:      nil,
 		endpoint: fmt.Sprintf("http://127.0.0.1:%d", config.MistUtilLoadPort),
 	}
 }
 
+func NewRemoteBalancer(config *balancer.Config) balancer.Balancer {
+	return &MistBalancer{
+		config:   config,
+		endpoint: fmt.Sprintf("http://%s:%d", config.MistHost, config.MistUtilLoadPort),
+	}
+}
+
 // start this load balancer instance, execing MistUtilLoad if necessary
 func (b *MistBalancer) Start(ctx context.Context) error {
 	b.killPreviousBalancer(ctx)
@@ -288,17 +293,17 @@ func (b *MistBalancer) isBalancerRunning(ctx context.Context) bool {
 func (b *MistBalancer) execBalancer(ctx context.Context, balancerArgs []string) error {
 	args := append(balancerArgs, "-p", fmt.Sprintf("%d", b.config.MistUtilLoadPort), "-g", "4")
 	glog.Infof("Running MistUtilLoad with %v", args)
-	b.cmd = exec.CommandContext(ctx, "MistUtilLoad", args...)
+	cmd := exec.CommandContext(ctx, "MistUtilLoad", args...)
 
-	b.cmd.Stdout = os.Stdout
-	b.cmd.Stderr = os.Stderr
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
 
-	err := b.cmd.Start()
+	err := cmd.Start()
 	if err != nil {
 		return err
 	}
 
-	err = b.cmd.Wait()
+	err = cmd.Wait()
 	if err != nil {
 		return err
 	}

balancer/mist/mist_balancer_test.go

@@ -31,7 +31,6 @@ func start(t *testing.T) (*MistBalancer, *mockMistUtilLoad) {
 			OwnRegion:          "fra",
 			OwnRegionTagAdjust: 1000,
 		},
-		cmd:      nil,
 		endpoint: mul.Server.URL,
 	}
 	// Mock startup loop

clients/mist_client.go

@@ -37,29 +37,27 @@ type MistAPIClient interface {
 	DeleteStream(streamName string) error
 	NukeStream(streamName string) error
 	StopSessions(streamName string) error
-	AddTrigger(streamName []string, triggerName string, sync bool) error
+	AddTrigger(streamName []string, triggerName, triggerCallback string, sync bool) error
 	DeleteTrigger(streamName []string, triggerName string) error
 	GetStreamInfo(streamName string) (MistStreamInfo, error)
 	GetState() (MistState, error)
 }
 
 type MistClient struct {
-	ApiUrl          string
-	Username        string
-	Password        string
-	HttpReqUrl      string
-	TriggerCallback string
-	configMu        sync.Mutex
-	cache           *cache.Cache
+	ApiUrl     string
+	Username   string
+	Password   string
+	HttpReqUrl string
+	configMu   sync.Mutex
+	cache      *cache.Cache
 }
 
-func NewMistAPIClient(user, password, host string, port int, ownURL string) MistAPIClient {
+func NewMistAPIClient(user, password, host string, port int) MistAPIClient {
 	mist := &MistClient{
-		ApiUrl:          fmt.Sprintf("http://%s:%d", host, port),
-		Username:        user,
-		Password:        password,
-		TriggerCallback: ownURL,
-		cache:           cache.New(defaultCacheExpiration, cacheCleanupInterval),
+		ApiUrl:   fmt.Sprintf("http://%s:%d", host, port),
+		Username: user,
+		Password: password,
+		cache:    cache.New(defaultCacheExpiration, cacheCleanupInterval),
 	}
 	return mist
 }
@@ -310,15 +308,15 @@ func (mc *MistClient) StopSessions(streamName string) error {
 // 3. Add a new trigger (or update the existing one)
 // 4. Override the triggers
 // 5. Release the lock
-func (mc *MistClient) AddTrigger(streamNames []string, triggerName string, sync bool) error {
+func (mc *MistClient) AddTrigger(streamNames []string, triggerName, triggerCallback string, sync bool) error {
 	mc.configMu.Lock()
 	defer mc.configMu.Unlock()
 
 	triggers, err := mc.getCurrentTriggers()
 	if err != nil {
 		return err
 	}
-	c := commandAddTrigger(streamNames, triggerName, mc.TriggerCallback, triggers, sync)
+	c := commandAddTrigger(streamNames, triggerName, triggerCallback, triggers, sync)
 	resp, err := mc.sendCommand(c)
 	return validateAddTrigger(streamNames, triggerName, resp, err, sync)
 }