From a45206f03f1ab9003530f80978795f8c1e7b70f5 Mon Sep 17 00:00:00 2001 From: Mike Zupper Date: Mon, 2 Mar 2026 10:13:53 -0500 Subject: [PATCH 1/2] feat: initial implementation of options filtering for byoc jobs expose per-capability worker options in getNetworkCapabilities Worker options (model, vram, etc.) registered by BYOC workers are now returned under each orchestrator in /getNetworkCapabilities as capability_options keyed by capability name, alongside capabilities_prices. Changes: - Add GetAllWorkerOptionsByCapability() to ExternalCapabilities - Change GET /process/options on the orchestrator to return a per-capability map instead of a flat list - Add FetchCapabilityOptions() helper for single-orch HTTP fetch - Add CapabilityOptions field to OrchNetworkCapabilities - getNetworkCapabilitiesHandler fans out to each orch's OrchURI and attaches options via struct copies (not mutations of the cached pointers) - Fix race condition where direct mutation of cached OrchNetworkCapabilities pointers corrupted fields like capabilities_prices - added documentation for the BYOC implementation --- byoc/byoc.go | 4 + byoc/job_gateway.go | 131 +++++- byoc/job_gateway_test.go | 44 ++ byoc/job_orchestrator.go | 132 +++++- byoc/job_orchestrator_test.go | 71 +++ byoc/stream_orchestrator.go | 16 +- byoc/types.go | 16 +- common/types.go | 1 + core/ai_orchestrator.go | 42 +- core/external_capabilities.go | 283 +++++++++++- core/external_capabilities_test.go | 4 +- core/options_filter.go | 136 ++++++ core/options_filter_test.go | 39 ++ core/orch_test.go | 8 +- doc/byoc-job-filtering-architecture.md | 585 +++++++++++++++++++++++++ doc/byoc-technical-details.md | 437 ++++++++++++++++++ server/handlers.go | 39 +- 17 files changed, 1898 insertions(+), 90 deletions(-) create mode 100644 core/options_filter.go create mode 100644 core/options_filter_test.go create mode 100644 doc/byoc-job-filtering-architecture.md create mode 100644 doc/byoc-technical-details.md diff --git a/byoc/byoc.go b/byoc/byoc.go index 8eb966604f..52b56f50f3 100644 --- a/byoc/byoc.go +++ b/byoc/byoc.go @@ -178,6 +178,9 @@ func (bsg *BYOCGatewayServer) registerRoutes() { //TODO: add WHEP support + // Worker options aggregation + bsg.httpMux.Handle("GET /process/options", bsg.GetWorkerOptions()) + // Job submission routes for batch processing bsg.httpMux.Handle("/process/request/", bsg.SubmitJob()) } @@ -226,6 +229,7 @@ func (bso *BYOCOrchestratorServer) registerRoutes() { // Job submission routes for batch processing bso.httpMux.Handle("/process/request/", bso.ProcessJob()) bso.httpMux.Handle("/process/token", bso.GetJobToken()) + bso.httpMux.Handle("GET /process/options", bso.GetWorkerOptions()) bso.httpMux.Handle("/capability/register", bso.RegisterCapability()) bso.httpMux.Handle("/capability/unregister", bso.UnregisterCapability()) // Stream routes diff --git a/byoc/job_gateway.go b/byoc/job_gateway.go index f217442008..24f7b572e2 100644 --- a/byoc/job_gateway.go +++ b/byoc/job_gateway.go @@ -357,6 +357,17 @@ func getJobOrchestrators(ctx context.Context, node *core.LivepeerNode, capabilit tokenReq.Header.Set(jobEthAddressHdr, base64.StdEncoding.EncodeToString(reqSenderStr)) tokenReq.Header.Set(jobCapabilityHdr, capability) + // Pass the options filter so the orchestrator can return capacity that + // reflects only runners matching the filter (avoids wasted round-trips). + if len(params.OptionsFilter) > 0 { + filterJSON, err := json.Marshal(params.OptionsFilter) + if err == nil { + q := tokenReq.URL.Query() + q.Set("options_filter", string(filterJSON)) + tokenReq.URL.RawQuery = q.Encode() + } + } + resp, err := sendJobReqWithTimeout(tokenReq, respTimeout) if err != nil { clog.Errorf(ctx, "failed to get token from Orchestrator err=%v", err) @@ -419,17 +430,34 @@ func getJobOrchestrators(ctx context.Context, node *core.LivepeerNode, capabilit select { case token := <-tokenCh: if token.AvailableCapacity > 0 { - jobTokens = append(jobTokens, token) + if core.AnyOptionsMatch(params.OptionsFilter, token.WorkerOptions) { + if clog.V(common.VERBOSE) { + filterJSON, _ := json.Marshal(params.OptionsFilter) + optsJSON, _ := json.Marshal(token.WorkerOptions) + clog.V(common.VERBOSE).Infof(ctx, "job selection orch=%v accepted filter=%v all_options=%v", token.ServiceAddr, string(filterJSON), string(optsJSON)) + } + jobTokens = append(jobTokens, token) + } else { + if clog.V(common.VERBOSE) { + filterJSON, _ := json.Marshal(params.OptionsFilter) + optsJSON, _ := json.Marshal(token.WorkerOptions) + clog.V(common.VERBOSE).Infof(ctx, "job selection orch=%v rejected filter=%v worker_options=%v", token.ServiceAddr, string(filterJSON), string(optsJSON)) + } + } + } else { + clog.V(common.VERBOSE).Infof(ctx, "job selection orch=%v skipped no_capacity", token.ServiceAddr) } nbResp++ case <-errCh: nbResp++ case <-tokensCtx.Done(): //searchTimeout reached, return tokens received + clog.V(common.VERBOSE).Infof(ctx, "job selection timeout reached selected=%v", len(jobTokens)) return jobTokens, nil } } + clog.V(common.VERBOSE).Infof(ctx, "job selection selected=%v from=%v orchs", len(jobTokens), nbResp) // received enough tokens or all responses arrived return jobTokens, nil } @@ -458,6 +486,10 @@ func genOrchestratorReq(b common.Broadcaster) (*net.OrchestratorRequest, error) return &net.OrchestratorRequest{Address: b.Address().Bytes(), Sig: sig}, nil } +// getToken fetches a job token from a specific orchestrator URL with exponential +// backoff retry. It is used during stream reconnect / orchestrator failover where +// a brief retry is acceptable. For fan-out discovery use getOrchJobToken instead, +// which has no retry so that slow orchestrators don't stall the whole selection. func getToken(ctx context.Context, respTimeout time.Duration, orchUrl, capability, sender, senderSig string) (*JobToken, error) { start := time.Now() tokenReq, err := http.NewRequestWithContext(ctx, "GET", orchUrl+"/process/token", nil) @@ -515,3 +547,100 @@ func getToken(ctx context.Context, respTimeout time.Duration, orchUrl, capabilit } return nil, fmt.Errorf("failed to get token from Orchestrator after %d attempts", attempt) } + +// FetchWorkerOptions fans out GET /process/options to each orchestrator URL, +// merges the results, and returns the deduplicated union. timeout controls +// how long to wait for all responses. +// FetchCapabilityOptions calls GET /process/options on a single orchestrator URL +// and returns the per-capability options map. Returns nil on any error. +func FetchCapabilityOptions(ctx context.Context, orchURL string, timeout time.Duration) map[string][]map[string]interface{} { + reqCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, strings.TrimRight(orchURL, "/")+"/process/options", nil) + if err != nil { + clog.Errorf(ctx, "FetchCapabilityOptions orch=%v failed to create request err=%v", orchURL, err) + return nil + } + resp, err := httpClient.Do(req) + if err != nil { + clog.V(common.VERBOSE).Infof(ctx, "FetchCapabilityOptions orch=%v request failed err=%v", orchURL, err) + return nil + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + clog.V(common.VERBOSE).Infof(ctx, "FetchCapabilityOptions orch=%v non-200 status=%v", orchURL, resp.StatusCode) + return nil + } + var opts map[string][]map[string]interface{} + if err := json.NewDecoder(resp.Body).Decode(&opts); err != nil { + clog.Errorf(ctx, "FetchCapabilityOptions orch=%v failed to decode response err=%v", orchURL, err) + return nil + } + if clog.V(common.VERBOSE) { + optsJSON, _ := json.Marshal(opts) + clog.Infof(ctx, "FetchCapabilityOptions orch=%v received options=%v", orchURL, string(optsJSON)) + } + return opts +} + +// FetchWorkerOptions fans out GET /process/options to each orchestrator URL, +// merges the results, and returns the deduplicated union as a flat list. +// Used by the gateway's /process/options aggregator for model discovery. +func FetchWorkerOptions(ctx context.Context, orchs []common.OrchestratorLocalInfo, timeout time.Duration) []map[string]interface{} { + type orchResult struct { + capOpts map[string][]map[string]interface{} + } + resultCh := make(chan orchResult, len(orchs)) + + orchURLs := make([]string, len(orchs)) + for i, o := range orchs { + orchURLs[i] = o.URL.String() + } + clog.Infof(ctx, "FetchWorkerOptions querying num_orchs=%v urls=%v", len(orchs), orchURLs) + + reqCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + for _, orch := range orchs { + go func(orchURL string) { + resultCh <- orchResult{capOpts: FetchCapabilityOptions(reqCtx, orchURL, timeout)} + }(orch.URL.String()) + } + + // Collect and flatten all per-capability options; deduplicate by JSON fingerprint. + seen := make(map[string]struct{}) + all := make([]map[string]interface{}, 0) + for range orchs { + res := <-resultCh + for _, opts := range res.capOpts { + for _, opt := range opts { + key, _ := json.Marshal(opt) + if _, dup := seen[string(key)]; !dup { + seen[string(key)] = struct{}{} + all = append(all, opt) + } + } + } + } + clog.Infof(ctx, "FetchWorkerOptions total_unique=%v", len(all)) + return all +} + +// GetWorkerOptions fans out GET /process/options to every Orchestrator in the +// pool, merges the results, and returns the deduplicated union as a JSON array. +// This is the endpoint called by the gateway-proxy's /v1/models handler. +func (bsg *BYOCGatewayServer) GetWorkerOptions() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + orchs := bsg.node.OrchestratorPool.GetInfos() + all := FetchWorkerOptions(r.Context(), orchs, 2*time.Second) + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(all) + }) +} diff --git a/byoc/job_gateway_test.go b/byoc/job_gateway_test.go index dffa73a279..18f3e7ff36 100644 --- a/byoc/job_gateway_test.go +++ b/byoc/job_gateway_test.go @@ -370,6 +370,50 @@ func TestSubmitJob_OrchestratorSelectionParams(t *testing.T) { } +func TestGetJobOrchestrators_OptionsFilter(t *testing.T) { + newTokenServer := func(options []map[string]interface{}) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + token := createMockJobToken("http://" + r.Host) + token.WorkerOptions = options + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(token) + })) + } + + // s1: has llama-3 with 24GB — should pass filter + s1 := newTokenServer([]map[string]interface{}{{"model": "llama-3", "vram_gb": float64(24)}, {"model": "mistral-7b", "vram_gb": float64(24)}}) + // s2: only mistral-7b — should be rejected + s2 := newTokenServer([]map[string]interface{}{{"model": "mistral-7b", "vram_gb": float64(24)}}) + // s3: llama-3 but only 8GB — should be rejected + s3 := newTokenServer([]map[string]interface{}{{"model": "llama-3", "vram_gb": float64(8)}}) + defer s1.Close() + defer s2.Close() + defer s3.Close() + + node := mockJobLivepeerNode() + node.OrchestratorPool = newStubOrchestratorPool(node, []string{s1.URL, s2.URL, s3.URL}) + + params := JobParameters{ + OptionsFilter: map[string]string{ + "model": "llama-3", + "vram_gb": ">=16", + }, + } + + tokens, err := getJobOrchestrators( + context.Background(), + node, + "test-capability", + params, + 300*time.Millisecond, + 200*time.Millisecond, + ) + assert.NoError(t, err) + assert.Len(t, tokens, 1) + assert.Equal(t, "llama-3", tokens[0].WorkerOptions[0]["model"]) + assert.Equal(t, float64(24), tokens[0].WorkerOptions[0]["vram_gb"]) +} + func TestSetupGatewayJob(t *testing.T) { // Prepare a JobRequest with valid fields jobDetails := JobRequestDetails{StreamId: "test-stream"} diff --git a/byoc/job_orchestrator.go b/byoc/job_orchestrator.go index 210474992e..7456c84726 100644 --- a/byoc/job_orchestrator.go +++ b/byoc/job_orchestrator.go @@ -65,6 +65,7 @@ func (bs *BYOCOrchestratorServer) RegisterCapability() http.Handler { w.WriteHeader(http.StatusOK) w.Write([]byte("ok")) + cap.SetWorkerOptions(cap.WorkerOptions) clog.Infof(context.TODO(), "registered capability remoteAddr=%v capability=%v url=%v price=%v auth_token=%v", remoteAddr, cap.Name, cap.Url, big.NewRat(cap.PricePerUnit, cap.PriceScaling), cap.AuthToken != "") }) } @@ -87,26 +88,57 @@ func (bs *BYOCOrchestratorServer) UnregisterCapability() http.Handler { return } defer r.Body.Close() - extCapName := string(body) remoteAddr := getRemoteAddr(r) - err = orch.RemoveExternalCapability(extCapName) - if err != nil { - clog.Errorf(context.TODO(), "Error removing capability: %v", err) - http.Error(w, fmt.Sprintf("Error removing capability: %v", err), http.StatusBadRequest) - return + // Try JSON {name, url} format first; fall back to plain capability name string. + var unregReq struct { + Name string `json:"name"` + Url string `json:"url"` + } + capName := string(body) + var removeErr error + if jsonErr := json.Unmarshal(body, &unregReq); jsonErr == nil && unregReq.Name != "" { + capName = unregReq.Name + if unregReq.Url != "" { + bs.node.ExternalCapabilities.RemoveCapabilityRunner(unregReq.Name, unregReq.Url) + } else { + removeErr = orch.RemoveExternalCapability(capName) + } + } else { + removeErr = orch.RemoveExternalCapability(capName) } - - w.Header().Set("Content-Type", "application/json") - if err != nil { - w.WriteHeader(http.StatusBadRequest) - w.Write([]byte(fmt.Sprintf("Error removing capability: %v", err))) + if removeErr != nil { + clog.Errorf(context.TODO(), "Error removing capability: %v", removeErr) + http.Error(w, fmt.Sprintf("Error removing capability: %v", removeErr), http.StatusBadRequest) return } w.WriteHeader(http.StatusOK) w.Write([]byte("ok")) - clog.Infof(context.TODO(), "removed capability remoteAddr=%v capability=%v", remoteAddr, extCapName) + clog.Infof(context.TODO(), "removed capability remoteAddr=%v capability=%v", remoteAddr, capName) + }) +} + +// GetWorkerOptions returns the cached WorkerOptions for all registered +// capabilities on this Orchestrator. Called by the Gateway's /process/options aggregator. +func (bso *BYOCOrchestratorServer) GetWorkerOptions() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + options := map[string][]map[string]interface{}{} + if bso.node != nil && bso.node.ExternalCapabilities != nil { + if all := bso.node.ExternalCapabilities.GetAllWorkerOptionsByCapability(); len(all) > 0 { + options = all + } + } + optsJSON, _ := json.Marshal(options) + clog.Infof(r.Context(), "GetWorkerOptions remoteAddr=%v num_capabilities=%v options=%v", r.RemoteAddr, len(options), string(optsJSON)) + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(options) }) } @@ -146,10 +178,24 @@ func (bso *BYOCOrchestratorServer) GetJobToken() http.Handler { return } + // Read optional options filter from query param. When present, the + // returned AvailableCapacity will reflect only runners matching the + // filter, avoiding wasted round-trips to orchestrators whose capacity + // is held by runners that don't match the gateway's requirements. + var optionsFilter map[string]string + if filterStr := r.URL.Query().Get("options_filter"); filterStr != "" { + _ = json.Unmarshal([]byte(filterStr), &optionsFilter) + } + w.Header().Set("Content-Type", "application/json") jobToken := JobToken{SenderAddress: nil, TicketParams: nil, Balance: 0, Price: nil} - capacity := orch.CheckExternalCapabilityCapacity(jobCapsHdr) + var capacity int64 + if bso.node != nil && bso.node.ExternalCapabilities != nil { + capacity = bso.node.ExternalCapabilities.GetFilteredCapacity(jobCapsHdr, optionsFilter) + } else { + capacity = orch.CheckExternalCapabilityCapacity(jobCapsHdr) + } senderAddr := ethcommon.HexToAddress(jobSenderAddr.Addr) @@ -190,6 +236,11 @@ func (bso *BYOCOrchestratorServer) GetJobToken() http.Handler { capBalInt = capBalInt / 1000 } + var workerOptions []map[string]interface{} + if bso.node != nil && bso.node.ExternalCapabilities != nil { + workerOptions = bso.node.ExternalCapabilities.GetCapabilityWorkerOptions(jobCapsHdr) + } + jobToken = JobToken{ SenderAddress: jobSenderAddr, TicketParams: ticketParams, @@ -197,6 +248,7 @@ func (bso *BYOCOrchestratorServer) GetJobToken() http.Handler { Price: jobPrice, ServiceAddr: orch.ServiceURI().String(), AvailableCapacity: capacity, + WorkerOptions: workerOptions, } //send response indicating compatible @@ -274,8 +326,8 @@ func (bso *BYOCOrchestratorServer) processJob(ctx context.Context, w http.Respon req.Header.Add("Content-Type", r.Header.Get("Content-Type")) // Add Authorization header if auth token is set for this capability - if extCap, ok := bso.node.ExternalCapabilities.Capabilities[orchJob.Req.Capability]; ok { - if extCap.AuthToken != "" { + if nameMap, ok := bso.node.ExternalCapabilities.Capabilities[orchJob.Req.Capability]; ok { + if extCap, ok := nameMap[orchJob.Req.CapabilityUrl]; ok && extCap.AuthToken != "" { req.Header.Add("Authorization", "Bearer "+extCap.AuthToken) } } @@ -571,11 +623,53 @@ func (bso *BYOCOrchestratorServer) verifyJobCreds(ctx context.Context, jobCreds return nil, errSegSig } - if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil { - return nil, errZeroCapacity + // Use the node's ExternalCapabilities runner registry only when runners are + // actually registered for this capability; otherwise fall back to the orch + // interface (used by tests and legacy deployments). + var hasRunners bool + if bso.node != nil && bso.node.ExternalCapabilities != nil { + _, hasRunners = bso.node.ExternalCapabilities.GetCapability(jobData.Capability) + } + if hasRunners { + // Extract options filter from job parameters so runner selection respects + // the same constraint the gateway used to pick this orchestrator. + var jobParams JobParameters + _ = json.Unmarshal([]byte(jobData.Parameters), &jobParams) + filter := jobParams.OptionsFilter + + // Atomically select and (optionally) reserve the best matching runner, + // ensuring Reserve and GetUrl always refer to the same runner. + if reserveCapacity { + runner, err := bso.node.ExternalCapabilities.SelectAndReserveRunner(jobData.Capability, filter) + if err != nil { + return nil, errZeroCapacity + } + jobData.CapabilityUrl = runner.Url + if clog.V(common.VERBOSE) { + filterJSON, _ := json.Marshal(filter) + optsJSON, _ := json.Marshal(runner.WorkerOptions) + clog.V(common.VERBOSE).Infof(ctx, "orch runner selected capability=%v url=%v load=%v capacity=%v filter=%v worker_options=%v", + jobData.Capability, runner.Url, runner.Load, runner.Capacity, string(filterJSON), string(optsJSON)) + } + } else { + runner := bso.node.ExternalCapabilities.SelectRunner(jobData.Capability, filter) + if runner != nil { + jobData.CapabilityUrl = runner.Url + if clog.V(common.VERBOSE) { + filterJSON, _ := json.Marshal(filter) + optsJSON, _ := json.Marshal(runner.WorkerOptions) + clog.V(common.VERBOSE).Infof(ctx, "orch runner selected (no reserve) capability=%v url=%v load=%v capacity=%v filter=%v worker_options=%v", + jobData.Capability, runner.Url, runner.Load, runner.Capacity, string(filterJSON), string(optsJSON)) + } + } + } + } else { + // Fallback to interface methods (e.g. in tests with mocked orchestrator) + if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil { + return nil, errZeroCapacity + } + jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability) } - - jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability) return jobData, nil } diff --git a/byoc/job_orchestrator_test.go b/byoc/job_orchestrator_test.go index 55253d9669..a327bb482e 100644 --- a/byoc/job_orchestrator_test.go +++ b/byoc/job_orchestrator_test.go @@ -768,6 +768,77 @@ func TestGetJobToken_Success(t *testing.T) { assert.Equal(t, int64(1000), token.Balance) } +func TestGetJobToken_IncludesWorkerOptions(t *testing.T) { + mockVerifySig := func(addr ethcommon.Address, msg string, sig []byte) bool { + return true + } + mockJobPriceInfo := func(addr ethcommon.Address, cap string) (*net.PriceInfo, error) { + return &net.PriceInfo{PricePerUnit: 10, PixelsPerUnit: 1}, nil + } + mockTicketParams := func(addr ethcommon.Address, price *net.PriceInfo) (*net.TicketParams, error) { + return &net.TicketParams{ + Recipient: ethcommon.HexToAddress("0x1111111111111111111111111111111111111111").Bytes(), + FaceValue: big.NewInt(1000).Bytes(), + WinProb: big.NewInt(1).Bytes(), + Seed: big.NewInt(1234).Bytes(), + ExpirationBlock: big.NewInt(100000).Bytes(), + }, nil + } + + mockJobOrch := newMockJobOrchestrator() + mockJobOrch.verifySignature = mockVerifySig + mockJobOrch.jobPriceInfo = mockJobPriceInfo + mockJobOrch.ticketParams = mockTicketParams + + node := mockJobLivepeerNode() + if node.ExternalCapabilities == nil { + node.ExternalCapabilities = core.NewExternalCapabilities() + } + node.ExternalCapabilities.Capabilities["test-cap"] = map[string]*core.ExternalCapability{ + "http://runner-a:8000": {Name: "test-cap", WorkerOptions: []map[string]interface{}{{"model": "llama-3", "vram_gb": 24.0}}}, + "http://runner-b:8000": {Name: "test-cap", WorkerOptions: []map[string]interface{}{{"model": "mistral-7b", "vram_gb": 16.0}}}, + } + mockJobOrch.node = node + + bso := &BYOCOrchestratorServer{ + node: node, + orch: mockJobOrch, + } + + gateway := newMockJobOrchestrator() + sig, _ := gateway.Sign([]byte(hexutil.Encode(gateway.Address().Bytes()))) + js := &JobSender{Addr: hexutil.Encode(gateway.Address().Bytes()), Sig: hexutil.Encode(sig)} + jsBytes, _ := json.Marshal(js) + jsBase64 := base64.StdEncoding.EncodeToString(jsBytes) + + req := httptest.NewRequest("GET", "/process/token", nil) + req.Header.Set(jobEthAddressHdr, jsBase64) + req.Header.Set(jobCapabilityHdr, "test-cap") + w := httptest.NewRecorder() + + handler := bso.GetJobToken() + handler.ServeHTTP(w, req) + + resp := w.Result() + assert.Equal(t, http.StatusOK, resp.StatusCode) + + var token JobToken + body, _ := io.ReadAll(resp.Body) + _ = json.Unmarshal(body, &token) + assert.Len(t, token.WorkerOptions, 2) + // Order is non-deterministic (Go map iteration); check by model name. + optsByModel := map[string]map[string]interface{}{} + for _, opt := range token.WorkerOptions { + if m, ok := opt["model"].(string); ok { + optsByModel[m] = opt + } + } + assert.Contains(t, optsByModel, "llama-3") + assert.Equal(t, float64(24), optsByModel["llama-3"]["vram_gb"]) + assert.Contains(t, optsByModel, "mistral-7b") + assert.Equal(t, float64(16), optsByModel["mistral-7b"]["vram_gb"]) +} + func TestProcessJob_MethodNotAllowed(t *testing.T) { bso := &BYOCOrchestratorServer{ node: mockJobLivepeerNode(), diff --git a/byoc/stream_orchestrator.go b/byoc/stream_orchestrator.go index 359e269a61..f823fbaa3e 100644 --- a/byoc/stream_orchestrator.go +++ b/byoc/stream_orchestrator.go @@ -149,7 +149,7 @@ func (bso *BYOCOrchestratorServer) StartStream() http.Handler { return } - req, err := bso.createWorkerReq(ctx, workerRoute, orchJob.Req.Capability, orchJob.Req.ID, bytes.NewBuffer(reqBodyBytes)) + req, err := bso.createWorkerReq(ctx, workerRoute, orchJob.Req.CapabilityUrl, orchJob.Req.Capability, orchJob.Req.ID, bytes.NewBuffer(reqBodyBytes)) if err != nil { clog.Errorf(ctx, "failed to create worker request err=%v", err) respondWithError(w, "Failed to create worker request", http.StatusInternalServerError) @@ -236,7 +236,7 @@ func (bso *BYOCOrchestratorServer) monitorOrchStream(job *orchJob) { return case <-pmtTicker.C: // Check payment status - extCap, ok := bso.node.ExternalCapabilities.Capabilities[capability] + extCap, ok := bso.node.ExternalCapabilities.GetCapability(capability) if !ok { clog.Errorf(ctx, "Capability not found for payment monitoring, exiting monitoring capability=%s", capability) return @@ -274,7 +274,7 @@ func (bso *BYOCOrchestratorServer) monitorOrchStream(job *orchJob) { // if not, send stop to worker and exit monitoring stream, exists := bso.node.ExternalCapabilities.GetStream(streamID) if !exists { - req, err := bso.createWorkerReq(ctx, job.Req.CapabilityUrl+"/stream/stop", job.Req.Capability, streamID, nil) + req, err := bso.createWorkerReq(ctx, job.Req.CapabilityUrl+"/stream/stop", job.Req.CapabilityUrl, job.Req.Capability, streamID, nil) if err != nil { clog.Errorf(ctx, "Error creating request to worker %v: %v", job.Req.CapabilityUrl, err) return @@ -327,7 +327,7 @@ func (bso *BYOCOrchestratorServer) StopStream() http.Handler { r.Body.Close() workerRoute := orchJob.Req.CapabilityUrl + "/stream/stop" - req, err := bso.createWorkerReq(ctx, workerRoute, orchJob.Req.Capability, jobDetails.StreamId, bytes.NewBuffer(body)) + req, err := bso.createWorkerReq(ctx, workerRoute, orchJob.Req.CapabilityUrl, orchJob.Req.Capability, jobDetails.StreamId, bytes.NewBuffer(body)) if err != nil { clog.Errorf(ctx, "failed to create /stream/stop request to worker err=%v", err) http.Error(w, err.Error(), http.StatusBadRequest) @@ -378,7 +378,7 @@ func (bso *BYOCOrchestratorServer) UpdateStream() http.Handler { r.Body.Close() workerRoute := orchJob.Req.CapabilityUrl + "/stream/params" - req, err := bso.createWorkerReq(ctx, workerRoute, orchJob.Req.Capability, jobDetails.StreamId, bytes.NewBuffer(body)) + req, err := bso.createWorkerReq(ctx, workerRoute, orchJob.Req.CapabilityUrl, orchJob.Req.Capability, jobDetails.StreamId, bytes.NewBuffer(body)) if err != nil { clog.Errorf(ctx, "failed to create /stream/params request to worker err=%v", err) http.Error(w, err.Error(), http.StatusBadRequest) @@ -402,7 +402,7 @@ func (bso *BYOCOrchestratorServer) UpdateStream() http.Handler { // createWorkerReq creates an HTTP request to send to the worker. // handles setting stream id and auth headers for worker -func (bso *BYOCOrchestratorServer) createWorkerReq(ctx context.Context, workerRoute, capability, streamId string, body io.Reader) (*http.Request, error) { +func (bso *BYOCOrchestratorServer) createWorkerReq(ctx context.Context, workerRoute, capabilityUrl, capability, streamId string, body io.Reader) (*http.Request, error) { req, err := http.NewRequestWithContext(ctx, "POST", workerRoute, body) if err != nil { return nil, err @@ -414,8 +414,8 @@ func (bso *BYOCOrchestratorServer) createWorkerReq(ctx context.Context, workerRo } // Add Authorization header if auth token is set for this capability - if extCap, ok := bso.node.ExternalCapabilities.Capabilities[capability]; ok { - if extCap.AuthToken != "" { + if nameMap, ok := bso.node.ExternalCapabilities.Capabilities[capability]; ok { + if extCap, ok := nameMap[capabilityUrl]; ok && extCap.AuthToken != "" { req.Header.Add("Authorization", "Bearer "+extCap.AuthToken) } } diff --git a/byoc/types.go b/byoc/types.go index fa9b4a5e8b..0d6b0469e8 100644 --- a/byoc/types.go +++ b/byoc/types.go @@ -118,7 +118,8 @@ type JobRequestDetails struct { type JobParameters struct { // Gateway - Orchestrators JobOrchestratorsFilter `json:"orchestrators,omitempty"` // list of orchestrators to use for the job + Orchestrators JobOrchestratorsFilter `json:"orchestrators,omitempty"` // list of orchestrators to use for the job + OptionsFilter map[string]string `json:"options_filter,omitempty"` // worker options capability filter // Orchestrator EnableVideoIngress bool `json:"enable_video_ingress,omitempty"` @@ -132,12 +133,13 @@ type JobOrchestratorsFilter struct { } type JobToken struct { - SenderAddress *JobSender `json:"sender_address,omitempty"` - TicketParams *net.TicketParams `json:"ticket_params,omitempty"` - Balance int64 `json:"balance,omitempty"` - Price *net.PriceInfo `json:"price,omitempty"` - ServiceAddr string `json:"service_addr,omitempty"` - AvailableCapacity int64 `json:"available_capacity,omitempty"` + SenderAddress *JobSender `json:"sender_address,omitempty"` + TicketParams *net.TicketParams `json:"ticket_params,omitempty"` + Balance int64 `json:"balance,omitempty"` + Price *net.PriceInfo `json:"price,omitempty"` + ServiceAddr string `json:"service_addr,omitempty"` + AvailableCapacity int64 `json:"available_capacity,omitempty"` + WorkerOptions []map[string]interface{} `json:"worker_options,omitempty"` LastNonce uint32 } diff --git a/common/types.go b/common/types.go index 1fef7d7886..2bb854cebe 100644 --- a/common/types.go +++ b/common/types.go @@ -177,4 +177,5 @@ type OrchNetworkCapabilities struct { PriceInfo *net.PriceInfo `json:"price_info"` CapabilitiesPrices []*net.PriceInfo `json:"capabilities_prices"` Hardware []*net.HardwareInformation `json:"hardware"` + CapabilityOptions map[string][]map[string]interface{} `json:"capability_options,omitempty"` } diff --git a/core/ai_orchestrator.go b/core/ai_orchestrator.go index 0155fe85da..a3e0b186b5 100644 --- a/core/ai_orchestrator.go +++ b/core/ai_orchestrator.go @@ -1139,51 +1139,23 @@ func (orch *orchestrator) RemoveExternalCapability(extCapability string) error { } func (orch *orchestrator) GetUrlForCapability(extCapability string) string { - for _, capability := range orch.node.ExternalCapabilities.Capabilities { - if capability.Name == extCapability { - return capability.Url - } + cap, ok := orch.node.ExternalCapabilities.GetCapability(extCapability) + if !ok || cap == nil { + return "" } - - return "" + return cap.Url } func (orch *orchestrator) CheckExternalCapabilityCapacity(extCapability string) int64 { - if cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability]; !ok { - return 0 - } else { - if cap.Load < cap.Capacity { - return int64(cap.Capacity - cap.Load) - } else { - return 0 - } - } + return orch.node.ExternalCapabilities.GetTotalCapacity(extCapability) } func (orch *orchestrator) ReserveExternalCapabilityCapacity(extCapability string) error { - cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability] - if ok { - cap.Mu.Lock() - defer cap.Mu.Unlock() - - cap.Load++ - return nil - } else { - return errors.New("external capability not found") - } + return orch.node.ExternalCapabilities.ReserveCapacity(extCapability) } func (orch *orchestrator) FreeExternalCapabilityCapacity(extCapability string) error { - cap, ok := orch.node.ExternalCapabilities.Capabilities[extCapability] - if ok { - cap.Mu.Lock() - defer cap.Mu.Unlock() - - cap.Load-- - return nil - } else { - return errors.New("external capability not found") - } + return orch.node.ExternalCapabilities.FreeCapacity(extCapability) } func (orch *orchestrator) JobPriceInfo(sender ethcommon.Address, jobCapability string) (*net.PriceInfo, error) { diff --git a/core/external_capabilities.go b/core/external_capabilities.go index 27ba79e45f..eb083b7430 100644 --- a/core/external_capabilities.go +++ b/core/external_capabilities.go @@ -23,7 +23,7 @@ type ExternalCapability struct { PriceScaling int64 `json:"price_scaling"` PriceCurrency string `json:"currency"` AuthToken string `json:"token"` - + WorkerOptions []map[string]interface{} `json:"worker_options,omitempty"` price *AutoConvertedPrice Mu sync.RWMutex @@ -104,13 +104,13 @@ func (sd *StreamInfo) cleanup() { type ExternalCapabilities struct { capm sync.Mutex - Capabilities map[string]*ExternalCapability + Capabilities map[string]map[string]*ExternalCapability // outer key = capability name, inner key = runner URL Streams map[string]*StreamInfo } func NewExternalCapabilities() *ExternalCapabilities { return &ExternalCapabilities{ - Capabilities: make(map[string]*ExternalCapability), + Capabilities: make(map[string]map[string]*ExternalCapability), Streams: make(map[string]*StreamInfo)} } @@ -196,11 +196,239 @@ func (extCaps *ExternalCapabilities) RemoveCapability(extCap string) { delete(extCaps.Capabilities, extCap) } +// RemoveCapabilityRunner removes a single runner URL from a capability. If it is +// the last runner for that capability, the capability entry is also removed. +func (extCaps *ExternalCapabilities) RemoveCapabilityRunner(name, url string) { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + + runners, ok := extCaps.Capabilities[name] + if !ok { + return + } + delete(runners, url) + if len(runners) == 0 { + delete(extCaps.Capabilities, name) + } +} + +// GetCapability returns any one runner for the given capability name. +func (extCaps *ExternalCapabilities) GetCapability(extCap string) (*ExternalCapability, bool) { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[extCap] + if !ok { + return nil, false + } + for _, cap := range runners { + return cap, true + } + return nil, false +} + +// GetCapabilityRunner returns the specific runner entry for a capability name + URL pair. +func (extCaps *ExternalCapabilities) GetCapabilityRunner(name, url string) (*ExternalCapability, bool) { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[name] + if !ok { + return nil, false + } + cap, ok := runners[url] + return cap, ok +} + +// GetTotalCapacity returns the sum of available capacity across all runners for +// the given capability name. It holds capm for the entire read. +func (extCaps *ExternalCapabilities) GetTotalCapacity(name string) int64 { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[name] + if !ok { + return 0 + } + var total int64 + for _, cap := range runners { + if cap.Load < cap.Capacity { + total += int64(cap.Capacity - cap.Load) + } + } + return total +} + +// GetFilteredCapacity returns the sum of available capacity across runners for +// the given capability name that also satisfy the options filter. +// If filter is empty, all runners are counted (equivalent to GetTotalCapacity). +func (extCaps *ExternalCapabilities) GetFilteredCapacity(name string, filter map[string]string) int64 { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[name] + if !ok { + return 0 + } + var total int64 + for _, cap := range runners { + if !AnyOptionsMatch(filter, cap.GetWorkerOptionsCopy()) { + continue + } + if cap.Load < cap.Capacity { + total += int64(cap.Capacity - cap.Load) + } + } + return total +} + +// ReserveCapacity atomically finds the first runner with available capacity and +// increments its Load. Returns an error if no runner has available capacity. +func (extCaps *ExternalCapabilities) ReserveCapacity(name string) error { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[name] + if !ok { + return fmt.Errorf("external capability not found: %s", name) + } + for _, cap := range runners { + cap.Mu.Lock() + if cap.Load < cap.Capacity { + cap.Load++ + cap.Mu.Unlock() + return nil + } + cap.Mu.Unlock() + } + return fmt.Errorf("no available capacity for capability: %s", name) +} + +// FreeCapacity decrements the Load of the first runner with non-zero Load. +func (extCaps *ExternalCapabilities) FreeCapacity(name string) error { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[name] + if !ok { + return fmt.Errorf("external capability not found: %s", name) + } + for _, cap := range runners { + cap.Mu.Lock() + if cap.Load > 0 { + cap.Load-- + cap.Mu.Unlock() + return nil + } + cap.Mu.Unlock() + } + return fmt.Errorf("external capability not found: %s", name) +} + +// SelectRunner returns the runner with the most available capacity for the given +// capability name that also satisfies the options filter. If filter is empty all +// runners are considered. Returns nil if no matching runner is found. +func (extCaps *ExternalCapabilities) SelectRunner(name string, filter map[string]string) *ExternalCapability { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[name] + if !ok { + return nil + } + var best *ExternalCapability + var bestAvail int + for _, cap := range runners { + if !AnyOptionsMatch(filter, cap.GetWorkerOptionsCopy()) { + continue + } + avail := cap.Capacity - cap.Load + if avail > bestAvail { + bestAvail = avail + best = cap + } + } + return best +} + +// SelectAndReserveRunner atomically selects the runner with the most available +// capacity that satisfies the options filter, then increments its Load. +// If filter is empty all runners are considered. +// Returns an error if no matching runner with available capacity is found. +func (extCaps *ExternalCapabilities) SelectAndReserveRunner(name string, filter map[string]string) (*ExternalCapability, error) { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + runners, ok := extCaps.Capabilities[name] + if !ok { + return nil, fmt.Errorf("no runners registered for capability %v", name) + } + var best *ExternalCapability + var bestAvail int + for _, cap := range runners { + if !AnyOptionsMatch(filter, cap.GetWorkerOptionsCopy()) { + continue + } + avail := cap.Capacity - cap.Load + if avail > bestAvail { + bestAvail = avail + best = cap + } + } + if best == nil || bestAvail <= 0 { + return nil, fmt.Errorf("no available capacity for capability %v", name) + } + best.Mu.Lock() + best.Load++ + best.Mu.Unlock() + return best, nil +} + +// GetCapabilityWorkerOptions returns the aggregated WorkerOptions from all runners +// registered for the given capability name. +func (extCaps *ExternalCapabilities) GetCapabilityWorkerOptions(extCap string) []map[string]interface{} { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + + runners, ok := extCaps.Capabilities[extCap] + if !ok { + return nil + } + var result []map[string]interface{} + for _, cap := range runners { + result = append(result, cap.GetWorkerOptionsCopy()...) + } + return result +} + +// GetAllWorkerOptions returns the cached WorkerOptions from every registered runner +// across all capabilities, flattened into a single slice. +func (extCaps *ExternalCapabilities) GetAllWorkerOptions() []map[string]interface{} { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + + var result []map[string]interface{} + for _, runners := range extCaps.Capabilities { + for _, cap := range runners { + result = append(result, cap.GetWorkerOptionsCopy()...) + } + } + return result +} + +// GetAllWorkerOptionsByCapability returns the cached WorkerOptions grouped by +// capability name. Each key is a capability name; the value is the merged +// options from all runners registered for that capability. +func (extCaps *ExternalCapabilities) GetAllWorkerOptionsByCapability() map[string][]map[string]interface{} { + extCaps.capm.Lock() + defer extCaps.capm.Unlock() + + result := make(map[string][]map[string]interface{}) + for capName, runners := range extCaps.Capabilities { + for _, cap := range runners { + result[capName] = append(result[capName], cap.GetWorkerOptionsCopy()...) + } + } + return result +} + func (extCaps *ExternalCapabilities) RegisterCapability(extCapability string) (*ExternalCapability, error) { extCaps.capm.Lock() defer extCaps.capm.Unlock() if extCaps.Capabilities == nil { - extCaps.Capabilities = make(map[string]*ExternalCapability) + extCaps.Capabilities = make(map[string]map[string]*ExternalCapability) } var extCap ExternalCapability err := json.Unmarshal([]byte(extCapability), &extCap) @@ -219,14 +447,20 @@ func (extCaps *ExternalCapabilities) RegisterCapability(extCapability string) (* if err != nil { panic(fmt.Errorf("error converting price: %v", err)) } - if cap, ok := extCaps.Capabilities[extCap.Name]; ok { - cap.Url = extCap.Url - cap.Capacity = extCap.Capacity - cap.price = extCap.price - cap.AuthToken = extCap.AuthToken + if nameMap, ok := extCaps.Capabilities[extCap.Name]; ok { + if cap, ok := nameMap[extCap.Url]; ok { + cap.Url = extCap.Url + cap.Capacity = extCap.Capacity + cap.price = extCap.price + cap.AuthToken = extCap.AuthToken + return cap, err + } } - extCaps.Capabilities[extCap.Name] = &extCap + if extCaps.Capabilities[extCap.Name] == nil { + extCaps.Capabilities[extCap.Name] = make(map[string]*ExternalCapability) + } + extCaps.Capabilities[extCap.Name][extCap.Url] = &extCap return &extCap, err } @@ -236,3 +470,32 @@ func (extCap *ExternalCapability) GetPrice() *big.Rat { defer extCap.Mu.RUnlock() return extCap.price.Value() } + +func (extCap *ExternalCapability) SetWorkerOptions(options []map[string]interface{}) { + extCap.Mu.Lock() + defer extCap.Mu.Unlock() + extCap.WorkerOptions = copyWorkerOptionsList(options) +} + +func (extCap *ExternalCapability) GetWorkerOptionsCopy() []map[string]interface{} { + extCap.Mu.RLock() + defer extCap.Mu.RUnlock() + return copyWorkerOptionsList(extCap.WorkerOptions) +} + + +func copyWorkerOptionsList(in []map[string]interface{}) []map[string]interface{} { + if len(in) == 0 { + return nil + } + out := make([]map[string]interface{}, len(in)) + for i, m := range in { + mc := make(map[string]interface{}, len(m)) + for k, v := range m { + mc[k] = v + } + out[i] = mc + } + return out +} + diff --git a/core/external_capabilities_test.go b/core/external_capabilities_test.go index ed0ff1e469..efb2b0eee9 100644 --- a/core/external_capabilities_test.go +++ b/core/external_capabilities_test.go @@ -48,7 +48,7 @@ func TestExternalCapabilities_RegisterCapability(t *testing.T) { // Verify it's in the map assert.Contains(t, extCaps.Capabilities, "test-cap") - assert.Equal(t, cap, extCaps.Capabilities["test-cap"]) + assert.Equal(t, cap, extCaps.Capabilities["test-cap"]["http://localhost:8000"]) }) t.Run("Register with missing price_scaling", func(t *testing.T) { @@ -115,7 +115,7 @@ func TestExternalCapabilities_RegisterCapability(t *testing.T) { assert.Equal(t, int64(2000), updatedCap.PriceScaling) // Verify it's in the map - storedCap := extCaps.Capabilities["update-test"] + storedCap := extCaps.Capabilities["update-test"]["http://localhost:9000"] assert.Equal(t, "http://localhost:9000", storedCap.Url) assert.Equal(t, 10, storedCap.Capacity) assert.NotNil(t, storedCap.price) diff --git a/core/options_filter.go b/core/options_filter.go new file mode 100644 index 0000000000..e9ad2528ec --- /dev/null +++ b/core/options_filter.go @@ -0,0 +1,136 @@ +package core + +import ( + "fmt" + "strconv" + "strings" +) + +// AnyOptionsMatch returns true if at least one entry in options passes EvaluateOptions. +// An empty options slice with a non-empty filter returns false. +func AnyOptionsMatch(filter map[string]string, options []map[string]interface{}) bool { + if len(filter) == 0 { + return true + } + for _, opt := range options { + if EvaluateOptions(filter, opt) { + return true + } + } + return false +} + +// FindMatchingOption returns the first options entry that passes EvaluateOptions, +// or nil if none match (or filter is empty). +func FindMatchingOption(filter map[string]string, options []map[string]interface{}) map[string]interface{} { + for _, opt := range options { + if EvaluateOptions(filter, opt) { + return opt + } + } + return nil +} + +// EvaluateOptions checks whether all filter constraints pass against worker options. +func EvaluateOptions(filter map[string]string, options map[string]interface{}) bool { + if len(filter) == 0 { + return true + } + + for key, filterVal := range filter { + workerVal, ok := options[key] + if !ok { + return false + } + + filterVal = strings.TrimSpace(filterVal) + if filterVal == "" { + return false + } + + switch { + case strings.HasPrefix(filterVal, ">="): + if !evaluateMath(filterVal[2:], workerVal, ">=") { + return false + } + case strings.HasPrefix(filterVal, "<="): + if !evaluateMath(filterVal[2:], workerVal, "<=") { + return false + } + case strings.HasPrefix(filterVal, ">"): + if !evaluateMath(filterVal[1:], workerVal, ">") { + return false + } + case strings.HasPrefix(filterVal, "<"): + if !evaluateMath(filterVal[1:], workerVal, "<") { + return false + } + default: + workerStr := strings.TrimSpace(fmt.Sprintf("%v", workerVal)) + if !strings.EqualFold(filterVal, workerStr) { + return false + } + } + } + + return true +} + +func evaluateMath(expectedStr string, workerVal interface{}, operator string) bool { + expectedFloat, err := strconv.ParseFloat(strings.TrimSpace(expectedStr), 64) + if err != nil { + return false + } + + workerFloat, ok := parseFloat(workerVal) + if !ok { + return false + } + + switch operator { + case ">=": + return workerFloat >= expectedFloat + case "<=": + return workerFloat <= expectedFloat + case ">": + return workerFloat > expectedFloat + case "<": + return workerFloat < expectedFloat + default: + return false + } +} + +func parseFloat(v interface{}) (float64, bool) { + switch x := v.(type) { + case float64: + return x, true + case float32: + return float64(x), true + case int: + return float64(x), true + case int8: + return float64(x), true + case int16: + return float64(x), true + case int32: + return float64(x), true + case int64: + return float64(x), true + case uint: + return float64(x), true + case uint8: + return float64(x), true + case uint16: + return float64(x), true + case uint32: + return float64(x), true + case uint64: + return float64(x), true + case string: + f, err := strconv.ParseFloat(strings.TrimSpace(x), 64) + return f, err == nil + default: + return 0, false + } +} diff --git a/core/options_filter_test.go b/core/options_filter_test.go new file mode 100644 index 0000000000..e909aa8e5c --- /dev/null +++ b/core/options_filter_test.go @@ -0,0 +1,39 @@ +package core + +import "testing" + +func TestEvaluateOptions(t *testing.T) { + workerOptions := map[string]interface{}{ + "model": "llama-3", + "vram_gb": 24.0, + "cuda_enabled": true, + "rtt_ms": "1200", + } + + tests := []struct { + name string + filter map[string]string + expected bool + }{ + {name: "Empty filter passes", filter: map[string]string{}, expected: true}, + {name: "Exact string passes", filter: map[string]string{"model": "llama-3"}, expected: true}, + {name: "Exact string case insensitive passes", filter: map[string]string{"model": "LLaMA-3"}, expected: true}, + {name: "Boolean exact passes", filter: map[string]string{"cuda_enabled": "true"}, expected: true}, + {name: "Boolean mismatch fails", filter: map[string]string{"cuda_enabled": "false"}, expected: false}, + {name: "Math less-than passes", filter: map[string]string{"rtt_ms": "<1500"}, expected: true}, + {name: "Math greater-than-equal passes", filter: map[string]string{"vram_gb": ">=16"}, expected: true}, + {name: "Missing key fails", filter: map[string]string{"gpu_temp": "<80"}, expected: false}, + {name: "Math condition fails", filter: map[string]string{"vram_gb": ">32"}, expected: false}, + {name: "Invalid numeric filter fails", filter: map[string]string{"vram_gb": ">=abc"}, expected: false}, + {name: "Combined filters pass", filter: map[string]string{"model": "llama-3", "vram_gb": ">=16", "rtt_ms": "<1500"}, expected: true}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + result := EvaluateOptions(tc.filter, workerOptions) + if result != tc.expected { + t.Errorf("Expected %v but got %v for filter %v", tc.expected, result, tc.filter) + } + }) + } +} diff --git a/core/orch_test.go b/core/orch_test.go index 22edcf021f..63e87e9d80 100644 --- a/core/orch_test.go +++ b/core/orch_test.go @@ -1684,8 +1684,8 @@ func TestBYOCExternalCapsPriceInfo(t *testing.T) { addr1 := "0x1000000000000000000000000000000000000000" - n.ExternalCapabilities.Capabilities["my-service"] = &ExternalCapability{Name: "my-service"} - n.ExternalCapabilities.Capabilities["another-service"] = &ExternalCapability{Name: "another-service"} + n.ExternalCapabilities.Capabilities["my-service"] = map[string]*ExternalCapability{"http://localhost": {Name: "my-service"}} + n.ExternalCapabilities.Capabilities["another-service"] = map[string]*ExternalCapability{"http://localhost": {Name: "another-service"}} n.SetPriceForExternalCapability("default", "my-service", big.NewRat(10, 1)) n.SetPriceForExternalCapability("default", "another-service", big.NewRat(20, 1)) @@ -1730,7 +1730,7 @@ func TestBYOCExternalCapsSenderPricing(t *testing.T) { addr2 := "0x2000000000000000000000000000000000000000" addr3 := "0x3000000000000000000000000000000000000000" - n.ExternalCapabilities.Capabilities["my-service"] = &ExternalCapability{Name: "my-service"} + n.ExternalCapabilities.Capabilities["my-service"] = map[string]*ExternalCapability{"http://localhost": {Name: "my-service"}} n.SetPriceForExternalCapability("default", "my-service", big.NewRat(10, 1)) n.SetPriceForExternalCapability(addr1, "my-service", big.NewRat(100, 1)) n.SetPriceForExternalCapability(addr2, "my-service", big.NewRat(200, 1)) @@ -1776,7 +1776,7 @@ func TestBYOCExternalCapsPriceEdgeCases(t *testing.T) { if tt.nilExtCaps { n.ExternalCapabilities = nil } else { - n.ExternalCapabilities.Capabilities["svc"] = &ExternalCapability{Name: "svc"} + n.ExternalCapabilities.Capabilities["svc"] = map[string]*ExternalCapability{"http://localhost": {Name: "svc"}} n.SetPriceForExternalCapability("default", "svc", tt.price) } diff --git a/doc/byoc-job-filtering-architecture.md b/doc/byoc-job-filtering-architecture.md new file mode 100644 index 0000000000..9e075a0255 --- /dev/null +++ b/doc/byoc-job-filtering-architecture.md @@ -0,0 +1,585 @@ +# Technical Specification: BYOC Options-Based Filtering & Evaluation Engine + +## 1. Overview + +This specification details a mechanism for dynamic, capability-based routing in the Bring Your Own Compute (BYOC) network. By decoupling domain-specific worker capabilities from the core Orchestrator and Gateway routing logic, BYOC workers can self-report their configuration (`options`), and clients can define strict requirements (`filters`). The Gateway evaluates these filters using a lightweight matching engine to determine Orchestrator eligibility. + +## 2. Data Structures + +To facilitate capability matching, both the client's job request payload and the Orchestrator's returned token must be extended to include schema-less JSON objects. + +### 2.1. Client Job Request (`JobParameters`) + +The client defines its requirements inside the existing `JobParameters` struct. The filter values are always passed as strings to accommodate operator prefixes. + +```go +type JobParameters struct { + // Existing fields... + + // Key-value map of required capabilities and their constraints + // Example: {"model": "llama-3", "vram_gb": ">=16", "cuda_enabled": "true"} + OptionsFilter map[string]string `json:"options_filter,omitempty"` +} + +``` + +### 2.2. Orchestrator Job Token (`JobToken`) + +The Orchestrator caches and returns the specific worker's self-reported capabilities. + +```go +type JobToken struct { + // Existing fields... + + // Key-value map of the worker's self-reported capabilities + // Example: {"model": "llama-3", "vram_gb": 24.0, "cuda_enabled": true} + WorkerOptions map[string]interface{} `json:"worker_options,omitempty"` +} + +``` + +## 3. The Evaluation Engine (v1) + +The Gateway implements a localized evaluation engine that compares the client's `OptionsFilter` against the `JobToken`'s `WorkerOptions`. + +### 3.1. Supported Operators + +To prevent performance bottlenecks and security risks (e.g., ReDoS), the evaluation engine is strictly limited to the following operations: + +* **Exact String Match:** Implicitly evaluated if no operator is present (e.g., `"model": "llama-3"`). +* **Boolean Check:** Evaluated as an exact match for `"true"` or `"false"` (e.g., `"cuda_enabled": "true"`). +* **Simple Math:** Supported via prefixes on numerical values: +* `<` (Less than) +* `>` (Greater than) +* `<=` (Less than or equal to) +* `>=` (Greater than or equal to) + + + +### 3.2. Evaluation Logic & Rules + +The engine processes the filter iteratively. For a token to pass, **all** keys present in the `OptionsFilter` must be successfully evaluated against `WorkerOptions`. + +1. **Key Existence:** If an `OptionsFilter` key does not exist in `WorkerOptions`, the evaluation immediately fails (returns `false`). +2. **Type Inference & Parsing:** * The engine inspects the prefix of the `OptionsFilter` value string. +* If a math operator (`<, >, <=, >=`) is detected, the engine attempts to cast the corresponding `WorkerOptions` value to a `float64`. It then parses the remaining string in the `OptionsFilter` as a `float64`. If either parsing fails, the evaluation fails. + + +3. **Strict Evaluation:** +* If no math operator is present, the engine converts both the filter value and the option value to strings and performs a case-insensitive exact match. +* If a math operator is present, the engine performs the requested mathematical comparison on the parsed floats. + + + +## 4. Architecture & Data Flow + +To avoid synchronous delays during job requests, Orchestrators handle option retrieval asynchronously. + +1. **Worker Registration & Polling:** When a BYOC worker registers with an Orchestrator, the Orchestrator periodically polls the worker's `/options` endpoint (e.g., every 30 seconds) and caches the resulting JSON. +2. **Job Request Initialization:** A client sends a job to the Gateway, including an `OptionsFilter` in the `JobParameters`. +3. **Token Gathering:** The Gateway requests job tokens from available Orchestrators. The Orchestrators immediately respond with their cached `WorkerOptions` injected into the `JobToken`. +4. **Gateway Evaluation:** The Gateway passes each `JobToken` through the Evaluation Engine. +5. **Execution:** The Gateway drops any tokens that fail the evaluation and routes the job to the optimal remaining Orchestrator. + +--- + +## 5. Future Enhancements + +To maintain a stable v1 release, several advanced routing and telemetry features have been deferred to future iterations. + +* **Trustless Performance Routing (Gateway-Observed Metrics):** While workers self-report static *capabilities* (`options`), relying on workers to self-report dynamic *performance* metrics (latency, queue depth) introduces trust vulnerabilities. Future iterations will introduce a `MetricsFilter`, which the Gateway will evaluate against its own historically observed and locally tracked performance data for each Orchestrator/Worker pair. +* **Fleet-Wide State Sharing (Redis Integration):** + Currently, each Gateway node must discover Orchestrator capabilities and track performance independently. Implementing a centralized state store (like Redis) will allow a fleet of Gateway nodes to share a unified pool of cached worker options and real-time failure metrics, drastically reducing job routing latency. +* **Regular Expression (Regex) Matching:** + Future versions of the evaluation engine may support a `~=` operator for Regex matching to allow clients more flexibility in capability targeting. This will require strict implementation safeguards, including execution timeouts and regex sanitization, to protect the Gateway from ReDoS attacks. +* **Nested JSON Evaluation:** + Expanding the evaluation engine to traverse deep JSON structures (e.g., `{"hardware": {"gpu": {"vram": ">=16"}}}`) using dot-notation string parsing. + +--- + +### Core Evaluation Logic + +This code handles type inference for the math operators, gracefully falls back to string formatting for exact matches, and safely handles the `interface{}` types coming from the JSON unmarshaling. + +```go +package byoc + +import ( + "fmt" + "strconv" + "strings" +) + +// EvaluateOptions checks if a JobToken's options satisfy a JobRequest's filter. +func EvaluateOptions(filter map[string]string, options map[string]interface{}) bool { + // If no filter is provided, the token implicitly passes. + if len(filter) == 0 { + return true + } + + for key, filterVal := range filter { + workerVal, exists := options[key] + if !exists { + return false // Fail immediately if the required capability is missing + } + + filterVal = strings.TrimSpace(filterVal) + + // Route to math evaluation if a supported prefix is found + if strings.HasPrefix(filterVal, ">=") { + if !evaluateMath(filterVal[2:], workerVal, ">=") { return false } + } else if strings.HasPrefix(filterVal, "<=") { + if !evaluateMath(filterVal[2:], workerVal, "<=") { return false } + } else if strings.HasPrefix(filterVal, ">") { + if !evaluateMath(filterVal[1:], workerVal, ">") { return false } + } else if strings.HasPrefix(filterVal, "<") { + if !evaluateMath(filterVal[1:], workerVal, "<") { return false } + } else { + // Fallback to exact string/boolean match (case-insensitive) + workerStr := fmt.Sprintf("%v", workerVal) + if strings.ToLower(filterVal) != strings.ToLower(strings.TrimSpace(workerStr)) { + return false + } + } + } + + return true +} + +// evaluateMath safely attempts to cast interface values to float64 and evaluates the operator. +func evaluateMath(expectedStr string, workerVal interface{}, operator string) bool { + expectedFloat, err := strconv.ParseFloat(strings.TrimSpace(expectedStr), 64) + if err != nil { + return false // Filter format is invalid (e.g., ">=abc") + } + + var workerFloat float64 + switch v := workerVal.(type) { + case float64: + workerFloat = v + case float32: + workerFloat = float64(v) + case int: + workerFloat = float64(v) + case int64: + workerFloat = float64(v) + case string: + parsed, err := strconv.ParseFloat(strings.TrimSpace(v), 64) + if err != nil { + return false // Worker value cannot be evaluated mathematically + } + workerFloat = parsed + default: + return false // Unsupported type for math operations + } + + // Apply the operator + switch operator { + case ">=": return workerFloat >= expectedFloat + case "<=": return workerFloat <= expectedFloat + case ">": return workerFloat > expectedFloat + case "<": return workerFloat < expectedFloat + } + + return false +} + +``` + +--- + +### Example Test Cases + +Here is a quick unit test structure to validate the logic against the scenarios you mentioned: + +```go +package byoc + +import ( + "testing" +) + +func TestEvaluateOptions(t *testing.T) { + // Simulated worker options returned from the Orchestrator + workerOptions := map[string]interface{}{ + "model": "llama-3", + "vram_gb": 24.0, // Unmarshaled from JSON number + "cuda_enabled": true, // Unmarshaled from JSON boolean + "rtt_ms": "1200", // Edge case: numbers stored as strings + } + + tests := []struct { + name string + filter map[string]string + expected bool + }{ + { + name: "Exact string match passes", + filter: map[string]string{"model": "llama-3"}, + expected: true, + }, + { + name: "Boolean evaluation as string passes", + filter: map[string]string{"cuda_enabled": "true"}, + expected: true, + }, + { + name: "Math operator less-than passes", + filter: map[string]string{"rtt_ms": "<1500"}, + expected: true, + }, + { + name: "Math operator greater-than-or-equal passes", + filter: map[string]string{"vram_gb": ">=16"}, + expected: true, + }, + { + name: "Missing key fails", + filter: map[string]string{"gpu_temp": "<80"}, + expected: false, + }, + { + name: "Math operator fails condition", + filter: map[string]string{"vram_gb": ">32"}, + expected: false, + }, + { + name: "Combined filters pass", + filter: map[string]string{"model": "llama-3", "vram_gb": ">=16", "rtt_ms": "<1500"}, + expected: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + result := EvaluateOptions(tc.filter, workerOptions) + if result != tc.expected { + t.Errorf("Expected %v but got %v for filter %v", tc.expected, result, tc.filter) + } + }) + } +} + +``` + +--- + +## How to use in job_gateway.go + +The best place to intercept and filter the orchestrators is **during the token collection phase**, before the Gateway even attempts to route the job. By filtering the tokens as they arrive from the Orchestrators over the channel, you prevent invalid orchestrators from ever entering the `gatewayJob.Orchs` retry loop in `submitJob`. + +Here is the step-by-step injection: + +### Step 1: Update the Shared Structs + +Make sure your request and token structs (likely in your `core` or `net` packages) have the new fields. + +```go +// Inside the package where JobToken is defined +type JobToken struct { + // ... existing fields ... + AvailableCapacity int `json:"availableCapacity"` + WorkerOptions map[string]interface{} `json:"worker_options,omitempty"` // Add this +} + +// Inside the package where JobParameters is defined +type JobParameters struct { + // ... existing fields ... + OptionsFilter map[string]string `json:"options_filter,omitempty"` // Add this +} + +``` + +### Step 2: Inject the Filter into `job_gateway.go` + +Locate the token gathering loop in `job_gateway.go` (around line 170-190 in the snippet you shared earlier). You want to wrap the `append` operation with your new `EvaluateOptions` check. + +Here is the modified block: + +```go +// ... existing code in job_gateway.go ... + +var jobTokens []JobToken +nbResp := 0 +numAvailableOrchs := node.OrchestratorPool.Size() +tokenCh := make(chan JobToken, numAvailableOrchs) +errCh := make(chan error, numAvailableOrchs) + +tokensCtx, cancel := context.WithTimeout(clog.Clone(context.Background(), ctx), timeout) +defer cancel() + +// Shuffle and get job tokens +for _, i := range rand.Perm(len(orchs)) { + //do not send to excluded Orchestrators + if slices.Contains(params.Orchestrators.Exclude, orchs[i].URL.String()) { + numAvailableOrchs-- + continue + } + //if include is set, only send to those Orchestrators + if len(params.Orchestrators.Include) > 0 && !slices.Contains(params.Orchestrators.Include, orchs[i].URL.String()) { + numAvailableOrchs-- + continue + } + + go getOrchJobToken(ctx, orchs[i].URL, *reqSender, respTimeout, tokenCh, errCh) +} + +// THE INJECTION POINT: Filter tokens as they are received +for nbResp < numAvailableOrchs && len(jobTokens) < numAvailableOrchs { + select { + case token := <-tokenCh: + // 1. Check if Orchestrator has capacity + if token.AvailableCapacity > 0 { + + // 2. NEW: Evaluate the BYOC Worker Options against the client's filter + // Assuming 'params' is the JobParameters struct available in this scope + if EvaluateOptions(params.OptionsFilter, token.WorkerOptions) { + jobTokens = append(jobTokens, token) + } else { + clog.V(common.DEBUG).Infof(ctx, "Orchestrator %v rejected: failed options filter", token.ServiceAddr) + } + + } + nbResp++ + case <-errCh: + nbResp++ + case <-tokensCtx.Done(): + //searchTimeout reached, return tokens received + return jobTokens, nil + } +} + +// received enough tokens or all responses arrived +return jobTokens, nil + +// ... rest of the file ... + +``` + +### Why this approach works best: + +1. **Fails Fast:** The Gateway doesn't waste time signing payloads or initiating HTTP requests to Orchestrators that don't have the right hardware/models. +2. **Keeps `submitJob` Clean:** The main retry loop in `submitJob` remains completely untouched. It just receives a pre-vetted list of `gatewayJob.Orchs` and loops through them exactly as it did before. +3. **No Extra Latency:** Because the Orchestrator is returning its cached `WorkerOptions` directly inside the `/process/token` response payload, evaluating the filter locally adds virtually zero milliseconds to the Gateway's routing overhead. + +--- + + +## Orchestrator-side code changes + +To complete the loop, we need to set up the Orchestrator so it can seamlessly pass these options down to the Gateway. + +Because the Gateway's token request (`/process/token`) is in the "hot path" of job routing, the Orchestrator cannot afford to make a synchronous HTTP request to the BYOC worker to ask for its options. If the worker is slow to respond, the Gateway's token request times out, and the job fails to route. + +The solution is an **asynchronous polling loop** with a thread-safe cache. The Orchestrator constantly asks the worker for its options in the background and stores them in memory. When the Gateway asks for a token, the Orchestrator instantly attaches the cached data. + +Here is how to wire up the Orchestrator side: + +### Step 1: Thread-Safe State on the Orchestrator + +We need a place to store the options on the Orchestrator's internal representation of the BYOC worker. Because a background thread will be writing to this cache while multiple HTTP request threads might be reading from it, we must use a `sync.RWMutex` to prevent race conditions. + +```go +package byoc + +import ( + "context" + "encoding/json" + "net/http" + "sync" + "time" + "github.com/livepeer/go-livepeer/clog" +) + +// WorkerClient represents the Orchestrator's connection to a BYOC worker. +type WorkerClient struct { + URL string // The address of the BYOC worker container + + // Thread-safe cache for worker options + mu sync.RWMutex + cachedOptions map[string]interface{} +} + +``` + +### Step 2: The Background Polling Loop + +When the worker registers with the Orchestrator (or when the Orchestrator initializes), it should spin up a background Goroutine that polls the worker's `/options` endpoint on a set interval (e.g., every 30 seconds). + +```go +// StartOptionPolling begins the background loop to fetch worker capabilities. +func (wc *WorkerClient) StartOptionPolling(ctx context.Context) { + // Fetch immediately on startup so we don't have to wait for the first tick + wc.fetchAndUpdateOptions(ctx) + + ticker := time.NewTicker(30 * time.Second) + + go func() { + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return // Stop polling if the Orchestrator shuts down or worker disconnects + case <-ticker.C: + wc.fetchAndUpdateOptions(ctx) + } + } + }() +} + +func (wc *WorkerClient) fetchAndUpdateOptions(ctx context.Context) { + // Send a GET request to the worker's capabilities endpoint + resp, err := http.Get(wc.URL + "/options") + if err != nil { + clog.Errorf(ctx, "Failed to poll options from worker %s: %v", wc.URL, err) + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + clog.Errorf(ctx, "Worker %s returned status %d for /options", wc.URL, resp.StatusCode) + return + } + + var newOptions map[string]interface{} + if err := json.NewDecoder(resp.Body).Decode(&newOptions); err != nil { + clog.Errorf(ctx, "Failed to decode options from worker %s: %v", wc.URL, err) + return + } + + // Safely update the cache + wc.mu.Lock() + wc.cachedOptions = newOptions + wc.mu.Unlock() + + clog.V(5).Infof(ctx, "Successfully refreshed options for worker %s", wc.URL) +} + +``` + +### Step 3: Injecting the Cache into the Token Request + +Finally, locate the HTTP handler on the Orchestrator that serves the `/process/token` endpoint. When constructing the `JobToken` to send back to the Gateway, safely read from the cache. + +```go +// Inside the Orchestrator's HTTP handler for /process/token +func (s *OrchestratorServer) handleProcessToken(w http.ResponseWriter, r *http.Request) { + // ... existing token generation logic (verifying signatures, capacity, etc.) ... + + // Assuming 'workerClient' is the instance managing the specific BYOC worker + workerClient := s.getWorkerClientForJob(r) + + // Safely read the cached options + workerClient.mu.RLock() + // Create a shallow copy to prevent the Gateway struct from holding a reference to the internal map + optionsCopy := make(map[string]interface{}, len(workerClient.cachedOptions)) + for k, v := range workerClient.cachedOptions { + optionsCopy[k] = v + } + workerClient.mu.RUnlock() + + // Construct the response + jobToken := JobToken{ + // ... existing fields (Token, ServiceAddr, AvailableCapacity) ... + WorkerOptions: optionsCopy, // Inject the copied cache + } + + // Send the JSON response back to the Gateway + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(jobToken) +} + +``` + +--- + +## Worker Code Changes + +Let’s dive into the supply side: the BYOC worker. + +To make this concrete, let's assume this worker is running an AI video generation pipeline. The goal is to build a lightweight HTTP server inside the container that serves a JSON representation of its hardware and loaded software stack. + +Here is how you would structure the worker to expose its capabilities. + +### 1. The Strategy: Static vs. Dynamic Discovery + +When the worker container boots up, it should assemble its "resume" by combining two types of data: + +* **Static Configuration:** Things defined by the container image or startup environment variables (e.g., the specific model loaded, the AI framework). +* **Dynamic Hardware Discovery:** Things the container detects from the host machine at runtime (e.g., OS, GPU model, total VRAM). + +### 2. The Implementation (Python Example) + +Since AI pipelines relying on tools like Diffusers typically run in Python, setting up a fast, lightweight web server using a framework like FastAPI is the standard approach. + +Here is what the worker code would look like to serve that `/options` endpoint: + +```python +from fastapi import FastAPI +import uvicorn +import subprocess + +app = FastAPI() + +# 1. Static capabilities defined by the container's purpose +STATIC_OPTIONS = { + "worker_type": "ai-video-generation", + "framework": "diffusers", + "os": "Ubuntu 24.04" +} + +def get_vram_gb() -> float: + # In a real scenario, you might parse nvidia-smi output here: + # subprocess.check_output(["nvidia-smi", "--query-gpu=memory.total", "--format=csv,noheader,nounits"]) + # For this example, we'll return a known hardware state: + return 32.0 + +def get_gpu_model() -> str: + # Similarly, this could be dynamically probed + return "NVIDIA GeForce RTX 5090" + +@app.get("/options") +def serve_options(): + """ + The endpoint the Livepeer Orchestrator polls asynchronously. + """ + # Merge static config with dynamic hardware state + options = STATIC_OPTIONS.copy() + options["gpu_model"] = get_gpu_model() + options["vram_gb"] = get_vram_gb() + options["cuda_enabled"] = True + + return options + +# The actual BYOC job processing endpoint would go here +@app.post("/process") +def process_job(request: dict): + # Handle the video generation job... + pass + +if __name__ == "__main__": + # Run the worker API on port 8080 + uvicorn.run(app, host="0.0.0.0", port=8080) + +``` + +### 3. The Output Payload + +When the Orchestrator runs its background polling loop against `http://:8080/options`, it receives this exact JSON payload: + +```json +{ + "worker_type": "ai-video-generation", + "framework": "diffusers", + "os": "Ubuntu 24.04", + "gpu_model": "NVIDIA GeForce RTX 5090", + "vram_gb": 32.0, + "cuda_enabled": true +} + +``` + +### Why this design shines + +1. **Zero Orchestrator Configuration:** The Orchestrator doesn't need to know *anything* about GPUs, VRAM, or Ubuntu. It just blindly caches this JSON dictionary and hands it to the Gateway. +2. **Worker Autonomy:** If you decide to swap the hardware or update the container to run a different model, you just restart the worker. The `/options` endpoint instantly updates, the Orchestrator caches the new data on its next poll, and the Gateway immediately starts routing jobs based on the new capabilities. diff --git a/doc/byoc-technical-details.md b/doc/byoc-technical-details.md new file mode 100644 index 0000000000..5a6bd13da7 --- /dev/null +++ b/doc/byoc-technical-details.md @@ -0,0 +1,437 @@ +# BYOC Technical Details + +This document is an implementation-level reference for **Bring Your Own Container (BYOC)** in +go-livepeer. For a quick-start tutorial see [`doc/byoc.md`](byoc.md), and for the Gateway +stream API reference see [`doc/byoc-streaming.md`](byoc-streaming.md). + +--- + +## 1. Overview + +BYOC lets an external worker process register itself with a Livepeer Orchestrator at runtime. +The Orchestrator then advertises that capability to Gateways, which can route paid jobs to it +through the standard Livepeer payment pipeline. + +`Capability_BYOC = 37` (`core/capabilities.go:91`) is a sentinel value used to tag BYOC price +entries in the capability price list. It is **not** a real pipeline — no built-in code handles +it as a workload. Its sole job is to distinguish externally-registered capability prices from +built-in AI capability prices when serialised over the network. + +--- + +## 2. Architecture: Two Models + +BYOC supports two interaction patterns. Choose the one that fits your workload. + +### Job Model (Batch / Request–Response) + +- **When to use**: stateless, discrete tasks — LLM inference, text processing, image analysis, + audio transcription, etc. +- The Gateway receives a `POST /process/request/` and proxies it through an + Orchestrator to the worker's base URL. +- The whole round-trip lives inside a single HTTP request (or an SSE stream for long-running + jobs). +- Payment is charged once at the end based on wall-clock seconds elapsed. + +### Stream Model (Long-Running / Stateful) + +- **When to use**: continuous data feeds — live video processing, real-time inference loops, + persistent pipelines. +- A `POST /process/stream/start` on the Gateway opens a persistent session identified by a + `streamId`. +- Media is exchanged over **Trickle** pub/sub channels that live on the Orchestrator. +- Payment is debited on a running clock: the Orchestrator debits every 23 s, the Gateway tops + up every 50 s. + +--- + +## 3. Registration + +### `POST /capability/register` + +Served by the **Orchestrator** at `BYOCOrchestratorServer.RegisterCapability()` +(`byoc/job_orchestrator.go:32`). + +**Authentication**: `Authorization: ` header — the same secret used for transcoder +attachment. + +**Request body** (JSON): maps directly to `ExternalCapability` (`core/external_capabilities.go:17`): + +```json +{ + "name": "my-pipeline", + "description": "human-readable description", + "url": "http://worker-host:5000", + "capacity": 4, + "price_per_unit": 1, + "price_scaling": 600, + "currency": "USD" +} +``` + +| Field | Type | Description | +|-----------------|--------|-------------| +| `name` | string | Unique capability identifier; used as routing key. | +| `description` | string | Free-form, not used by routing logic. | +| `url` | string | Base URL of the worker. Sub-paths are appended by the Orchestrator. | +| `capacity` | int | Maximum concurrent jobs. Managed via `Load` counter. | +| `price_per_unit`| int64 | Numerator of price fraction. | +| `price_scaling` | int64 | Denominator of price fraction (defaults to 1 if 0). | +| `currency` | string | Currency code for auto-conversion (e.g. `"USD"`, `"ETH"`, `"wei"`). | + +### Storage + +Registrations are stored in two places on the Orchestrator node: + +1. **`ExternalCapabilities.Capabilities`** (`core/external_capabilities.go:106`) — a + `map[string]*ExternalCapability` keyed by capability name. Holds the URL, capacity, and + computed price. +2. **`jobPriceInfo`** (`core/livepeernode.go:366`) — a `map[senderEthAddr]map[capName]*big.Rat` + used to look up the price when building payment responses. Populated via + `SetPriceForExternalCapability`. + +### Re-registration Behaviour + +`RegisterCapability` (`core/external_capabilities.go:198`) uses **last-writer-wins** semantics: +the incoming registration always overwrites the stored entry. Notably, the replacement struct +starts with `Load = 0`, so any active-job tracking is silently lost. Workers should not +re-register while jobs are in flight. + +### `POST /capability/unregister` + +Request body: plain-text capability name. Removes the entry from `ExternalCapabilities.Capabilities`. + +--- + +## 4. BYOC vs Built-in AI Capabilities + +The table below compares BYOC against representative built-in AI capabilities to show where +the implementation differs. + +| Property | LLM (33) | TextToImage (27) | AudioToText (31) | LiveVideoToVideo (35) | **BYOC (37)** | +|---|---|---|---|---|---| +| Registration time | Node startup (config file) | Node startup | Node startup | Node startup | **Runtime (`POST /capability/register`)** | +| In capability bitstring | Yes | Yes | Yes | Yes | **No** | +| `PerCapability` constraints (model IDs) | Yes | Yes | Yes | Yes | **No** | +| Warm/cold session routing | Yes | Yes | Yes | Yes (warm only) | **No** | +| Price store | `priceInfoForCaps` | `priceInfoForCaps` | `priceInfoForCaps` | `priceInfoForCaps` | **`jobPriceInfo`** | +| Capacity tracking | `Capacities` map in `Capabilities` | same | same | `Capacities` map | **`ExternalCapability.Load`** | +| Per-gateway price override | Yes (`priceInfoForCaps[ethAddr]`) | Yes | Yes | Yes | **No** | + +Key implications: + +- **Bitstring absence**: `Capability_BYOC` is never set in a node's capability bitstring, so + BYOC capabilities never appear in `CompatibleWith` filtering. Gateways discover BYOC + capabilities through the price list alone (`Capability_BYOC` + capability name as + `Constraint`). +- **No per-gateway price override**: built-in capabilities allow different prices per + Gateway ETH address via `priceInfoForCaps`. BYOC uses `jobPriceInfo` which is populated + from the registration payload and has no per-gateway variant. + +--- + +## 5. Job Model (Batch / Request–Response) + +### Gateway Route + +``` +/process/request/ → SubmitJob() (byoc/job_gateway.go) +/process/request/ +``` + +The Gateway: +1. Looks up available Orchestrators that advertise the requested BYOC capability. +2. Signs the job request and forwards it to the Orchestrator's `/process/request/`. + +### Orchestrator Processing (`byoc/job_orchestrator.go`) + +Per-call steps in `processJob`: + +1. **Signature verification** (`verifyJobCreds`): decodes the `Livepeer` header, checks the + sender's Ethereum signature over `request + parameters`. +2. **Capacity reservation** (`ReserveExternalCapabilityCapacity`): atomically increments + `ExternalCapability.Load`; returns `503` if `Load >= Capacity`. +3. **Payment verification** (`confirmPayment`): processes any ticket in the + `Livepeer-Payment` header and checks that the resulting balance covers at least 60 seconds + of compute at the registered rate. +4. **Sub-path forwarding**: strips the `/process/request/` prefix and appends the remainder + to the worker base URL: + ```go + // byoc/job_orchestrator.go:261 + workerRoute = workerRoute + "/" + workerResourceRoute + ``` +5. **Response proxy**: for non-SSE responses, reads the full body, charges for compute, and + returns. For SSE responses, streams lines to the client. + +### SSE Streaming + +When the worker response is `Content-Type: text/event-stream`, the Orchestrator: + +- Forwards lines from the worker to the client in real time. +- Runs a **balance ticker every 5 seconds** that debits `rate × 5` from the sender's balance. + If balance goes negative the stream is terminated with an `insufficient balance` event. +- Injects a final `data: {"balance": }` line just before `[DONE]`. + +### Job Charge + +`chargeForCompute` (`byoc/job_orchestrator.go:507`): + +```go +took := time.Since(start) +orch.DebitFees(sender, manifestID, price, int64(math.Ceil(took.Seconds()))) +``` + +Charge = `rate × ⌈seconds⌉`. Applied on every exit path (success, worker error, connection +error). + +--- + +## 6. Stream Model (Long-Running) + +### Gateway Routes (`byoc/byoc.go:164`) + +``` +POST /process/stream/start +POST /process/stream/{streamId}/update +POST /process/stream/{streamId}/stop +POST /process/stream/{streamId}/status (GET) +POST /process/stream/{streamId}/data (GET, SSE) +POST /process/stream/{streamId}/rtmp +POST /process/stream/{streamId}/whip +``` + +### Orchestrator Routes (`byoc/byoc.go:225`) + +``` +POST /ai/stream/start +POST /ai/stream/stop +POST /ai/stream/update +POST /ai/stream/payment +``` + +### Trickle Channel Setup + +On `/ai/stream/start` the Orchestrator creates Trickle channels on its local trickle server +and passes the URLs back to the Gateway in response headers. Channels are created +conditionally based on `JobParameters` flags: + +| Channel | Flag | Direction | MIME type | Response header | +|---------|------|-----------|-----------|-----------------| +| **pub** (video ingress) | `enable_video_ingress` | Gateway → Worker | `video/MP2T` | `X-Publish-Url` | +| **sub** (video egress) | `enable_video_egress` | Worker → Gateway | `video/MP2T` | `X-Subscribe-Url` | +| **control** | always | Gateway → Worker | `application/json` | `X-Control-Url` | +| **events** | always | Worker → Gateway | `application/json` | `X-Events-Url` | +| **data** | `enable_data_output` | Worker → Gateway | `application/jsonl` | `X-Data-Url` | + +The worker receives all enabled channel URLs in the body of `POST {url}/stream/start` as JSON +fields (`subscribe_url`, `publish_url`, `control_url`, `events_url`, `data_url`). + +### State Lifecycle + +``` +Gateway Orchestrator +─────────────────────────────────── ───────────────────────────────────── +BYOCStreamPipelines[streamId] ExternalCapabilities.Streams[streamId] + (created on POST /process/stream/start) (created on POST /ai/stream/start) + cancelled on stop/error cancelled on stop/balance-zero +``` + +The Gateway's `monitorStream` goroutine owns the pipeline lifecycle and calls +`removeStreamPipeline` on teardown. The Orchestrator's `monitorOrchStream` goroutine owns +the `Streams` entry and calls `RemoveStream` on teardown. + +--- + +## 7. Worker API Contract + +The external worker must expose an HTTP server at the URL registered in +`POST /capability/register`. + +### Job Model + +No specific path contract — the worker may expose any paths. The Orchestrator appends whatever +sub-path the client used after `/process/request/` and forwards the full request body and +`Content-Type` header unchanged. + +### Stream Model + +The Orchestrator calls the following fixed paths relative to the registered worker URL: + +| Method | Path | When called | Body | +|--------|------|-------------|------| +| `POST` | `{url}/stream/start` | Stream start | JSON with trickle URLs + original client body merged | +| `POST` | `{url}/stream/stop` | Stream stop | Original client stop body | +| `POST` | `{url}/stream/params` | Stream update | Original client update body | + +**Headers passed through on all stream calls:** + +- `X-Stream-Id` — the stream identifier, useful when a reverse proxy sits in front of multiple + worker instances. +- `Content-Type` — forwarded from the Gateway client request. + +The `stream/start` body includes: + +```json +{ + "gateway_request_id": "", + "control_url": "", + "events_url": "", + "subscribe_url": "", // only if enable_video_ingress + "publish_url": "", // only if enable_video_egress + "data_url": "", // only if enable_data_output + // ... original client body fields merged in +} +``` + +--- + +## 8. Payment System + +### Registration Fields + +```json +{ + "price_per_unit": 1, + "price_scaling": 600, + "currency": "USD" +} +``` + +These form a rational number `PricePerUnit / PriceScaling`. The `currency` field drives +automatic fiat-to-wei conversion via `AutoConvertedPrice`. + +### Core Formula + +``` +cost = (PricePerUnit / PriceScaling) × seconds +``` + +For example, `price_per_unit=1, price_scaling=600, currency="USD"` means +`1/600 USD per second = $0.10/minute = $6.00/hour`. + +### `DebitFees` Implementation (`core/orchestrator.go:475`) + +```go +priceRat := big.NewRat(price.GetPricePerUnit(), price.GetPixelsPerUnit()) +node.Balances.Debit(addr, manifestID, priceRat.Mul(priceRat, big.NewRat(pixels, 1))) +``` + +`PixelsPerUnit` is the field name inherited from the video transcoding world; in BYOC it acts +as the price scaling denominator (seconds-based), not a pixel count. The `units` argument +passed to `DebitFees` is always a number of seconds. + +### Stream Payment Lifecycle + +``` +t=0 Stream start request arrives + └─ confirmPayment checks balance ≥ rate × 60 (1 min pre-fund gate) + └─ chargeForCompute debits rate × ⌈start_latency_secs⌉ + +t+23s Orchestrator monitorOrchStream ticker fires + └─ DebitFees(sender, capability, price, 23) + └─ if balance < 0: warn, set shouldStopNextRound=true + +t+46s Orchestrator ticker fires again + └─ DebitFees again + └─ if balance still < 0 AND shouldStopNextRound: RemoveStream → stop + +t+50s Gateway monitorStream ticker fires + └─ getToken → fetch fresh ticket params + orchestrator balance + └─ createPayment → ticket batch covering next interval + └─ POST /ai/stream/payment with payment header + + /ai/stream/payment handler: + └─ validates request, then ONLY returns current balance in header + └─ does NOT debit — debit is the Orchestrator's responsibility +``` + +**Balance gate**: `minBal = rate × 60` (`byoc/job_orchestrator.go:470`). A stream is +rejected with `402 Payment Required` if the sender's balance is below this threshold at +stream start. + +**Cutoff**: two consecutive negative-balance rounds on the Orchestrator (~46 s of grace after +the balance goes negative). + +**Gateway payment interval**: 50 s (`stream_gateway.go:321`). + +**Orchestrator debit interval**: 23 s (`stream_orchestrator.go:233`). + +### Job Payment + +``` +chargeForCompute(start, price, sender, capability) + ← rate × ⌈time.Since(start).Seconds()⌉ +``` + +Applied once at the end of the request on all exit paths (success, error, timeout). + +### Worked Example + +Rate: **$0.10 per minute** + +```json +{ + "price_per_unit": 1, + "price_scaling": 600, + "currency": "USD" +} +``` + +| Metric | Value | +|--------|-------| +| Rate (wei/s) | `AutoConvert(1/600 USD)` | +| Minimum deposit (1 min pre-fund) | $0.10 | +| Hourly cost | $6.00 | +| Cost of a 7 s job | `rate × ⌈7⌉ = rate × 7` ≈ $0.012 | + +For a free tier (no payment required), set `price_per_unit=0`. + +--- + +## 9. Limitations and Gotchas + +### Single URL per Capability Name + +`ExternalCapabilities.Capabilities` is keyed by `name`. There is no built-in load balancing +across multiple worker URLs for the same capability. If you need horizontal scaling, put a +reverse proxy behind a single registered URL or register each worker under a distinct name. + +### Re-registration Resets the Load Counter + +`RegisterCapability` replaces the stored `*ExternalCapability` with a fresh struct whose +`Load` field is 0. Any jobs currently counted against the old struct lose their tracking. +Re-registering while jobs are in flight can cause `Load` to go negative (via `FreeExternalCapabilityCapacity`), permitting more concurrent jobs than `capacity` allows. + +### No Per-Gateway Price Override + +Built-in AI capabilities support different prices per Gateway ETH address via +`priceInfoForCaps[ethAddr]`. BYOC uses `jobPriceInfo` populated from the registration payload; +there is no mechanism to charge different Gateways different rates for the same BYOC +capability. + +### `Capability_BYOC` Never Appears in the Capability Bitstring + +Because `Capability_BYOC` is never set in the node's `CapabilityString`, Orchestrators that +support BYOC capabilities will not match on `CompatibleWith` bitstring checks. Gateways +discover BYOC availability through the `CapabilitiesPrices` list (where BYOC entries use +`Capability=37` and the capability name as `Constraint`), not through bitstring filtering. + +--- + +## Key Files Reference + +| File | What it contains | +|------|-----------------| +| `core/external_capabilities.go` | `ExternalCapability`, `ExternalCapabilities`, `StreamInfo`, `RegisterCapability` | +| `core/orchestrator.go:258` | `GetCapabilitiesPrices` — injects BYOC prices into the network advertisement | +| `core/orchestrator.go:475` | `DebitFees` implementation | +| `core/livepeernode.go:366` | `SetPriceForExternalCapability`, `GetPriceForJob`, `jobPriceInfo` map | +| `core/capabilities.go:87–133` | Capability enum and `CapabilityNameLookup` | +| `common/types.go:172` | `OrchNetworkCapabilities` — the wire format for capability discovery | +| `byoc/byoc.go` | Route registration for Gateway (`BYOCGatewayServer`) and Orchestrator (`BYOCOrchestratorServer`) | +| `byoc/types.go` | `JobRequest`, `JobParameters`, `BYOCStreamPipeline`, header constants | +| `byoc/job_orchestrator.go` | `ProcessJob`, `processJob`, `setupOrchJob`, `confirmPayment`, `chargeForCompute` | +| `byoc/stream_orchestrator.go` | `StartStream`, `monitorOrchStream`, `ProcessStreamPayment` | +| `byoc/stream_gateway.go` | `monitorStream`, `sendPaymentForStream`, `setupStream` | +| `byoc/payment.go` | `createPayment`, `updateGatewayBalance`, `ticketCountForCost` | +| `core/ai_orchestrator.go:1151` | `CheckExternalCapabilityCapacity`, `ReserveExternalCapabilityCapacity`, `FreeExternalCapabilityCapacity` | diff --git a/server/handlers.go b/server/handlers.go index b6ac1928e7..3cd2334f00 100644 --- a/server/handlers.go +++ b/server/handlers.go @@ -18,6 +18,7 @@ import ( ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/signer/core/apitypes" "github.com/golang/glog" + "github.com/livepeer/go-livepeer/byoc" "github.com/livepeer/go-livepeer/clog" "github.com/livepeer/go-livepeer/common" "github.com/livepeer/go-livepeer/core" @@ -281,12 +282,42 @@ func (s *LivepeerServer) getNetworkCapabilitiesHandler() http.Handler { respond500(w, "network capabilities not available") } - networkCapabilities := &networkCapabilitiesResponse{ - CapabilitiesNames: core.CapabilityNameLookup, - Orchestrators: orchNetworkCaps, + // Fan out to each orchestrator to fetch per-capability worker options. + // Build shallow copies of each cached struct so we never mutate the + // shared pointers held by LivepeerNode.NetworkCapabilities. + const optionsTimeout = 2 * time.Second + type result struct { + uri string + opts map[string][]map[string]interface{} + } + resultCh := make(chan result, len(orchNetworkCaps)) + for _, orch := range orchNetworkCaps { + go func(uri string) { + resultCh <- result{uri: uri, opts: byoc.FetchCapabilityOptions(r.Context(), uri, optionsTimeout)} + }(orch.OrchURI) + } + optsByURI := make(map[string]map[string][]map[string]interface{}, len(orchNetworkCaps)) + for range orchNetworkCaps { + res := <-resultCh + if len(res.opts) > 0 { + optsByURI[res.uri] = res.opts + } + } + glog.Infof("getNetworkCapabilities fetched capability_options num_orchs=%v num_with_options=%v", len(orchNetworkCaps), len(optsByURI)) + + // Copy each cached struct before setting CapabilityOptions so we + // never write back to the pointers owned by the discovery cache. + responseOrchNetworkCaps := make([]*common.OrchNetworkCapabilities, len(orchNetworkCaps)) + for i, orch := range orchNetworkCaps { + cp := *orch + cp.CapabilityOptions = optsByURI[orch.OrchURI] + responseOrchNetworkCaps[i] = &cp } - respondJson(w, networkCapabilities) + respondJson(w, &networkCapabilitiesResponse{ + CapabilitiesNames: core.CapabilityNameLookup, + Orchestrators: responseOrchNetworkCaps, + }) return } else { respond400(w, "Node must be gateway node to get network capabilities") From bb9ade509cc985a04e85f37396a2e8a4aa4d8e33 Mon Sep 17 00:00:00 2001 From: Mike Zupper Date: Thu, 23 Apr 2026 07:47:19 -0400 Subject: [PATCH 2/2] remote_signer: add byoc-request payment type Adds a new RemoteType_ByocRequest = "byoc-request" value for the RemotePaymentRequest.Type field. Semantics: - Caller supplies InPixels explicitly (no auto-calculation). - Under BYOC pricing, PixelsPerUnit denominates wei-per-second, so InPixels here represents "seconds of compute to pre-fund." - Rejected with 400 if InPixels is not positive. This is the first wiring of non-LV2V BYOC through the remote signer. Existing behavior (LV2V and empty-type) is unchanged; the switch statement replacement is a refactor of the prior if/else chain. Co-Authored-By: Claude Opus 4.7 (1M context) --- server/remote_signer.go | 29 +++++++++++++++++++++++++---- server/remote_signer_test.go | 11 +++++++++++ 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/server/remote_signer.go b/server/remote_signer.go index 59a925828f..fb272ffd9d 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -28,6 +28,7 @@ const HTTPStatusRefreshSession = 480 const HTTPStatusPriceExceeded = 481 const HTTPStatusNoTickets = 482 const RemoteType_LiveVideoToVideo = "lv2v" +const RemoteType_ByocRequest = "byoc-request" const PipelineLiveVideoToVideo = "live-video-to-video" // SignOrchestratorInfo handles signing GetOrchestratorInfo requests for multiple orchestrators @@ -166,10 +167,17 @@ type RemotePaymentRequest struct { // Set if an ID is needed to tie into orch accounting for a session. Optional ManifestID string - // Number of pixels to generate a ticket for. Required if `type` is not set. + // Number of pixels to generate a ticket for. + // Required if `type` is not set or is "byoc-request". + // For BYOC (PixelsPerUnit denominates wei-per-second) this value represents + // seconds of compute to pre-fund. InPixels int64 `json:"inPixels"` - // Job type to automatically calculate payments. Valid values: `lv2v`. Optional. + // Job type. Valid values: `lv2v`, `byoc-request`. Optional. + // - "lv2v": the signer auto-calculates pixels from wall-clock video metadata. + // - "byoc-request": no auto-calculation; the caller supplies InPixels explicitly. + // Used for BYOC batch requests (`POST /process/request/{capability}`). + // - "" (empty): generic pre-computed path; the caller supplies InPixels explicitly. Type string `json:"type"` // Capabilities to include in the ticket. Optional; may be set for the lv2v job type. @@ -380,7 +388,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req lastUpdate = now } billableSecs := now.Sub(lastUpdate).Seconds() - if req.Type == RemoteType_LiveVideoToVideo { + switch req.Type { + case RemoteType_LiveVideoToVideo: info := defaultSegInfo if billableSecs <= 0 { // preload with 60 seconds of data for LV2V @@ -388,7 +397,19 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req } pixelsPerSec := float64(info.Height) * float64(info.Width) * float64(info.FPS) pixels = int64(pixelsPerSec * billableSecs) // pixels to charge for - } else if req.Type != "" { + case RemoteType_ByocRequest: + // BYOC batch request: caller computes and supplies the compute budget as + // InPixels. Under BYOC pricing, PixelsPerUnit denominates wei-per-second, + // so InPixels here represents "seconds of compute to pay for." + if req.InPixels <= 0 { + err = errors.New("byoc-request requires inPixels") + respondJsonError(ctx, w, err, http.StatusBadRequest) + return + } + // pixels already initialised from req.InPixels above. + case "": + // Generic pre-computed path: caller must supply InPixels explicitly. + default: err = errors.New("invalid job type") respondJsonError(ctx, w, err, http.StatusBadRequest) return diff --git a/server/remote_signer_test.go b/server/remote_signer_test.go index 12aed55461..c4bc4c15fa 100644 --- a/server/remote_signer_test.go +++ b/server/remote_signer_test.go @@ -233,6 +233,17 @@ func TestGenerateLivePayment_RequestValidationErrors(t *testing.T) { wantStatus: http.StatusBadRequest, wantMsg: "missing pixels or job type", }, + { + name: "byoc-request without inPixels", + req: func() RemotePaymentRequest { + r := baseReq() + r.Type = RemoteType_ByocRequest + r.InPixels = 0 + return r + }(), + wantStatus: http.StatusBadRequest, + wantMsg: "byoc-request requires inPixels", + }, { name: "num tickets exceeds limit", req: func() RemotePaymentRequest {