Merge branch 'master' into zcash

bchain/coins/blockchain.go

@@ -9,11 +9,12 @@ import (
 	"fmt"
 	"io/ioutil"
 	"reflect"
+	"time"
 
 	"github.com/juju/errors"
 )
 
-type blockChainFactory func(config json.RawMessage, pushHandler func(*bchain.MQMessage), metrics *common.Metrics) (bchain.BlockChain, error)
+type blockChainFactory func(config json.RawMessage, pushHandler func(*bchain.MQMessage)) (bchain.BlockChain, error)
 
 var blockChainFactories = make(map[string]blockChainFactory)
 
@@ -38,10 +39,107 @@ func NewBlockChain(coin string, configfile string, pushHandler func(*bchain.MQMe
 	if err != nil {
 		return nil, errors.Annotatef(err, "Error parsing file %v", configfile)
 	}
-	bc, err := bcf(config, pushHandler, metrics)
+	bc, err := bcf(config, pushHandler)
 	if err != nil {
 		return nil, err
 	}
-	bc.Initialize(bchain.NewMempool(bc, metrics))
+	bc.Initialize()
-	return bc, nil
+	return &blockChainWithMetrics{b: bc, m: metrics}, nil
+}
+
+type blockChainWithMetrics struct {
+	b bchain.BlockChain
+	m *common.Metrics
+}
+
+func (c *blockChainWithMetrics) observeRPCLatency(method string, start time.Time, err error) {
+	var e string
+	if err != nil {
+		e = err.Error()
+	}
+	c.m.RPCLatency.With(common.Labels{"method": method, "error": e}).Observe(float64(time.Since(start)) / 1e6) // in milliseconds
+}
+
+func (c *blockChainWithMetrics) Initialize() error {
+	return c.b.Initialize()
+}
+
+func (c *blockChainWithMetrics) Shutdown() error {
+	return c.b.Shutdown()
+}
+
+func (c *blockChainWithMetrics) IsTestnet() bool {
+	return c.b.IsTestnet()
+}
+
+func (c *blockChainWithMetrics) GetNetworkName() string {
+	return c.b.GetNetworkName()
+}
+
+func (c *blockChainWithMetrics) GetBestBlockHash() (v string, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetBestBlockHash", s, err) }(time.Now())
+	return c.b.GetBestBlockHash()
+}
+
+func (c *blockChainWithMetrics) GetBestBlockHeight() (v uint32, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetBestBlockHeight", s, err) }(time.Now())
+	return c.b.GetBestBlockHeight()
+}
+
+func (c *blockChainWithMetrics) GetBlockHash(height uint32) (v string, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetBlockHash", s, err) }(time.Now())
+	return c.b.GetBlockHash(height)
+}
+
+func (c *blockChainWithMetrics) GetBlockHeader(hash string) (v *bchain.BlockHeader, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetBlockHeader", s, err) }(time.Now())
+	return c.b.GetBlockHeader(hash)
+}
+
+func (c *blockChainWithMetrics) GetBlock(hash string, height uint32) (v *bchain.Block, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetBlock", s, err) }(time.Now())
+	return c.b.GetBlock(hash, height)
+}
+
+func (c *blockChainWithMetrics) GetMempool() (v []string, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetMempool", s, err) }(time.Now())
+	return c.b.GetMempool()
+}
+
+func (c *blockChainWithMetrics) GetTransaction(txid string) (v *bchain.Tx, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetTransaction", s, err) }(time.Now())
+	return c.b.GetTransaction(txid)
+}
+
+func (c *blockChainWithMetrics) EstimateSmartFee(blocks int, conservative bool) (v float64, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("EstimateSmartFee", s, err) }(time.Now())
+	return c.b.EstimateSmartFee(blocks, conservative)
+}
+
+func (c *blockChainWithMetrics) SendRawTransaction(tx string) (v string, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("SendRawTransaction", s, err) }(time.Now())
+	return c.b.SendRawTransaction(tx)
+}
+
+func (c *blockChainWithMetrics) ResyncMempool(onNewTxAddr func(txid string, addr string)) (err error) {
+	defer func(s time.Time) { c.observeRPCLatency("ResyncMempool", s, err) }(time.Now())
+	return c.b.ResyncMempool(onNewTxAddr)
+}
+
+func (c *blockChainWithMetrics) GetMempoolTransactions(address string) (v []string, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetMempoolTransactions", s, err) }(time.Now())
+	return c.b.GetMempoolTransactions(address)
+}
+
+func (c *blockChainWithMetrics) GetMempoolSpentOutput(outputTxid string, vout uint32) (v string) {
+	return c.b.GetMempoolSpentOutput(outputTxid, vout)
+}
+
+func (c *blockChainWithMetrics) GetMempoolEntry(txid string) (v *bchain.MempoolEntry, err error) {
+	defer func(s time.Time) { c.observeRPCLatency("GetMempoolEntry", s, err) }(time.Now())
+	return c.b.GetMempoolEntry(txid)
+}
+
+func (c *blockChainWithMetrics) GetChainParser() bchain.BlockChainParser {
+	return c.b.GetChainParser()
 }

bchain/coins/btc/bitcoinrpc.go

@@ -2,7 +2,6 @@ package btc
 
 import (
 	"blockbook/bchain"
-	"blockbook/common"
 	"bytes"
 	"encoding/hex"
 	"encoding/json"
@@ -29,7 +28,6 @@ type BitcoinRPC struct {
 	Network     string
 	Mempool     *bchain.Mempool
 	ParseBlocks bool
-	metrics     *common.Metrics
 	mq          *bchain.MQ
 }
 
@@ -43,7 +41,7 @@ type configuration struct {
 }
 
 // NewBitcoinRPC returns new BitcoinRPC instance.
-func NewBitcoinRPC(config json.RawMessage, pushHandler func(*bchain.MQMessage), metrics *common.Metrics) (bchain.BlockChain, error) {
+func NewBitcoinRPC(config json.RawMessage, pushHandler func(*bchain.MQMessage)) (bchain.BlockChain, error) {
 	var err error
 	var c configuration
 	err = json.Unmarshal(config, &c)
@@ -62,7 +60,6 @@ func NewBitcoinRPC(config json.RawMessage, pushHandler func(*bchain.MQMessage),
 		user:        c.RPCUser,
 		password:    c.RPCPass,
 		ParseBlocks: c.Parse,
-		metrics:     metrics,
 	}
 
 	mq, err := bchain.NewMQ(c.ZeroMQBinding, pushHandler)
@@ -75,8 +72,8 @@ func NewBitcoinRPC(config json.RawMessage, pushHandler func(*bchain.MQMessage),
 	return s, nil
 }
 
-func (b *BitcoinRPC) Initialize(mempool *bchain.Mempool) error {
+func (b *BitcoinRPC) Initialize() error {
-	b.Mempool = mempool
+	b.Mempool = bchain.NewMempool(b)
 
 	chainName, err := b.GetBlockChainInfo()
 	if err != nil {
@@ -289,7 +286,7 @@ func (b *BitcoinRPC) GetBestBlockHash() (string, error) {
 
 	res := resGetBestBlockHash{}
 	req := cmdGetBestBlockHash{Method: "getbestblockhash"}
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return "", err
@@ -306,7 +303,7 @@ func (b *BitcoinRPC) GetBestBlockHeight() (uint32, error) {
 
 	res := resGetBlockCount{}
 	req := cmdGetBlockCount{Method: "getblockcount"}
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return 0, err
@@ -323,7 +320,7 @@ func (b *BitcoinRPC) GetBlockChainInfo() (string, error) {
 
 	res := resGetBlockChainInfo{}
 	req := cmdGetBlockChainInfo{Method: "getblockchaininfo"}
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return "", err
@@ -341,7 +338,7 @@ func (b *BitcoinRPC) GetBlockHash(height uint32) (string, error) {
 	res := resGetBlockHash{}
 	req := cmdGetBlockHash{Method: "getblockhash"}
 	req.Params.Height = height
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return "", errors.Annotatef(err, "height %v", height)
@@ -360,7 +357,7 @@ func (b *BitcoinRPC) GetBlockHeader(hash string) (*bchain.BlockHeader, error) {
 	req := cmdGetBlockHeader{Method: "getblockheader"}
 	req.Params.BlockHash = hash
 	req.Params.Verbose = true
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "hash %v", hash)
@@ -420,7 +417,7 @@ func (b *BitcoinRPC) GetBlockRaw(hash string) ([]byte, error) {
 	req := cmdGetBlock{Method: "getblock"}
 	req.Params.BlockHash = hash
 	req.Params.Verbosity = 0
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "hash %v", hash)
@@ -440,7 +437,7 @@ func (b *BitcoinRPC) GetBlockList(hash string) (*bchain.Block, error) {
 	req := cmdGetBlock{Method: "getblock"}
 	req.Params.BlockHash = hash
 	req.Params.Verbosity = 1
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "hash %v", hash)
@@ -472,7 +469,7 @@ func (b *BitcoinRPC) GetBlockFull(hash string) (*bchain.Block, error) {
 	req := cmdGetBlock{Method: "getblock"}
 	req.Params.BlockHash = hash
 	req.Params.Verbosity = 2
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "hash %v", hash)
@@ -489,7 +486,7 @@ func (b *BitcoinRPC) GetMempool() ([]string, error) {
 
 	res := resGetMempool{}
 	req := cmdGetMempool{Method: "getrawmempool"}
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return nil, err
@@ -508,7 +505,7 @@ func (b *BitcoinRPC) GetTransaction(txid string) (*bchain.Tx, error) {
 	req := cmdGetRawTransaction{Method: "getrawtransaction"}
 	req.Params.Txid = txid
 	req.Params.Verbose = true
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "txid %v", txid)
@@ -547,7 +544,7 @@ func (b *BitcoinRPC) EstimateSmartFee(blocks int, conservative bool) (float64, e
 	} else {
 		req.Params.EstimateMode = "ECONOMICAL"
 	}
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return 0, err
@@ -565,7 +562,7 @@ func (b *BitcoinRPC) SendRawTransaction(tx string) (string, error) {
 	res := resSendRawTransaction{}
 	req := cmdSendRawTransaction{Method: "sendrawtransaction"}
 	req.Params = []string{tx}
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return "", err
@@ -584,7 +581,7 @@ func (b *BitcoinRPC) GetMempoolEntry(txid string) (*bchain.MempoolEntry, error)
 		Method: "getmempoolentry",
 		Params: []string{txid},
 	}
-	err := b.Call(req.Method, &req, &res)
+	err := b.Call(&req, &res)
 
 	if err != nil {
 		return nil, err
@@ -595,16 +592,7 @@ func (b *BitcoinRPC) GetMempoolEntry(txid string) (*bchain.MempoolEntry, error)
 	return res.Result, nil
 }
 
-func (b *BitcoinRPC) Call(method string, req interface{}, res interface{}) error {
+func (b *BitcoinRPC) Call(req interface{}, res interface{}) error {
-	start := time.Now()
-	err := b.call(req, res)
-	if err == nil {
-		b.metrics.RPCLatency.With(common.Labels{"method": method}).Observe(float64(time.Since(start)) / 1e6) // in milliseconds
-	}
-	return err
-}
-
-func (b *BitcoinRPC) call(req interface{}, res interface{}) error {
 	httpData, err := json.Marshal(req)
 	if err != nil {
 		return err

bchain/coins/zec/zcashrpc.go

@@ -3,7 +3,6 @@ package zec
 import (
 	"blockbook/bchain"
 	"blockbook/bchain/coins/btc"
-	"blockbook/common"
 	"encoding/json"
 
 	"github.com/golang/glog"
@@ -14,8 +13,8 @@ type ZCashRPC struct {
 	*btc.BitcoinRPC
 }
 
-func NewZCashRPC(config json.RawMessage, pushHandler func(*bchain.MQMessage), metrics *common.Metrics) (bchain.BlockChain, error) {
+func NewZCashRPC(config json.RawMessage, pushHandler func(*bchain.MQMessage)) (bchain.BlockChain, error) {
-	b, err := btc.NewBitcoinRPC(config, pushHandler, metrics)
+	b, err := btc.NewBitcoinRPC(config, pushHandler)
 	if err != nil {
 		return nil, err
 	}
@@ -25,8 +24,8 @@ func NewZCashRPC(config json.RawMessage, pushHandler func(*bchain.MQMessage), me
 	return z, nil
 }
 
-func (z *ZCashRPC) Initialize(mempool *bchain.Mempool) error {
+func (z *ZCashRPC) Initialize() error {
-	z.Mempool = mempool
+	z.Mempool = bchain.NewMempool(z)
 	z.Parser = &ZCashBlockParser{}
 	z.Testnet = false
 	z.Network = "livenet"
@@ -77,7 +76,7 @@ func (z *ZCashRPC) GetBlock(hash string, height uint32) (*bchain.Block, error) {
 	req := untypedArrayParams{Method: "getblock"}
 	req.Params = append(req.Params, hash)
 	req.Params = append(req.Params, true)
-	err := z.Call(req.Method, &req, &res)
+	err := z.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "hash %v", hash)
@@ -109,7 +108,7 @@ func (z *ZCashRPC) GetTransaction(txid string) (*bchain.Tx, error) {
 	req := untypedArrayParams{Method: "getrawtransaction"}
 	req.Params = append(req.Params, txid)
 	req.Params = append(req.Params, 1)
-	err := z.Call(req.Method, &req, &res)
+	err := z.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "txid %v", txid)
@@ -127,7 +126,7 @@ func (z *ZCashRPC) GetBlockHash(height uint32) (string, error) {
 	res := resGetBlockHash{}
 	req := untypedArrayParams{Method: "getblockhash"}
 	req.Params = append(req.Params, height)
-	err := z.Call(req.Method, &req, &res)
+	err := z.Call(&req, &res)
 
 	if err != nil {
 		return "", errors.Annotatef(err, "height %v", height)
@@ -146,7 +145,7 @@ func (z *ZCashRPC) GetBlockHeader(hash string) (*bchain.BlockHeader, error) {
 	req := untypedArrayParams{Method: "getblockheader"}
 	req.Params = append(req.Params, hash)
 	req.Params = append(req.Params, true)
-	err := z.Call(req.Method, &req, &res)
+	err := z.Call(&req, &res)
 
 	if err != nil {
 		return nil, errors.Annotatef(err, "hash %v", hash)

bchain/mempool.go

@@ -1,7 +1,6 @@
 package bchain
 
 import (
-	"blockbook/common"
 	"sync"
 	"time"
 
@@ -31,12 +30,11 @@ type Mempool struct {
 	txToInputOutput map[string]inputOutput
 	scriptToTx      map[string][]outpoint // TODO rename all occurences
 	inputs          map[outpoint]string
-	metrics         *common.Metrics
 }
 
 // NewMempool creates new mempool handler.
-func NewMempool(chain BlockChain, metrics *common.Metrics) *Mempool {
+func NewMempool(chain BlockChain) *Mempool {
-	return &Mempool{chain: chain, metrics: metrics}
+	return &Mempool{chain: chain}
 }
 
 // GetTransactions returns slice of mempool transactions for given output script.
@@ -83,7 +81,6 @@ func (m *Mempool) Resync(onNewTxAddr func(txid string, addr string)) error {
 	glog.V(1).Info("Mempool: resync")
 	txs, err := m.chain.GetMempool()
 	if err != nil {
-		m.metrics.MempoolResyncErrors.With(common.Labels{"error": err.Error()}).Inc()
 		return err
 	}
 	parser := m.chain.GetChainParser()
@@ -95,7 +92,6 @@ func (m *Mempool) Resync(onNewTxAddr func(txid string, addr string)) error {
 		if !exists {
 			tx, err := m.chain.GetTransaction(txid)
 			if err != nil {
-				m.metrics.MempoolResyncErrors.With(common.Labels{"error": err.Error()}).Inc()
 				glog.Error("cannot get transaction ", txid, ": ", err)
 				continue
 			}
@@ -126,8 +122,6 @@ func (m *Mempool) Resync(onNewTxAddr func(txid string, addr string)) error {
 		}
 	}
 	m.updateMappings(newTxToInputOutput, newScriptToTx, newInputs)
-	d := time.Since(start)
+	glog.Info("Mempool: resync finished in ", time.Since(start), ", ", len(m.txToInputOutput), " transactions in mempool")
-	glog.Info("Mempool: resync finished in ", d, ", ", len(m.txToInputOutput), " transactions in mempool")
-	m.metrics.MempoolResyncDuration.Observe(float64(d) / 1e6) // in milliseconds
 	return nil
 }

bchain/types.go

@@ -87,7 +87,7 @@ func (e *RPCError) Error() string {
 
 type BlockChain interface {
 	// life-cycle methods
-	Initialize(mempool *Mempool) error
+	Initialize() error
 	Shutdown() error
 	// chain info
 	IsTestnet() bool

common/metrics.go

@@ -16,7 +16,6 @@ type Metrics struct {
 	TxCacheEfficiency     *prometheus.CounterVec
 	RPCLatency            *prometheus.HistogramVec
 	IndexResyncErrors     *prometheus.CounterVec
-	MempoolResyncErrors   *prometheus.CounterVec
 	IndexDBSize           prometheus.Gauge
 }
 
@@ -88,7 +87,7 @@ func GetMetrics(coin string) (*Metrics, error) {
 			Buckets:     []float64{1, 5, 10, 25, 50, 75, 100, 250},
 			ConstLabels: Labels{"coin": coin},
 		},
-		[]string{"method"},
+		[]string{"method", "error"},
 	)
 	metrics.IndexResyncErrors = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
@@ -98,14 +97,6 @@ func GetMetrics(coin string) (*Metrics, error) {
 		},
 		[]string{"error"},
 	)
-	metrics.MempoolResyncErrors = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Name:        "blockbook_mempool_resync_errors",
-			Help:        "Number of errors of mempool resync operation",
-			ConstLabels: Labels{"coin": coin},
-		},
-		[]string{"error"},
-	)
 	metrics.IndexDBSize = prometheus.NewGauge(
 		prometheus.GaugeOpts{
 			Name:        "blockbook_index_db_size",