Merge pull request #385 from hermeznetwork/feature/integratepriceupdater

Update and integrate price updater

cli/node/cfg.buidler.toml

@@ -4,6 +4,11 @@ Explorer = true
 UpdateMetricsInterval = "10s"
 UpdateRecommendedFeeInterval = "10s"
 
+[PriceUpdater]
+Interval = "10s"
+URL = "https://api-pub.bitfinex.com/v2/"
+Type = "bitfinexV2"
+
 [Debug]
 APIAddress = "localhost:12345"
 

config/config.go

@@ -96,6 +96,11 @@ type Coordinator struct {
 
 // Node is the hermez node configuration.
 type Node struct {
+	PriceUpdater struct {
+		Interval Duration `valudate:"required"`
+		URL      string   `valudate:"required"`
+		Type     string   `valudate:"required"`
+	} `validate:"required"`
 	StateDB struct {
 		Path string
 	} `validate:"required"`

node/node.go

@@ -23,6 +23,7 @@ import (
 	"github.com/hermeznetwork/hermez-node/db/statedb"
 	"github.com/hermeznetwork/hermez-node/eth"
 	"github.com/hermeznetwork/hermez-node/log"
+	"github.com/hermeznetwork/hermez-node/priceupdater"
 	"github.com/hermeznetwork/hermez-node/prover"
 	"github.com/hermeznetwork/hermez-node/synchronizer"
 	"github.com/hermeznetwork/hermez-node/test/debugapi"
@@ -48,8 +49,9 @@ const (
 
 // Node is the Hermez Node
 type Node struct {
-	nodeAPI  *NodeAPI
-	debugAPI *debugapi.DebugAPI
+	nodeAPI      *NodeAPI
+	debugAPI     *debugapi.DebugAPI
+	priceUpdater *priceupdater.PriceUpdater
 	// Coordinator
 	coord *coordinator.Coordinator
 
@@ -242,17 +244,23 @@ func NewNode(mode Mode, cfg *config.Node) (*Node, error) {
 	if cfg.Debug.APIAddress != "" {
 		debugAPI = debugapi.NewDebugAPI(cfg.Debug.APIAddress, stateDB, sync)
 	}
+	priceUpdater, err := priceupdater.NewPriceUpdater(cfg.PriceUpdater.URL,
+		priceupdater.APIType(cfg.PriceUpdater.Type), historyDB)
+	if err != nil {
+		return nil, tracerr.Wrap(err)
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 	return &Node{
-		nodeAPI:  nodeAPI,
-		debugAPI: debugAPI,
-		coord:    coord,
-		sync:     sync,
-		cfg:      cfg,
-		mode:     mode,
-		sqlConn:  db,
-		ctx:      ctx,
-		cancel:   cancel,
+		nodeAPI:      nodeAPI,
+		debugAPI:     debugAPI,
+		priceUpdater: priceUpdater,
+		coord:        coord,
+		sync:         sync,
+		cfg:          cfg,
+		mode:         mode,
+		sqlConn:      db,
+		ctx:          ctx,
+		cancel:       cancel,
 	}, nil
 }
 
@@ -442,7 +450,23 @@ func (n *Node) StartSynchronizer() {
 			}
 		}
 	}()
-	// TODO: Run price updater.  This is required by the API and the TxSelector
+
+	n.wg.Add(1)
+	go func() {
+		for {
+			select {
+			case <-n.ctx.Done():
+				log.Info("PriceUpdater done")
+				n.wg.Done()
+				return
+			case <-time.After(n.cfg.PriceUpdater.Interval.Duration):
+				if err := n.priceUpdater.UpdateTokenList(); err != nil {
+					log.Errorw("PriceUpdater.UpdateTokenList()", "err", err)
+				}
+				n.priceUpdater.UpdatePrices(n.ctx)
+			}
+		}
+	}()
 }
 
 // StartDebugAPI starts the DebugAPI

priceupdater/priceupdater.go

@@ -1,8 +1,9 @@
 package priceupdater
 
 import (
+	"context"
+	"fmt"
 	"net/http"
-	"strconv"
 	"time"
 
 	"github.com/dghubble/sling"
@@ -13,52 +14,92 @@ import (
 
 const (
 	defaultMaxIdleConns    = 10
-	defaultIdleConnTimeout = 10
+	defaultIdleConnTimeout = 2 * time.Second
 )
 
+// APIType defines the token exchange API
+type APIType string
+
+const (
+	// APITypeBitFinexV2 is the http API used by bitfinex V2
+	APITypeBitFinexV2 APIType = "bitfinexV2"
+)
+
+func (t *APIType) valid() bool {
+	switch *t {
+	case APITypeBitFinexV2:
+		return true
+	default:
+		return false
+	}
+}
+
 // PriceUpdater definition
 type PriceUpdater struct {
 	db           *historydb.HistoryDB
 	apiURL       string
+	apiType      APIType
 	tokenSymbols []string
 }
 
 // NewPriceUpdater is the constructor for the updater
-func NewPriceUpdater(apiURL string, db *historydb.HistoryDB) PriceUpdater {
+func NewPriceUpdater(apiURL string, apiType APIType, db *historydb.HistoryDB) (*PriceUpdater, error) {
 	tokenSymbols := []string{}
-	return PriceUpdater{
+	if !apiType.valid() {
+		return nil, tracerr.Wrap(fmt.Errorf("Invalid apiType: %v", apiType))
+	}
+	return &PriceUpdater{
 		db:           db,
 		apiURL:       apiURL,
+		apiType:      apiType,
 		tokenSymbols: tokenSymbols,
+	}, nil
+}
+
+func getTokenPriceBitfinex(ctx context.Context, client *sling.Sling,
+	tokenSymbol string) (float64, error) {
+	state := [10]float64{}
+	req, err := client.New().Get("ticker/t" + tokenSymbol + "USD").Request()
+	if err != nil {
+		return 0, tracerr.Wrap(err)
+	}
+	res, err := client.Do(req.WithContext(ctx), &state, nil)
+	if err != nil {
+		return 0, tracerr.Wrap(err)
+	}
+	if res.StatusCode != http.StatusOK {
+		return 0, fmt.Errorf("http response is not is %v", res.StatusCode)
 	}
+	return state[6], nil
 }
 
 // UpdatePrices is triggered by the Coordinator, and internally will update the token prices in the db
-func (p *PriceUpdater) UpdatePrices() {
+func (p *PriceUpdater) UpdatePrices(ctx context.Context) {
 	tr := &http.Transport{
 		MaxIdleConns:       defaultMaxIdleConns,
-		IdleConnTimeout:    defaultIdleConnTimeout * time.Second,
+		IdleConnTimeout:    defaultIdleConnTimeout,
 		DisableCompression: true,
 	}
 	httpClient := &http.Client{Transport: tr}
 	client := sling.New().Base(p.apiURL).Client(httpClient)
 
-	state := [10]float64{}
-
 	for _, tokenSymbol := range p.tokenSymbols {
-		resp, err := client.New().Get("ticker/t" + tokenSymbol + "USD").ReceiveSuccess(&state)
-		errString := tokenSymbol + " not updated, error: "
-		if err != nil {
-			log.Error(errString + err.Error())
-			continue
+		var tokenPrice float64
+		var err error
+		switch p.apiType {
+		case APITypeBitFinexV2:
+			tokenPrice, err = getTokenPriceBitfinex(ctx, client, tokenSymbol)
 		}
-		if resp.StatusCode != http.StatusOK {
-			log.Error(errString + "response is not 200, is " + strconv.Itoa(resp.StatusCode))
-			continue
+		if ctx.Err() != nil {
+			return
 		}
-		err = p.db.UpdateTokenValue(tokenSymbol, state[6])
 		if err != nil {
-			log.Error(errString + err.Error())
+			log.Errorw("token price not updated (get error)",
+				"err", err, "token", tokenSymbol, "apiType", p.apiType)
+		}
+		if err = p.db.UpdateTokenValue(tokenSymbol, tokenPrice); err != nil {
+			log.Errorw("token price not updated (db error)",
+				"err", err, "token", tokenSymbol, "apiType", p.apiType)
 		}
 	}
 }

priceupdater/priceupdater_test.go

@@ -1,6 +1,7 @@
 package priceupdater
 
 import (
+	"context"
 	"math/big"
 	"os"
 	"testing"
@@ -11,6 +12,7 @@ import (
 	"github.com/hermeznetwork/hermez-node/db/historydb"
 	"github.com/hermeznetwork/hermez-node/test"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestPriceUpdater(t *testing.T) {
@@ -37,15 +39,16 @@ func TestPriceUpdater(t *testing.T) {
 	})
 	assert.NoError(t, historyDB.AddTokens(tokens))
 	// Init price updater
-	pu := NewPriceUpdater("https://api-pub.bitfinex.com/v2/", historyDB)
+	pu, err := NewPriceUpdater("https://api-pub.bitfinex.com/v2/", APITypeBitFinexV2, historyDB)
+	require.NoError(t, err)
 	// Update token list
 	assert.NoError(t, pu.UpdateTokenList())
 	// Update prices
-	pu.UpdatePrices()
+	pu.UpdatePrices(context.Background())
 	// Check that prices have been updated
 	limit := uint(10)
 	fetchedTokens, _, err := historyDB.GetTokens(nil, nil, "", nil, &limit, historydb.OrderAsc)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	// TokenID 0 (ETH) is always on the DB
 	assert.Equal(t, 2, len(fetchedTokens))
 	for _, token := range fetchedTokens {