Live Runner

byoc/types.go

@@ -3,6 +3,7 @@ package byoc
 import (
 	"context"
 	"crypto/tls"
+	"encoding/binary"
 	"errors"
 	"math/big"
 	gonet "net"
@@ -279,3 +280,65 @@ type byocLiveRequestParams struct {
 	// when the write for the last segment started
 	lastSegmentTime time.Time
 }
+
+// Prevents cross-protocol signature replay.
+const BYOCJobSigV1Prefix = "LP_BYOC_JOB_V1\x00\x00"
+
+// BYOCJobSigningInput holds the fields that are bound into a BYOC job signature.
+type BYOCJobSigningInput struct {
+	ID             string
+	Capability     string
+	Request        string
+	Parameters     string
+	TimeoutSeconds int
+}
+
+// FlattenBYOCJob produces a deterministic binary representation of a BYOC job
+// for signing, similar to SegTranscodingMetadata.Flatten() used by LV2V.
+//
+// Wire format:
+//
+//	version(16) || timeout(4,BE) || len(id)(4,BE) || id || len(cap)(4,BE) || cap
+//	           || len(req)(4,BE) || req || len(params)(4,BE) || params
+func FlattenBYOCJob(job *BYOCJobSigningInput) []byte {
+	idBytes := []byte(job.ID)
+	capBytes := []byte(job.Capability)
+	reqBytes := []byte(job.Request)
+	paramsBytes := []byte(job.Parameters)
+
+	size := 16 + 4 +
+		4 + len(idBytes) +
+		4 + len(capBytes) +
+		4 + len(reqBytes) +
+		4 + len(paramsBytes)
+
+	buf := make([]byte, size)
+	offset := 0
+
+	copy(buf[offset:], []byte(BYOCJobSigV1Prefix))
+	offset += 16
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(job.TimeoutSeconds))
+	offset += 4
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(idBytes)))
+	offset += 4
+	copy(buf[offset:], idBytes)
+	offset += len(idBytes)
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(capBytes)))
+	offset += 4
+	copy(buf[offset:], capBytes)
+	offset += len(capBytes)
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(reqBytes)))
+	offset += 4
+	copy(buf[offset:], reqBytes)
+	offset += len(reqBytes)
+
+	binary.BigEndian.PutUint32(buf[offset:], uint32(len(paramsBytes)))
+	offset += 4
+	copy(buf[offset:], paramsBytes)
+
+	return buf
+}

clog/clog.go

@@ -197,6 +197,23 @@ func infof(ctx context.Context, lastErr bool, publicLog bool, format string, arg
 // Example: Info(ctx, "hello", "key1", value1, "key2", value2)
 // This will log: "hello key1=value1 key2=value2"
 func Info(ctx context.Context, msg string, keyvals ...interface{}) {
+	msg = formatKeyvals(msg, keyvals...)
+	infof(ctx, false, false, msg)
+}
+
+// Warning logs a warning message with key-value pairs in a slog-like style.
+func Warning(ctx context.Context, msg string, keyvals ...interface{}) {
+	msg = formatKeyvals(msg, keyvals...)
+	Warningf(ctx, "%s", msg)
+}
+
+// Error logs an error message with key-value pairs in a slog-like style.
+func Error(ctx context.Context, msg string, keyvals ...interface{}) {
+	msg = formatKeyvals(msg, keyvals...)
+	Errorf(ctx, "%s", msg)
+}
+
+func formatKeyvals(msg string, keyvals ...interface{}) string {
 	if len(keyvals)%2 != 0 {
 		keyvals = append(keyvals[:len(keyvals)-1], "MISSING", keyvals[len(keyvals)-1])
 	}
@@ -225,7 +242,7 @@ func Info(ctx context.Context, msg string, keyvals ...interface{}) {
 		}
 	}
 
-	infof(ctx, false, false, sb.String())
+	return sb.String()
 }
 
 // V returns a Verbose instance for conditional logging at the specified level

cmd/livepeer/starter/flags.go

@@ -56,6 +56,9 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
 	// AI:
 	cfg.AIServiceRegistry = fs.Bool("aiServiceRegistry", *cfg.AIServiceRegistry, "Set to true to use an AI ServiceRegistry contract address")
 	cfg.AIWorker = fs.Bool("aiWorker", *cfg.AIWorker, "Set to true to run an AI worker")
+	cfg.AIServerless = fs.Bool("aiServerless", *cfg.AIServerless, "Set to true to use serverless AI worker")
+	cfg.UseLiveRunners = fs.Bool("useLiveRunners", *cfg.UseLiveRunners, "Set to true to use LiveRunner-backed runners for supported pipelines")
+	cfg.LiveRunnerConfig = fs.String("liveRunnerConfig", *cfg.LiveRunnerConfig, "Path to JSON config file for static LiveRunner registration")
 	cfg.AIModels = fs.String("aiModels", *cfg.AIModels, "Set models (pipeline:model_id) for AI worker to load upon initialization")
 	cfg.AIModelsDir = fs.String("aiModelsDir", *cfg.AIModelsDir, "Set directory where AI model weights are stored")
 	cfg.AIRunnerImage = fs.String("aiRunnerImage", *cfg.AIRunnerImage, "[Deprecated] Specify the base Docker image for the AI runner. Example: livepeer/ai-runner:0.0.1. Use -aiRunnerImageOverrides instead.")
@@ -67,6 +70,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
 
 	// Live AI:
 	cfg.MediaMTXApiPassword = fs.String("mediaMTXApiPassword", "", "HTTP basic auth password for MediaMTX API requests")
+	cfg.LiveRunnerAddr = fs.String("liveRunnerAddr", *cfg.LiveRunnerAddr, "Base URL used by live runners for heartbeat, control-plane callbacks, and internal trickle channels. Must be a full URL such as http://go-livepeer:8935")
 	cfg.LiveAITrickleHostForRunner = fs.String("liveAITrickleHostForRunner", "", "Trickle Host used by AI Runner; It's used to overwrite the publicly available Trickle Host")
 	cfg.LiveAIAuthApiKey = fs.String("liveAIAuthApiKey", "", "API key to use for Live AI authentication requests")
 	cfg.LiveAIHeartbeatURL = fs.String("liveAIHeartbeatURL", "", "Base URL for Live AI heartbeat requests")

cmd/livepeer/starter/starter.go

@@ -28,6 +28,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/golang/glog"
+	"github.com/livepeer/go-livepeer/ai/runner"
 	"github.com/livepeer/go-livepeer/ai/worker"
 	"github.com/livepeer/go-livepeer/build"
 	"github.com/livepeer/go-livepeer/common"
@@ -97,6 +98,9 @@ type LivepeerConfig struct {
 	Transcoder                 *bool
 	AIServiceRegistry          *bool
 	AIWorker                   *bool
+	AIServerless               *bool
+	UseLiveRunners             *bool
+	LiveRunnerConfig           *string
 	Gateway                    *bool
 	Broadcaster                *bool
 	OrchSecret                 *string
@@ -164,6 +168,7 @@ type LivepeerConfig struct {
 	AuthWebhookURL             *string
 	LiveAIAuthWebhookURL       *string
 	LiveAITrickleHostForRunner *string
+	LiveRunnerAddr             *string
 	OrchWebhookURL             *string
 	OrchBlacklist              *string
 	OrchMinLivepeerVersion     *string
@@ -237,6 +242,9 @@ func DefaultLivepeerConfig() LivepeerConfig {
 	// AI:
 	defaultAIServiceRegistry := false
 	defaultAIWorker := false
+	defaultAIServerless := false
+	defaultUseLiveRunners := false
+	defaultLiveRunnerConfig := ""
 	defaultAIModels := ""
 	defaultAIModelsDir := ""
 	defaultAIRunnerImage := "livepeer/ai-runner:latest"
@@ -246,6 +254,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 	defaultAIMinRunnerVersion := "[]"
 	defaultAIRunnerImageOverrides := ""
 	defaultLiveAIAuthWebhookURL := ""
+	defaultLiveRunnerAddr := ""
 	defaultLivePaymentInterval := 5 * time.Second
 	defaultLiveOutSegmentTimeout := 0 * time.Second
 	defaultGatewayHost := ""
@@ -363,6 +372,9 @@ func DefaultLivepeerConfig() LivepeerConfig {
 		// AI:
 		AIServiceRegistry:        &defaultAIServiceRegistry,
 		AIWorker:                 &defaultAIWorker,
+		AIServerless:             &defaultAIServerless,
+		UseLiveRunners:           &defaultUseLiveRunners,
+		LiveRunnerConfig:         &defaultLiveRunnerConfig,
 		AIModels:                 &defaultAIModels,
 		AIModelsDir:              &defaultAIModelsDir,
 		AIRunnerImage:            &defaultAIRunnerImage,
@@ -372,6 +384,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 		AIMinRunnerVersion:       &defaultAIMinRunnerVersion,
 		AIRunnerImageOverrides:   &defaultAIRunnerImageOverrides,
 		LiveAIAuthWebhookURL:     &defaultLiveAIAuthWebhookURL,
+		LiveRunnerAddr:           &defaultLiveRunnerAddr,
 		LivePaymentInterval:      &defaultLivePaymentInterval,
 		LiveOutSegmentTimeout:    &defaultLiveOutSegmentTimeout,
 		GatewayHost:              &defaultGatewayHost,
@@ -1319,6 +1332,15 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 
 	var aiCaps []core.Capability
 	capabilityConstraints := make(core.PerCapabilityConstraints)
+	var aiModelConfigs []core.AIModelConfig
+
+	if *cfg.AIModels != "" {
+		aiModelConfigs, err = core.ParseAIModelConfigs(*cfg.AIModels)
+		if err != nil {
+			glog.Errorf("Error parsing -aiModels: %v", err)
+			return
+		}
+	}
 
 	if *cfg.AIWorker {
 		gpus := []string{}
@@ -1378,10 +1400,28 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 			}
 		}
 
-		n.AIWorker, err = worker.NewWorker(imageOverrides, *cfg.AIVerboseLogs, gpus, modelsDir, containerCreatorID)
-		if err != nil {
-			glog.Errorf("Error starting AI worker: %v", err)
-			return
+		if *cfg.AIServerless {
+			if len(aiModelConfigs) != 1 {
+				glog.Errorf("Serverless requires exactly one AI model config, got %d", len(aiModelConfigs))
+				return
+			}
+			config := aiModelConfigs[0]
+			if config.Pipeline != "live-video-to-video" || config.ModelID != "scope" {
+				glog.Errorf("Serverless only supports live-video-to-video/scope, got %s/%s", config.Pipeline, config.ModelID)
+				return
+			}
+
+			n.AIWorker, err = worker.NewServerlessWorker(strings.TrimSpace(config.URL), config.Capacity)
+			if err != nil {
+				glog.Errorf("Error starting Serverless AI worker: %v", err)
+				return
+			}
+		} else {
+			n.AIWorker, err = worker.NewWorker(imageOverrides, *cfg.AIVerboseLogs, gpus, modelsDir, containerCreatorID)
+			if err != nil {
+				glog.Errorf("Error starting AI worker: %v", err)
+				return
+			}
 		}
 
 		defer func() {
@@ -1397,20 +1437,15 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	}
 
 	if *cfg.AIModels != "" {
-		configs, err := core.ParseAIModelConfigs(*cfg.AIModels)
-		if err != nil {
-			glog.Errorf("Error parsing -aiModels: %v", err)
-			return
-		}
-
-		for _, config := range configs {
+		for _, config := range aiModelConfigs {
 			pipelineCap, err := core.PipelineToCapability(config.Pipeline)
 			if err != nil {
 				panic(fmt.Errorf("Pipeline is not valid capability: %v\n", config.Pipeline))
 			}
 			if *cfg.AIWorker {
 				modelConstraint := &core.ModelConstraint{Warm: config.Warm, Capacity: 1}
 				modelsCount := 1
+
 				if config.Capacity != 0 {
 					if config.URL == "" {
 						// Use multiple same configs if External Container is not used and capacity is set
@@ -1854,6 +1889,12 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	if cfg.LiveAITrickleHostForRunner != nil {
 		n.LiveAITrickleHostForRunner = *cfg.LiveAITrickleHostForRunner
 	}
+	if cfg.LiveRunnerAddr != nil && *cfg.LiveRunnerAddr != "" {
+		n.LiveRunnerAddr, err = parseLiveRunnerAddr(*cfg.LiveRunnerAddr)
+		if err != nil {
+			glog.Exitf("invalid -liveRunnerAddr: %v", err)
+		}
+	}
 	if cfg.LiveAICapRefreshModels != nil && *cfg.LiveAICapRefreshModels != "" {
 		glog.Warningf("The -liveAICapRefreshModels flag is deprecated, capacity is now available for all models, use -liveAICapReportInterval to set the interval for reporting capacity metrics")
 	}
@@ -1935,6 +1976,7 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 				DiscoveryTimeout:        *cfg.DiscoveryTimeout,
 				LiveAICapReportInterval: *cfg.LiveAICapReportInterval,
 				IgnoreCapacityCheck:     true,
+				UseDiscoveryEndpoint:    true,
 			}.New()
 			if err != nil {
 				exit("Could not create orchestrator pool with DB cache: %v", err)
@@ -1984,6 +2026,30 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 
 		orch := core.NewOrchestrator(s.LivepeerNode, timeWatcher)
 
+		if *cfg.UseLiveRunners || *cfg.LiveRunnerConfig != "" {
+			if n.OrchSecret == "" && *cfg.LiveRunnerConfig == "" {
+				glog.Exit("running with -useLiveRunners requires -orchSecret")
+			}
+			n.LiveRunnerManager = runner.NewLiveRunnerRegistry(runner.LiveRunnerRegistryConfig{
+				Host:    liveRunnerHost{RunnerHost: orch, LivepeerNode: n},
+				Onchain: *cfg.Network != "offchain",
+			})
+			if n.OrchSecret == "" {
+				glog.Warning("No -orchSecret configured; dynamic LiveRunner heartbeat registration is disabled")
+			}
+			if *cfg.LiveRunnerConfig != "" {
+				configJSON, err := os.ReadFile(*cfg.LiveRunnerConfig)
+				if err != nil {
+					glog.Exitf("error reading -liveRunnerConfig: %v", err)
+				}
+				registration, err := n.LiveRunnerManager.(*runner.LiveRunnerRegistry).RegisterStaticRunnersJSON(configJSON)
+				if err != nil {
+					glog.Exitf("error registering -liveRunnerConfig: %v", err)
+				}
+				glog.Infof("Registered %d static live runners from %s", len(registration.Runners), *cfg.LiveRunnerConfig)
+			}
+		}
+
 		go func() {
 			err = server.StartTranscodeServer(orch, *cfg.HttpAddr, s.HTTPMux, n.WorkDir, n.TranscoderManager != nil, n.AIWorkerManager != nil, n)
 			if err != nil {
@@ -1997,10 +2063,20 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 		// check whether or not the orchestrator is available
 		if *cfg.TestOrchAvail && doingWork {
 			time.Sleep(2 * time.Second)
-			orchAvail := server.CheckOrchestratorAvailability(orch)
+			var (
+				checkName string
+				orchAvail bool
+			)
+			if *cfg.UseLiveRunners || *cfg.LiveRunnerConfig != "" {
+				checkName = "discovery"
+				orchAvail = server.CheckOrchestratorDiscoveryAvailability(orch)
+			} else {
+				checkName = "grpc ping"
+				orchAvail = server.CheckOrchestratorAvailability(orch)
+			}
 			if !orchAvail {
 				// shut down orchestrator
-				glog.Infof("Orchestrator not available at %v (%v); shutting down", orch.ServiceURI(), *cfg.HttpAddr)
+				glog.Infof("Orchestrator not available at %v (%v) via %s check; shutting down", orch.ServiceURI(), *cfg.HttpAddr, checkName)
 				tc <- struct{}{}
 			}
 		}
@@ -2143,7 +2219,11 @@ func parseHeaderMap(raw string) map[string]string {
 	for _, header := range strings.Split(raw, ",") {
 		parts := strings.SplitN(header, ":", 2)
 		if len(parts) == 2 {
-			headers[parts[0]] = parts[1]
+			key := strings.TrimSpace(parts[0])
+			value := strings.TrimSpace(parts[1])
+			if key != "" {
+				headers[key] = value
+			}
 		}
 	}
 	return headers
@@ -2177,7 +2257,10 @@ func getServiceURI(n *core.LivepeerNode, serviceAddr string) (*url.URL, error) {
 			// special value to signal this node is not to be used for work
 			return url.Parse("")
 		}
-		return url.ParseRequestURI("https://" + serviceAddr)
+		if !strings.HasPrefix(serviceAddr, "http://") && !strings.HasPrefix(serviceAddr, "https://") {
+			serviceAddr = "https://" + serviceAddr
+		}
+		return url.ParseRequestURI(serviceAddr)
 	}
 
 	// Infer address
@@ -2504,3 +2587,30 @@ func exit(msg string, args ...any) {
 	glog.Errorf(msg, args...)
 	os.Exit(2)
 }
+
+type liveRunnerHost struct {
+	runner.RunnerHost
+	*core.LivepeerNode
+}
+
+func (h liveRunnerHost) LiveRunnerURI() *url.URL {
+	if h.LivepeerNode != nil && h.LivepeerNode.LiveRunnerAddr != nil {
+		v := *h.LivepeerNode.LiveRunnerAddr
+		return &v
+	}
+	return h.RunnerHost.ServiceURI()
+}
+
+func parseLiveRunnerAddr(addr string) (*url.URL, error) {
+	parsed, err := url.ParseRequestURI(addr)
+	if err != nil {
+		return nil, err
+	}
+	if !parsed.IsAbs() || parsed.Host == "" {
+		return nil, fmt.Errorf("must be an absolute URL")
+	}
+	if parsed.Scheme != "http" && parsed.Scheme != "https" {
+		return nil, fmt.Errorf("scheme must be http or https")
+	}
+	return parsed, nil
+}