Enable Orchestrator to set pricing by broadcaster ETH address

CHANGELOG_PENDING.md

@@ -28,5 +28,6 @@
 
 #### Orchestrator
 - [#2591](https://github.com/livepeer/go-livepeer/pull/2591) Return from transcode loop if transcode session is ended by B (@yondonfu)
+- [#2592](https://github.com/livepeer/go-livepeer/pull/2592) Enable Orchestrator to set pricing by broadcaster ETH address
 
-#### Transcoder
+#### Transcoder

cmd/livepeer/livepeer.go

@@ -146,6 +146,7 @@ func parseLivepeerConfig() starter.LivepeerConfig {
 	// Unit of pixels for both O's basePriceInfo and B's MaxBroadcastPrice
 	cfg.PixelsPerUnit = flag.Int("pixelsPerUnit", *cfg.PixelsPerUnit, "Amount of pixels per unit. Set to '> 1' to have smaller price granularity than 1 wei / pixel")
 	cfg.AutoAdjustPrice = flag.Bool("autoAdjustPrice", *cfg.AutoAdjustPrice, "Enable/disable automatic price adjustments based on the overhead for redeeming tickets")
+	cfg.PricePerBroadcaster = flag.String("pricePerBroadcaster", *cfg.PricePerBroadcaster, `json list of price per broadcaster. Example: {"broadcasters":[{"ethaddress":"address1","priceperunit":1000,"pixelsperunit":1},{"ethaddress":"address2","priceperunit":1200,"pixelsperunit":1}]}`)
 	// Interval to poll for blocks
 	cfg.BlockPollingInterval = flag.Int("blockPollingInterval", *cfg.BlockPollingInterval, "Interval in seconds at which different blockchain event services poll for blocks")
 	// Redemption service

cmd/livepeer/starter/starter.go

@@ -2,6 +2,7 @@ package starter
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -105,6 +106,7 @@ type LivepeerConfig struct {
 	MaxPricePerUnit              *int
 	PixelsPerUnit                *int
 	AutoAdjustPrice              *bool
+	PricePerBroadcaster          *string
 	BlockPollingInterval         *int
 	Redeemer                     *bool
 	RedeemerAddr                 *string
@@ -172,6 +174,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 	defaultMaxPricePerUnit := 0
 	defaultPixelsPerUnit := 1
 	defaultAutoAdjustPrice := true
+	defaultpricePerBroadcaster := ""
 	defaultBlockPollingInterval := 5
 	defaultRedeemer := false
 	defaultRedeemerAddr := ""
@@ -242,6 +245,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 		MaxPricePerUnit:        &defaultMaxPricePerUnit,
 		PixelsPerUnit:          &defaultPixelsPerUnit,
 		AutoAdjustPrice:        &defaultAutoAdjustPrice,
+		PricePerBroadcaster:    &defaultpricePerBroadcaster,
 		BlockPollingInterval:   &defaultBlockPollingInterval,
 		Redeemer:               &defaultRedeemer,
 		RedeemerAddr:           &defaultRedeemerAddr,
@@ -372,7 +376,7 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	}
 
 	if *cfg.OrchSecret != "" {
-		n.OrchSecret, _ = common.GetPass(*cfg.OrchSecret)
+		n.OrchSecret, _ = common.ReadFromFile(*cfg.OrchSecret)
 	}
 
 	var transcoderCaps []core.Capability
@@ -694,9 +698,18 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 			if *cfg.PricePerUnit < 0 {
 				panic(fmt.Errorf("-pricePerUnit must be >= 0, provided %d", *cfg.PricePerUnit))
 			}
-			n.SetBasePrice(big.NewRat(int64(*cfg.PricePerUnit), int64(*cfg.PixelsPerUnit)))
+			n.SetBasePrice("default", big.NewRat(int64(*cfg.PricePerUnit), int64(*cfg.PixelsPerUnit)))
 			glog.Infof("Price: %d wei for %d pixels\n ", *cfg.PricePerUnit, *cfg.PixelsPerUnit)
 
+			if *cfg.PricePerBroadcaster != "" {
+				ppb := getBroadcasterPrices(*cfg.PricePerBroadcaster)
+				for _, p := range ppb {
+					price := big.NewRat(p.PricePerUnit, p.PixelsPerUnit)
+					n.SetBasePrice(p.EthAddress, price)
+					glog.Infof("Price: %v set for broadcaster %v", price.RatString(), p.EthAddress)
+				}
+			}
+
 			n.AutoAdjustPrice = *cfg.AutoAdjustPrice
 
 			ev, _ := new(big.Int).SetString(*cfg.TicketEV, 10)
@@ -1351,3 +1364,29 @@ func checkOrStoreChainID(dbh *common.DB, chainID *big.Int) error {
 
 	return nil
 }
+
+// Format of broadcasterPrices json
+// {"broadcasters":[{"ethaddress":"address1","priceperunit":1000,"pixelsperunit":1}, {"ethaddress":"address2","priceperunit":2000,"pixelsperunit":3}]}
+type BroadcasterPrices struct {
+	Prices []BroadcasterPrice `json:"broadcasters"`
+}
+
+type BroadcasterPrice struct {
+	EthAddress    string `json:"ethaddress"`
+	PricePerUnit  int64  `json:"priceperunit"`
+	PixelsPerUnit int64  `json:"pixelsperunit"`
+}
+
+func getBroadcasterPrices(broadcasterPrices string) []BroadcasterPrice {
+	var pricesSet BroadcasterPrices
+	prices, _ := common.ReadFromFile(broadcasterPrices)
+
+	err := json.Unmarshal([]byte(prices), &pricesSet)
+
+	if err != nil {
+		glog.Errorf("broadcaster prices could not be parsed")
+		return nil
+	}
+
+	return pricesSet.Prices
+}

cmd/livepeer/starter/starter_test.go

@@ -84,3 +84,18 @@ func TestIsLocalURL(t *testing.T) {
 	assert.Nil(err)
 	assert.False(isLocal)
 }
+
+func TestParseGetBroadcasterPrices(t *testing.T) {
+	assert := assert.New(t)
+
+	j := `{"broadcasters":[{"ethaddress":"0x0000000000000000000000000000000000000000","priceperunit":1000,"pixelsperunit":1}, {"ethaddress":"0x1000000000000000000000000000000000000000","priceperunit":2000,"pixelsperunit":3}]}`
+
+	prices := getBroadcasterPrices(j)
+	assert.NotNil(prices)
+	assert.Equal(2, len(prices))
+
+	price1 := big.NewRat(prices[0].PricePerUnit, prices[0].PixelsPerUnit)
+	price2 := big.NewRat(prices[1].PricePerUnit, prices[1].PixelsPerUnit)
+	assert.Equal(big.NewRat(1000, 1), price1)
+	assert.Equal(big.NewRat(2000, 3), price2)
+}

cmd/livepeer_cli/livepeer_cli.go

@@ -113,6 +113,7 @@ func (w *wizard) initializeOptions() []wizardOpt {
 		{desc: "Sign typed data", invoke: w.signTypedData},
 		{desc: "Vote in a poll", invoke: w.vote, orchestrator: true},
 		{desc: "Set max ticket face value", invoke: w.setMaxFaceValue, orchestrator: true},
+		{desc: "Set price for broadcaster", invoke: w.setPriceForBroadcaster, orchestrator: true},
 		{desc: "Exit", invoke: func() {
 			fmt.Println("Goodbye, my friend")
 			os.Exit(0)

cmd/livepeer_cli/wizard_stats.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -209,6 +210,11 @@ func (w *wizard) orchestratorStats() {
 		return
 	}
 
+	b_prices, err := w.getBroadcasterPrices()
+	if err != nil {
+		glog.Errorf("Error getting broadcaster prices: %v", err)
+	}
+
 	fmt.Println("+------------------+")
 	fmt.Println("|ORCHESTRATOR STATS|")
 	fmt.Println("+------------------+")
@@ -223,6 +229,7 @@ func (w *wizard) orchestratorStats() {
 		{"Fee Cut (%)", eth.FormatPerc(flipPerc(t.FeeShare))},
 		{"Last Reward Round", t.LastRewardRound.String()},
 		{"Base price per pixel", fmt.Sprintf("%v wei / %v pixels", priceInfo.Num(), priceInfo.Denom())},
+		{"Base price for broadcasters", b_prices},
 	}
 
 	for _, v := range data {
@@ -468,3 +475,32 @@ func (w *wizard) currentBlock() (*big.Int, error) {
 
 	return new(big.Int).SetBytes(body), nil
 }
+
+func (w *wizard) getBroadcasterPrices() (string, error) {
+	resp, err := http.Get(fmt.Sprintf("http://%v:%v/status", w.host, w.httpPort))
+
+	if err != nil {
+		return "", err
+	}
+
+	defer resp.Body.Close()
+	result, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+
+	var status map[string]interface{}
+	err = json.Unmarshal(result, &status)
+	if err != nil {
+		return "", err
+	}
+
+	prices := new(bytes.Buffer)
+	for b, p := range status["BroadcasterPrices"].(map[string]interface{}) {
+		if b != "default" {
+			fmt.Fprintf(prices, "%s: %s per pixel\n", b, p)
+		}
+	}
+
+	return prices.String(), nil
+}

cmd/livepeer_cli/wizard_transcoder.go

@@ -305,3 +305,36 @@ func (w *wizard) setMaxFaceValue() {
 		return
 	}
 }
+
+func (w *wizard) setPriceForBroadcaster() {
+	fmt.Println("Enter the ETH address of the broadcaster")
+	ethaddr := w.readStringAndValidate(func(in string) (string, error) {
+		if "" == in {
+			return "", fmt.Errorf("no broadcaster eth address input")
+		}
+		if in[0:2] != "0x" || len(in) != 42 {
+			return "", fmt.Errorf("broadcaster eth address input not in correct format")
+		}
+
+		return in, nil
+	})
+
+	fmt.Println("Enter price per unit:")
+	price := w.readDefaultInt(0)
+	fmt.Println("Enter pixels per unit:")
+	pixels := w.readDefaultInt(1)
+	data := url.Values{
+		"pricePerUnit":       {fmt.Sprintf("%v", strconv.Itoa(price))},
+		"pixelsPerUnit":      {fmt.Sprintf("%v", strconv.Itoa(pixels))},
+		"broadcasterEthAddr": {fmt.Sprintf("%v", ethaddr)},
+	}
+	result, ok := httpPostWithParams(fmt.Sprintf("http://%v:%v/setPriceForBroadcaster", w.host, w.httpPort), data)
+	if ok {
+		fmt.Printf("Price for broadcaster %v set to %v gwei per %v pixels", ethaddr, price, pixels)
+		return
+	} else {
+		fmt.Printf("Error setting price for broadcaster: %v", result)
+		return
+	}
+
+}

common/getpass.go → common/readfromfile.go

@@ -6,10 +6,10 @@ import (
 	"os"
 )
 
-// GetPass attempts to read a file for a password at the supplied location.
+// ReadFromFile attempts to read a file at the supplied location.
 // If it fails, then the original supplied string will be returned to the caller.
 // A valid string will always be returned, regardless of whether an error occurred.
-func GetPass(s string) (string, error) {
+func ReadFromFile(s string) (string, error) {
 	info, err := os.Stat(s)
 	if os.IsNotExist(err) {
 		// If the supplied string is not a path to a file,

common/getpass_test.go → common/readfromfile_test.go

@@ -9,40 +9,40 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestGetPassNoFileExists(t *testing.T) {
+func TestReadFromFileNoFileExists(t *testing.T) {
 	assert := assert.New(t)
 
 	input := "/tmp/../../../nothing"
 	expectedOutput := "/tmp/../../../nothing"
 
-	// GetPass should return the string it was supplied
-	output, err := GetPass(input)
+	// ReadFromFile should return the string it was supplied
+	output, err := ReadFromFile(input)
 
 	assert.Nil(err)
-	// GetPass should return the originaly supplied string
+	// ReadFromFile should return the originaly supplied string
 	assert.Equal(expectedOutput, output)
 }
 
-func TestGetPassDirectoryExists(t *testing.T) {
+func TestReadFromFileDirectoryExists(t *testing.T) {
 	assert := assert.New(t)
 
 	input := "/tmp"
 	expectedOutput := "/tmp"
 
-	// GetPass should return the string it was supplied
-	output, err := GetPass(input)
+	// ReadFromFile should return the string it was supplied
+	output, err := ReadFromFile(input)
 
 	assert.NotNil(err)
-	// GetPass should return the originaly supplied string
+	// ReadFromFile should return the originaly supplied string
 	assert.Equal(expectedOutput, output)
 }
 
-func TestGetPassEmptyFileExists(t *testing.T) {
+func TestReadFromFileEmptyFileExists(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
 
-	tmpFile := "TestGetPassEmptyFileExists.txt"
-	expectedOutput := "TestGetPassEmptyFileExists.txt"
+	tmpFile := "TestReadFromFileEmptyFileExists.txt"
+	expectedOutput := "TestReadFromFileEmptyFileExists.txt"
 
 	emptyFile, err := os.Create(tmpFile)
 	emptyFile.Close()
@@ -53,22 +53,22 @@ func TestGetPassEmptyFileExists(t *testing.T) {
 		require.Nil(err)
 	}()
 
-	// GetPass should return an error
-	output, err := GetPass(tmpFile)
+	// ReadFromFile should return an error
+	output, err := ReadFromFile(tmpFile)
 
 	assert.NotNil(err)
-	// GetPass should return the originaly supplied string
+	// ReadFromFile should return the originaly supplied string
 	assert.Equal(expectedOutput, output)
 }
 
-func TestGetPassFileExistsOneLine(t *testing.T) {
+func TestReadFromFile_FileExistsOneLine(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
 
 	input := `something`
 	expectedOutput := "something"
 
-	tmpFile := "TestGetPassFileExistsOneLine.txt"
+	tmpFile := "TestReadFromFile_FileExistsOneLine.txt"
 
 	file, err := os.Create(tmpFile)
 	require.Nil(err)
@@ -81,14 +81,14 @@ func TestGetPassFileExistsOneLine(t *testing.T) {
 		require.Nil(err)
 	}()
 
-	output, err := GetPass(tmpFile)
+	output, err := ReadFromFile(tmpFile)
 
 	assert.Nil(err)
-	// GetPass should the first line of the text file
+	// ReadFromFile should the first line of the text file
 	assert.Equal(expectedOutput, output)
 }
 
-func TestGetPassFileExistsMultiline(t *testing.T) {
+func TestReadFromFile_FileExistsMultiline(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
 
@@ -97,7 +97,7 @@ func TestGetPassFileExistsMultiline(t *testing.T) {
 somethingelse`
 	expectedOutput := "something"
 
-	tmpFile := "TestGetPassFileExistsMultiline.txt"
+	tmpFile := "TestReadFromFile_FileExistsMultiline.txt"
 
 	file, err := os.Create(tmpFile)
 	require.Nil(err)
@@ -110,9 +110,9 @@ somethingelse`
 		require.Nil(err)
 	}()
 
-	output, err := GetPass(tmpFile)
+	output, err := ReadFromFile(tmpFile)
 
 	assert.Nil(err)
-	// GetPass should the first line of the text file
+	// ReadFromFile should the first line of the text file
 	assert.Equal(expectedOutput, output)
 }

common/types.go

@@ -36,6 +36,7 @@ type NodeStatus struct {
 	RegisteredTranscodersNumber int
 	RegisteredTranscoders       []RemoteTranscoderInfo
 	LocalTranscoding            bool // Indicates orchestrator that is also transcoder
+	BroadcasterPrices           map[string]*big.Rat
 	// xxx add transcoder's version here
 }