ai/live: Remote signer discovery (#3866)

This adds an optional remote discovery feature to the remote signer node. Client using the remote signer can now discover orchestrators and their supported capabilities on the network. The remote discovery endpoint is compatible with the existing orchestrator webhook, making it usable by go-livepeer as well as local gateway SDKs.

### Usage

* Remote signer node gets a -remoteDiscovery flag. This is only usable in remote signer mode for now and is opt-in. All the other gateway-side orchestrator configuration flags are supported here: -orchWebhookUrl, -orchAddr , -orchBlocklist, -extraNodes etc. Additionally, operators can tune the orchestrator refresh interval with the existing -liveAICapReportInterval flag.

* If the gateway is using a remote signer (via -remoteSignerUrl) and has no other orchestrator sources (eg, webhook, orchestrator list) then it will fall back to using the remote signer's discovery endpoint.

* The discovery endpoint is GET /discover-orchestrators. It returns a JSON list of [{"address":...,"capabilities":[...]}] where the address is the URI the orchestrator can be reached at, and the capabilities is a list of capabilities the orchestrator supports, eg live-video-to-video/streamdiffusion. There is also an optional caps query parameter to return only the subset of orchestrators matching the capability (exact string match). Multiple caps parameters can be included (OR).

### Remote Discovery Implementation

* Remote discovery builds on the existing DB discovery pool's orchestrator polling. The DB discovery pool periodically (via -liveAICapReportInterval) fetches orchestrator info from the network and updates the node's capability cache.

* The remote discovery pool lazily reads from this cache and builds a capability-indexed view for quick lookups.

* Price filtering happens during refresh, so it doesn't return orchestrators that will be later rejected by the remote signer due to price

### Supporting Changes

* DB Discovery pool and Webhook Discovery pool are updated to use a "builder pattern" of config structs + initializer, matching the regular orchestrator pool. The existing constructor interfaces are left untouched, but the plumbing underneath uses the new builder pattern. This is done to support the IgnoreCapacityCheck flag (see below)

* To avoid returning capacity errors during the DB Discovery refresh call (it is capabilities we want), we need to thread through the IgnoreCapacityCheck flag a couple places from the webhook, DB discovery, etc. Note that this flag is only in enabled when the node is in remote signer mode, so this change should not affect other modes such as the gateway.

* DB Discovery cache refreshes now incorporate the ExtraNodes field in order to fully traverse the orchestrators available on the network.

* To support price filtering during remote discovery, the PriceInfo field is added to OrchNetworkCapabilities.

* The LiveAICapReportInterval field was added to the LivepeerNode; this is just storing an existing flag so the remote signer node can use it.

* Bug fix in the broadcast price config to mitigate cross-test contamination due to the use of globals + inconsistent cleanup from other tests

* Tests, docs, etc.

Author: j0sh

cmd/livepeer/starter/flags.go

@@ -140,6 +140,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
 	cfg.TestOrchAvail = fs.Bool("startupAvailabilityCheck", *cfg.TestOrchAvail, "Set to false to disable the startup Orchestrator availability check on the configured serviceAddr")
 	cfg.RemoteSigner = fs.Bool("remoteSigner", *cfg.RemoteSigner, "Set to true to run remote signer service")
 	cfg.RemoteSignerUrl = fs.String("remoteSignerUrl", *cfg.RemoteSignerUrl, "URL of remote signer service to use (e.g., http://localhost:8935). Gateway only.")
+	cfg.RemoteDiscovery = fs.Bool("remoteDiscovery", *cfg.RemoteDiscovery, "Enable orchestrator discovery on remote signers")
 
 	// Gateway metrics
 	cfg.KafkaBootstrapServers = fs.String("kafkaBootstrapServers", *cfg.KafkaBootstrapServers, "URL of Kafka Bootstrap Servers")

cmd/livepeer/starter/starter.go

@@ -169,6 +169,7 @@ type LivepeerConfig struct {
 	TestOrchAvail              *bool
 	RemoteSigner               *bool
 	RemoteSignerUrl            *string
+	RemoteDiscovery            *bool
 	AIRunnerImage              *string
 	AIRunnerImageOverrides     *string
 	AIVerboseLogs              *bool
@@ -306,6 +307,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 	defaultTestOrchAvail := true
 	defaultRemoteSigner := false
 	defaultRemoteSignerUrl := ""
+	defaultRemoteDiscovery := false
 
 	// Gateway logs
 	defaultKafkaBootstrapServers := ""
@@ -429,6 +431,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 		TestOrchAvail:   &defaultTestOrchAvail,
 		RemoteSigner:    &defaultRemoteSigner,
 		RemoteSignerUrl: &defaultRemoteSignerUrl,
+		RemoteDiscovery: &defaultRemoteDiscovery,
 
 		// Gateway logs
 		KafkaBootstrapServers: &defaultKafkaBootstrapServers,
@@ -1635,6 +1638,10 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 			n.OrchestratorPool = discovery.NewWebhookPool(bcast, whurl, *cfg.DiscoveryTimeout)
 		} else if len(orchURLs) > 0 {
 			n.OrchestratorPool = discovery.NewOrchestratorPool(bcast, orchURLs, common.Score_Trusted, orchBlacklist, *cfg.DiscoveryTimeout)
+		} else if n.RemoteSignerUrl != nil {
+			orchDiscoveryURL := n.RemoteSignerUrl.ResolveReference(&url.URL{Path: "/discover-orchestrators"})
+			glog.Info("Using remote signer orchestrator discovery endpoint ", orchDiscoveryURL)
+			n.OrchestratorPool = discovery.NewWebhookPool(bcast, orchDiscoveryURL, *cfg.DiscoveryTimeout)
 		}
 
 		// When the node is on-chain mode always cache the on-chain orchestrators and poll for updates
@@ -1648,7 +1655,7 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 				exit("Could not create orchestrator pool with DB cache: %v", err)
 			}
 
-			//if orchURLs is empty and webhook pool not used, use the DB orchestrator pool cache as orchestrator pool
+			// If orchURLs is empty and webhook pool not used, use the DB orchestrator pool cache.
 			if *cfg.OrchWebhookURL == "" && len(orchURLs) == 0 {
 				n.OrchestratorPool = dbOrchPoolCache
 			}
@@ -1793,6 +1800,12 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	if cfg.LiveAIHeartbeatInterval != nil {
 		n.LiveAIHeartbeatInterval = *cfg.LiveAIHeartbeatInterval
 	}
+	if cfg.LiveAICapReportInterval != nil {
+		n.LiveAICapReportInterval = *cfg.LiveAICapReportInterval
+	}
+	if cfg.RemoteDiscovery != nil {
+		n.RemoteDiscovery = *cfg.RemoteDiscovery
+	}
 	if cfg.LiveAIHeartbeatHeaders != nil {
 		n.LiveAIHeartbeatHeaders = make(map[string]string)
 		headers := strings.Split(*cfg.LiveAIHeartbeatHeaders, ",")
@@ -1842,8 +1855,69 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 		}()
 	}
 
-	// Start remote signer server if in remote signer mode
+	// Set up orchestrator pool for remote signer mode and start server
 	if n.NodeType == core.RemoteSignerNode {
+		bcast := core.NewBroadcaster(n)
+		orchBlacklist := parseOrchBlacklist(cfg.OrchBlacklist)
+		n.ExtraNodes = *cfg.ExtraNodes
+
+		// Set up orchestrator discovery - same logic as BroadcasterNode
+		if *cfg.OrchWebhookURL != "" {
+			whurl, err := validateURL(*cfg.OrchWebhookURL)
+			if err != nil {
+				glog.Exit("Error setting orch webhook URL ", err)
+			}
+			glog.Info("Using orchestrator webhook URL ", whurl)
+			n.OrchestratorPool = discovery.WebhookPoolConfig{
+				Broadcaster:         bcast,
+				Callback:            whurl,
+				DiscoveryTimeout:    *cfg.DiscoveryTimeout,
+				IgnoreCapacityCheck: true,
+			}.New()
+		} else if len(orchURLs) > 0 {
+			pool, err := discovery.NewOrchestratorPoolWithConfig(discovery.OrchestratorPoolConfig{
+				Broadcaster:         bcast,
+				URIs:                orchURLs,
+				Score:               common.Score_Trusted,
+				OrchBlacklist:       orchBlacklist,
+				DiscoveryTimeout:    *cfg.DiscoveryTimeout,
+				IgnoreCapacityCheck: true,
+				ExtraNodes:          *cfg.ExtraNodes,
+			})
+			if err != nil {
+				glog.Exit("Error initializing orchestrator pool ", err)
+			}
+			n.OrchestratorPool = pool
+		}
+
+		// When the node is on-chain mode always cache the on-chain orchestrators and poll for updates
+		if *cfg.Network != "offchain" && *cfg.RemoteDiscovery {
+			ctx, cancel := context.WithCancel(ctx)
+			defer cancel()
+			dbOrchPoolCache, err := discovery.DBOrchestratorPoolCacheConfig{
+				Ctx:                     ctx,
+				Node:                    n,
+				RoundsManager:           timeWatcher,
+				OrchBlacklist:           orchBlacklist,
+				DiscoveryTimeout:        *cfg.DiscoveryTimeout,
+				LiveAICapReportInterval: *cfg.LiveAICapReportInterval,
+				IgnoreCapacityCheck:     true,
+			}.New()
+			if err != nil {
+				exit("Could not create orchestrator pool with DB cache: %v", err)
+			}
+
+			// If orchURLs is empty and webhook pool not used, use the DB orchestrator pool cache
+			if *cfg.OrchWebhookURL == "" && len(orchURLs) == 0 {
+				n.OrchestratorPool = dbOrchPoolCache
+			}
+		}
+
+		if n.RemoteDiscovery && n.OrchestratorPool == nil {
+			exit("RemoteDiscovery is set but no orchestrator pool could be configured")
+		}
+
+		// Start remote signer server
 		go func() {
 			*cfg.HttpAddr = defaultAddr(*cfg.HttpAddr, "127.0.0.1", OrchestratorRpcPort)
 			glog.Info("Starting remote signer server on ", *cfg.HttpAddr)

common/types.go

@@ -174,6 +174,7 @@ type OrchNetworkCapabilities struct {
 	LocalAddress       string                     `json:"local_address"`
 	OrchURI            string                     `json:"orch_uri"`
 	Capabilities       *net.Capabilities          `json:"capabilities"`
+	PriceInfo          *net.PriceInfo             `json:"price_info"`
 	CapabilitiesPrices []*net.PriceInfo           `json:"capabilities_prices"`
 	Hardware           []*net.HardwareInformation `json:"hardware"`
 }

core/livepeernode.go

@@ -152,6 +152,7 @@ type LivepeerNode struct {
 	RemoteSignerUrl *url.URL
 	RemoteEthAddr   ethcommon.Address // eth address of the remote signer
 	InfoSig         []byte            // sig over eth address for the OrchestratorInfo request
+	RemoteDiscovery bool              // expose remote discovery endpoint when enabled
 
 	// Thread safety for config fields
 	mu                  sync.RWMutex
@@ -177,6 +178,7 @@ type LivepeerNode struct {
 	LiveAIHeartbeatURL         string
 	LiveAIHeartbeatHeaders     map[string]string
 	LiveAIHeartbeatInterval    time.Duration
+	LiveAICapReportInterval    time.Duration
 	LivePaymentInterval        time.Duration
 	LiveOutSegmentTimeout      time.Duration
 	LiveAISaveNSegments        *int

discovery/db_discovery.go

@@ -34,10 +34,33 @@ type DBOrchestratorPoolCache struct {
 	bcast                 common.Broadcaster
 	orchBlacklist         []string
 	discoveryTimeout      time.Duration
+	ignoreCapacityCheck   bool
 	node                  *core.LivepeerNode
 }
 
 func NewDBOrchestratorPoolCache(ctx context.Context, node *core.LivepeerNode, rm common.RoundsManager, orchBlacklist []string, discoveryTimeout time.Duration, liveAICapReportInterval time.Duration) (*DBOrchestratorPoolCache, error) {
+	return DBOrchestratorPoolCacheConfig{
+		Ctx:                     ctx,
+		Node:                    node,
+		RoundsManager:           rm,
+		OrchBlacklist:           orchBlacklist,
+		DiscoveryTimeout:        discoveryTimeout,
+		LiveAICapReportInterval: liveAICapReportInterval,
+	}.New()
+}
+
+type DBOrchestratorPoolCacheConfig struct {
+	Ctx                     context.Context
+	Node                    *core.LivepeerNode
+	RoundsManager           common.RoundsManager
+	OrchBlacklist           []string
+	DiscoveryTimeout        time.Duration
+	LiveAICapReportInterval time.Duration
+	IgnoreCapacityCheck     bool
+}
+
+func (cfg DBOrchestratorPoolCacheConfig) New() (*DBOrchestratorPoolCache, error) {
+	node := cfg.Node
 	if node.Eth == nil {
 		return nil, fmt.Errorf("could not create DBOrchestratorPoolCache: LivepeerEthClient is nil")
 	}
@@ -46,10 +69,11 @@ func NewDBOrchestratorPoolCache(ctx context.Context, node *core.LivepeerNode, rm
 		store:                 node.Database,
 		lpEth:                 node.Eth,
 		ticketParamsValidator: node.Sender,
-		rm:                    rm,
+		rm:                    cfg.RoundsManager,
 		bcast:                 core.NewBroadcaster(node),
-		orchBlacklist:         orchBlacklist,
-		discoveryTimeout:      discoveryTimeout,
+		orchBlacklist:         cfg.OrchBlacklist,
+		discoveryTimeout:      cfg.DiscoveryTimeout,
+		ignoreCapacityCheck:   cfg.IgnoreCapacityCheck,
 		node:                  node,
 	}
 
@@ -62,7 +86,7 @@ func NewDBOrchestratorPoolCache(ctx context.Context, node *core.LivepeerNode, rm
 			return err
 		}
 
-		if err := dbo.pollOrchestratorInfo(ctx, liveAICapReportInterval); err != nil {
+		if err := dbo.pollOrchestratorInfo(cfg.Ctx, cfg.LiveAICapReportInterval); err != nil {
 			return err
 		}
 		return nil
@@ -152,7 +176,19 @@ func (dbo *DBOrchestratorPoolCache) GetOrchestrators(ctx context.Context, numOrc
 		return true
 	}
 
-	orchPool := NewOrchestratorPoolWithPred(dbo.bcast, uris, pred, common.Score_Untrusted, dbo.orchBlacklist, dbo.discoveryTimeout)
+	orchPool, err := NewOrchestratorPoolWithConfig(OrchestratorPoolConfig{
+		Broadcaster:         dbo.bcast,
+		URIs:                uris,
+		Pred:                pred,
+		Score:               common.Score_Untrusted,
+		OrchBlacklist:       dbo.orchBlacklist,
+		DiscoveryTimeout:    dbo.discoveryTimeout,
+		IgnoreCapacityCheck: dbo.ignoreCapacityCheck,
+		ExtraNodes:          dbo.bcast.ExtraNodes(),
+	})
+	if err != nil {
+		return nil, err
+	}
 	orchInfos, err := orchPool.GetOrchestrators(ctx, numOrchestrators, suspender, caps, scorePred)
 	if err != nil || len(orchInfos) <= 0 {
 		return nil, err
@@ -303,16 +339,24 @@ func (dbo *DBOrchestratorPoolCache) cacheOrchInfos() error {
 	}
 
 	type orchPollingInfo struct {
+		level    int
 		orchInfo *net.OrchestratorInfo
 		dbOrch   *common.DBOrch
 	}
 
-	resc, errc := make(chan orchPollingInfo, len(orchs)), make(chan error, len(orchs))
+	nodesPerOrch := dbo.bcast.ExtraNodes()
+	// Each base orchestrator can contribute itself plus up to nodesPerOrch first-level advertised nodes.
+	maxOrchs := len(orchs) * (nodesPerOrch + 1)
+	resc, errc := make(chan orchPollingInfo, maxOrchs), make(chan error, maxOrchs)
 	timeout := getOrchestratorTimeoutLoop // Needs to be same or longer than GRPCConnectTimeout in server/rpc.go
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
+	getOrchInfoRPC := serverGetOrchInfo
+	if pool, ok := dbo.node.OrchestratorPool.(*orchestratorPool); ok && pool.getOrchInfo != nil {
+		getOrchInfoRPC = pool.getOrchInfo
+	}
 
-	getOrchInfo := func(orch common.OrchestratorLocalInfo) {
+	getOrchInfo := func(orch common.OrchestratorLocalInfo, level int) {
 		uri, err := parseURI(orch.URL.String())
 		if err != nil {
 			errc <- err
@@ -323,7 +367,9 @@ func (dbo *DBOrchestratorPoolCache) cacheOrchInfos() error {
 			errc <- fmt.Errorf("skipping orch=%v, URI not set", orch.URL.String())
 			return
 		}
-		info, err := serverGetOrchInfo(ctx, dbo.bcast, uri, server.GetOrchestratorInfoParams{})
+		info, err := getOrchInfoRPC(ctx, dbo.bcast, uri, server.GetOrchestratorInfoParams{
+			IgnoreCapacityCheck: dbo.ignoreCapacityCheck,
+		})
 		if err != nil {
 			errc <- err
 			return
@@ -361,13 +407,30 @@ func (dbo *DBOrchestratorPoolCache) cacheOrchInfos() error {
 			}
 		}
 
-		resc <- orchPollingInfo{orchInfo: info, dbOrch: dbOrch}
+		resc <- orchPollingInfo{
+			level:    level,
+			orchInfo: info,
+			dbOrch:   dbOrch,
+		}
 	}
 
+	seen := make(map[string]bool, maxOrchs)
 	numOrchs := 0
-	for _, orch := range orchs {
+	startOrchLookup := func(orch common.OrchestratorLocalInfo, level int) {
+		if orch.URL == nil {
+			return
+		}
+		key := orch.URL.String()
+		if key == "" || seen[key] {
+			return
+		}
+		seen[key] = true
 		numOrchs++
-		go getOrchInfo(orch)
+		go getOrchInfo(orch, level)
+	}
+
+	for _, orch := range orchs {
+		startOrchLookup(orch, 0)
 	}
 
 	var orchNetworkCapabilities []*common.OrchNetworkCapabilities
@@ -376,6 +439,22 @@ func (dbo *DBOrchestratorPoolCache) cacheOrchInfos() error {
 		case res := <-resc:
 			//add response to network capabilities
 			orchNetworkCapabilities = append(orchNetworkCapabilities, orchInfoToOrchNetworkCapabilities(res.orchInfo))
+
+			// discover newly advertised nodes. only recurse the first level.
+			if res.level == 0 && len(res.orchInfo.GetNodes()) > 0 {
+				for idx, inst := range res.orchInfo.GetNodes() {
+					if idx >= nodesPerOrch {
+						break
+					}
+					u, err := parseURI(inst)
+					if err != nil {
+						glog.Errorf("Invalid node URL orch=%v node=%v err=%q", res.orchInfo.GetTranscoder(), inst, err)
+						continue
+					}
+					startOrchLookup(common.OrchestratorLocalInfo{URL: u, Score: common.Score_Untrusted}, res.level+1)
+				}
+			}
+
 			//update db with response
 			if res.dbOrch != nil {
 				if err := dbo.store.UpdateOrch(res.dbOrch); err != nil {
@@ -507,6 +586,7 @@ func orchInfoToOrchNetworkCapabilities(info *net.OrchestratorInfo) *common.OrchN
 		orch.LocalAddress = ethcommon.BytesToAddress(info.GetAddress()).Hex()
 		orch.OrchURI = info.GetTranscoder()
 		orch.Capabilities = info.GetCapabilities()
+		orch.PriceInfo = info.GetPriceInfo()
 		orch.Hardware = info.GetHardware()
 		orch.CapabilitiesPrices = info.GetCapabilitiesPrices()
 		if info.GetTicketParams() != nil {