Make configurable

Author: mjh1

cmd/livepeer/starter/flags.go

@@ -2,6 +2,7 @@ package starter
 
 import (
 	"flag"
+	"time"
 )
 
 func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
@@ -76,6 +77,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
 	cfg.LivePaymentInterval = fs.Duration("livePaymentInterval", *cfg.LivePaymentInterval, "Interval to pay process Gateway <> Orchestrator Payments for Live AI Video")
 	cfg.LiveOutSegmentTimeout = fs.Duration("liveOutSegmentTimeout", *cfg.LiveOutSegmentTimeout, "Timeout duration to wait the output segment to be available in the Live AI pipeline; defaults to no timeout")
 	cfg.LiveAISaveNSegments = fs.Int("liveAISaveNSegments", 10, "Set how many segments to save to disk for debugging (both input and output)")
+	cfg.LiveAICapReportInterval = fs.Duration("liveAICapReportInterval", 25*time.Minute, "Interval to report Live AI container capacity metrics")
 
 	// Onchain:
 	cfg.EthAcctAddr = fs.String("ethAcctAddr", *cfg.EthAcctAddr, "Existing Eth account address. For use when multiple ETH accounts exist in the keystore directory")

cmd/livepeer/starter/starter.go

@@ -184,7 +184,7 @@ type LivepeerConfig struct {
 	LiveAIHeartbeatInterval    *time.Duration
 	LivePaymentInterval        *time.Duration
 	LiveOutSegmentTimeout      *time.Duration
-	LiveAICapRefreshModels     *string
+	LiveAICapReportInterval    *time.Duration
 	LiveAISaveNSegments        *int
 }
 
@@ -1591,7 +1591,7 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 		if *cfg.Network != "offchain" {
 			ctx, cancel := context.WithCancel(ctx)
 			defer cancel()
-			dbOrchPoolCache, err := discovery.NewDBOrchestratorPoolCache(ctx, n, timeWatcher, orchBlacklist, *cfg.DiscoveryTimeout)
+			dbOrchPoolCache, err := discovery.NewDBOrchestratorPoolCache(ctx, n, timeWatcher, orchBlacklist, *cfg.DiscoveryTimeout, cfg.LiveAICapReportInterval)
 			if err != nil {
 				exit("Could not create orchestrator pool with DB cache: %v", err)
 			}

discovery/db_discovery.go

@@ -22,13 +22,8 @@ import (
 	"github.com/golang/glog"
 )
 
-var cacheRefreshInterval = 10 * time.Second
 var networkCapabilitiesReportingInterval = 25 * time.Minute
 
-var getTicker = func() *time.Ticker {
-	return time.NewTicker(cacheRefreshInterval)
-}
-
 type ticketParamsValidator interface {
 	ValidateTicketParams(ticketParams *pm.TicketParams) error
 }
@@ -45,7 +40,7 @@ type DBOrchestratorPoolCache struct {
 	lastNetworkCapabilitiesReported time.Time
 }
 
-func NewDBOrchestratorPoolCache(ctx context.Context, node *core.LivepeerNode, rm common.RoundsManager, orchBlacklist []string, discoveryTimeout time.Duration) (*DBOrchestratorPoolCache, error) {
+func NewDBOrchestratorPoolCache(ctx context.Context, node *core.LivepeerNode, rm common.RoundsManager, orchBlacklist []string, discoveryTimeout time.Duration, liveAICapReportInterval time.Duration) (*DBOrchestratorPoolCache, error) {
 	if node.Eth == nil {
 		return nil, fmt.Errorf("could not create DBOrchestratorPoolCache: LivepeerEthClient is nil")
 	}
@@ -70,7 +65,7 @@ func NewDBOrchestratorPoolCache(ctx context.Context, node *core.LivepeerNode, rm
 			return err
 		}
 
-		if err := dbo.pollOrchestratorInfo(ctx); err != nil {
+		if err := dbo.pollOrchestratorInfo(ctx, liveAICapReportInterval); err != nil {
 			return err
 		}
 		return nil
@@ -256,13 +251,13 @@ func (dbo *DBOrchestratorPoolCache) cacheOrchestratorStake() error {
 	return nil
 }
 
-func (dbo *DBOrchestratorPoolCache) pollOrchestratorInfo(ctx context.Context) error {
+func (dbo *DBOrchestratorPoolCache) pollOrchestratorInfo(ctx context.Context, liveAICapReportInterval time.Duration) error {
 	if err := dbo.cacheOrchInfos(); err != nil {
 		glog.Errorf("unable to poll orchestrator info: %v", err)
 		return err
 	}
 
-	ticker := getTicker()
+	ticker := time.NewTicker(liveAICapReportInterval)
 	go func() {
 		for {
 			select {
@@ -432,9 +427,6 @@ func reportAICapacityFromNetworkCapabilities(orchNetworkCapabilities []*common.O
 				InUse: int(model.CapacityInUse),
 			}
 			modelCapacities[modelID].Orchestrators[orchCap.OrchURI] = capacity
-
-			glog.Infof("HELLOO AI container %s %s inUse:%d idle:%d",
-				modelID, orchCap.OrchURI, capacity.InUse, capacity.Idle)
 		}
 	}
 

discovery/discovery.go

@@ -246,7 +246,7 @@ func (o *orchestratorPool) GetOrchestrators(ctx context.Context, numOrchestrator
 	for _, i := range rand.Perm(numAvailableOrchs) {
 		go getOrchInfo(ctx, common.OrchestratorDescriptor{linfos[i], nil}, 0, odCh, errCh, allOrchDescrCh)
 	}
-	// TODO revert the changes from https://github.com/livepeer/go-livepeer/commit/905c3144707f1b682667828f350006c5a2e0a3a8#diff-f032101b11930379f2dd15d9a82631fa0a38d87f16760d7bc5423d25c56d3b4f
+	// TODO revert the changes from https://github.com/livepeer/go-livepeer/commit/905c3144707f1b682667828f350006c5a2e0a3a8#diff-f032101b11930379f2dd15d9a82631fa0a38d87f16760d7bc5423d25c56d3b4f ?
 	// go reportLiveAICapacity(allOrchDescrCh, caps)
 
 	// use a timer to time out the entire get info loop below