feat(ai): add hardware info from Orchestrators and expand network information available.

common/db.go

@@ -51,6 +51,7 @@
 	ActivationRound   int64
 	DeactivationRound int64
 	Stake             int64 // Stored as a fixed point number
+	RemoteInfo        string
 }
 
 // DBOrch is the type binding for a row result from the unbondingLocks table
@@ -69,7 +70,7 @@
 	UpdatedLastDay bool
 }
 
-var LivepeerDBVersion = 1
+var LivepeerDBVersion = 2
 
 var ErrDBTooNew = errors.New("DB Too New")
 
@@ -89,7 +90,8 @@
 		pricePerPixel int64,
 		activationRound int64,
 		deactivationRound int64,
-		stake int64
+		stake int64,
+		remoteInfo STRING DEFAULT '{}'
 	);
 
 	CREATE TABLE IF NOT EXISTS unbondingLocks (
@@ -133,14 +135,15 @@
 	CREATE INDEX IF NOT EXISTS idx_blockheaders_number ON blockheaders(number);
 `
 
-func NewDBOrch(ethereumAddr string, serviceURI string, pricePerPixel int64, activationRound int64, deactivationRound int64, stake int64) *DBOrch {
+func NewDBOrch(ethereumAddr string, serviceURI string, pricePerPixel int64, activationRound int64, deactivationRound int64, stake int64, remoteInfo string) *DBOrch {
 	return &DBOrch{
 		ServiceURI:        serviceURI,
 		EthereumAddr:      ethereumAddr,
 		PricePerPixel:     pricePerPixel,
 		ActivationRound:   activationRound,
 		DeactivationRound: deactivationRound,
 		Stake:             stake,
+		RemoteInfo:        remoteInfo,
 	}
 }
 
@@ -184,6 +187,23 @@
 	} else if dbVersion < LivepeerDBVersion {
 		// Upgrade stepwise up to the correct version using the migration
 		// procedure for each version
+		if dbVersion == 1 && LivepeerDBVersion == 2 {
+			glog.Info("Upgrading DB from version 1 to 2")
+			// Add remoteInfo column to orchestrators table
+			_, err = db.Exec("ALTER TABLE orchestrators ADD COLUMN remoteInfo STRING default '{}'")
+			if err != nil {
+				glog.Error("Error adding remoteInfo column to orchestrators table ", err)
+				d.Close()
+				return nil, err
+			}
+			//update the db version
+			_, err = db.Exec("INSERT OR REPLACE INTO kv(key, value) VALUES('dbVersion', '2')")
+			if err != nil {
+				glog.Error("Error updating DB version to 2 ", err)
+				d.Close()
+				return nil, err
+			}
+		}
 	} else if dbVersion == LivepeerDBVersion {
 		// all good; nothing to do
 	}
@@ -208,8 +228,8 @@
 
 	// updateOrch prepared statement
 	stmt, err = db.Prepare(`
-	INSERT INTO orchestrators(updatedAt, ethereumAddr, serviceURI, pricePerPixel, activationRound, deactivationRound, stake, createdAt)
-	VALUES(datetime(), :ethereumAddr, :serviceURI, :pricePerPixel, :activationRound, :deactivationRound, :stake, datetime())
+	INSERT INTO orchestrators(updatedAt, ethereumAddr, serviceURI, pricePerPixel, activationRound, deactivationRound, stake, remoteInfo, createdAt)
+	VALUES(datetime(), :ethereumAddr, :serviceURI, :pricePerPixel, :activationRound, :deactivationRound, :stake, :remoteInfo, datetime())
 	ON CONFLICT(ethereumAddr) DO UPDATE SET
 	updatedAt = excluded.updatedAt,
 	serviceURI =
@@ -231,7 +251,11 @@
 	stake =
 		CASE WHEN excluded.stake == 0
 		THEN orchestrators.stake
-		ELSE excluded.stake END
+		ELSE excluded.stake END,
+	remoteInfo =
+		CASE WHEN trim(excluded.remoteInfo) == ""
+		THEN orchestrators.remoteInfo
+		ELSE trim(excluded.remoteInfo) END
 	`)
 	if err != nil {
 		glog.Error("Unable to prepare updateOrch ", err)
@@ -454,9 +478,9 @@
 }
 
 func (db *DB) SetChainID(id *big.Int) error {
 	if err := db.updateKVStore("chainID", id.String()); err != nil {
 		return err
 	}
 	return nil
 }
 
@@ -493,6 +517,7 @@
 		sql.Named("activationRound", orch.ActivationRound),
 		sql.Named("deactivationRound", orch.DeactivationRound),
 		sql.Named("stake", orch.Stake),
+		sql.Named("remoteInfo", orch.RemoteInfo),
 	)
 
 	if err != nil {
@@ -522,13 +547,14 @@
 			activationRound   int64
 			deactivationRound int64
 			stake             int64
+			remoteInfo        string
 		)
-		if err := rows.Scan(&serviceURI, &ethereumAddr, &pricePerPixel, &activationRound, &deactivationRound, &stake); err != nil {
+		if err := rows.Scan(&serviceURI, &ethereumAddr, &pricePerPixel, &activationRound, &deactivationRound, &stake, &remoteInfo); err != nil {
 			glog.Error("db: Unable to fetch orchestrator ", err)
 			continue
 		}
 
-		orchs = append(orchs, NewDBOrch(serviceURI, ethereumAddr, pricePerPixel, activationRound, deactivationRound, stake))
+		orchs = append(orchs, NewDBOrch(serviceURI, ethereumAddr, pricePerPixel, activationRound, deactivationRound, stake, remoteInfo))
 	}
 	return orchs, nil
 }
@@ -807,7 +833,7 @@
 }
 
 func buildSelectOrchsQuery(filter *DBOrchFilter) (string, error) {
-	query := "SELECT ethereumAddr, serviceURI, pricePerPixel, activationRound, deactivationRound, stake FROM orchestrators "
+	query := "SELECT ethereumAddr, serviceURI, pricePerPixel, activationRound, deactivationRound, stake, remoteInfo FROM orchestrators "
 	fil, err := buildFilterOrchsQuery(filter)
 	if err != nil {
 		return "", err

common/db_test.go

@@ -2,6 +2,7 @@ package common
 
 import (
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"math"
 	"math/big"
@@ -242,7 +243,7 @@ func TestSelectUpdateOrchs_AddingUpdatingRow_NoError(t *testing.T) {
 	assert.Equal(orchs[0].Stake, int64(0))
 
 	// updating row with same orchAddress
-	orchUpdate := NewDBOrch(orchAddress, "127.0.0.1:8937", 1000, 5, 10, 50)
+	orchUpdate := NewDBOrch(orchAddress, "127.0.0.1:8937", 1000, 5, 10, 50, "")
 	err = dbh.UpdateOrch(orchUpdate)
 	require.Nil(err)
 
@@ -334,6 +335,30 @@ func TestSelectUpdateOrchs_AddingUpdatingRow_NoError(t *testing.T) {
 	assert.Equal(updatedOrch[0].DeactivationRound, deactivationRoundUpdate.DeactivationRound)
 	assert.Equal(updatedOrch[0].PricePerPixel, priceUpdate.PricePerPixel)
 	assert.Equal(updatedOrch[0].Stake, stakeUpdate.Stake)
+
+	// Updating only remoteInfo
+	remoteInfo := make(map[string]interface{})
+	remoteInfo["transcoder"] = "http://transcoder.uri:5555"
+	riJson, _ := json.Marshal(remoteInfo)
+	remoteInfoUpdate := &DBOrch{
+		EthereumAddr: orchAddress,
+		RemoteInfo:   string(riJson),
+	}
+
+	assert.Equal(updatedOrch[0].RemoteInfo, "")
+
+	err = dbh.UpdateOrch(remoteInfoUpdate)
+	require.Nil(err)
+
+	updatedOrch, err = dbh.SelectOrchs(nil)
+	assert.Len(updatedOrch, 1)
+	assert.NoError(err)
+	assert.Equal(updatedOrch[0].ServiceURI, serviceURIUpdate.ServiceURI)
+	assert.Equal(updatedOrch[0].ActivationRound, activationRoundUpdate.ActivationRound)
+	assert.Equal(updatedOrch[0].DeactivationRound, deactivationRoundUpdate.DeactivationRound)
+	assert.Equal(updatedOrch[0].PricePerPixel, priceUpdate.PricePerPixel)
+	assert.Equal(updatedOrch[0].Stake, stakeUpdate.Stake)
+	assert.Equal(updatedOrch[0].RemoteInfo, remoteInfoUpdate.RemoteInfo)
 }
 
 func TestSelectUpdateOrchs_AddingMultipleRows_NoError(t *testing.T) {
@@ -347,7 +372,7 @@ func TestSelectUpdateOrchs_AddingMultipleRows_NoError(t *testing.T) {
 	// adding one row
 	orchAddress := pm.RandAddress().String()
 
-	orch := NewDBOrch(orchAddress, "127.0.0.1:8936", 1, 0, 0, 0)
+	orch := NewDBOrch(orchAddress, "127.0.0.1:8936", 1, 0, 0, 0, "")
 	err = dbh.UpdateOrch(orch)
 	require.Nil(err)
 
@@ -359,7 +384,7 @@ func TestSelectUpdateOrchs_AddingMultipleRows_NoError(t *testing.T) {
 	// adding second row
 	orchAddress = pm.RandAddress().String()
 
-	orchAdd := NewDBOrch(orchAddress, "127.0.0.1:8938", 1, 0, 0, 0)
+	orchAdd := NewDBOrch(orchAddress, "127.0.0.1:8938", 1, 0, 0, 0, "")
 	err = dbh.UpdateOrch(orchAdd)
 	require.Nil(err)
 
@@ -384,7 +409,7 @@ func TestOrchCount(t *testing.T) {
 	require.Nil(err)
 
 	for i := 0; i < 10; i++ {
-		orch := NewDBOrch("https://127.0.0.1:"+strconv.Itoa(8936+i), pm.RandAddress().String(), 1, int64(i), int64(5+i), 0)
+		orch := NewDBOrch("https://127.0.0.1:"+strconv.Itoa(8936+i), pm.RandAddress().String(), 1, int64(i), int64(5+i), 0, "")
 		orch.PricePerPixel, err = PriceToFixed(big.NewRat(1, int64(5+i)))
 		require.Nil(err)
 		err = dbh.UpdateOrch(orch)
@@ -440,7 +465,7 @@ func TestDBFilterOrchs(t *testing.T) {
 	require.Nil(err)
 
 	for i := 0; i < 10; i++ {
-		orch := NewDBOrch(pm.RandAddress().String(), "https://127.0.0.1:"+strconv.Itoa(8936+i), 1, int64(i), int64(5+i), 0)
+		orch := NewDBOrch(pm.RandAddress().String(), "https://127.0.0.1:"+strconv.Itoa(8936+i), 1, int64(i), int64(5+i), 0, "")
 		orch.PricePerPixel, err = PriceToFixed(big.NewRat(1, int64(5+i)))
 		require.Nil(err)
 		err = dbh.UpdateOrch(orch)
@@ -578,7 +603,7 @@ func TestIsOrchActive(t *testing.T) {
 	assert.EqualError(err, "Orchestrator not found")
 
 	activationRound := 5
-	orch := NewDBOrch(addr, "https://127.0.0.1:8936", 1, int64(activationRound), int64(activationRound+2), 0)
+	orch := NewDBOrch(addr, "https://127.0.0.1:8936", 1, int64(activationRound), int64(activationRound+2), 0, "")
 	dbh.UpdateOrch(orch)
 
 	// inactive in round 4

core/ai.go

@@ -31,6 +31,7 @@ type AI interface {
 	Warm(context.Context, string, string, worker.RunnerEndpoint, worker.OptimizationFlags) error
 	Stop(context.Context) error
 	HasCapacity(pipeline, modelID string) bool
+	HardwareInformation() []worker.HardwareInformation
 }
 
 // Custom type to parse a big.Rat from a JSON number.