Add basic prometheus metrics (#519)

- Adds counters for blocks processed and votes submitted
- Starts prometheus collection on `/metrics`

Makefile

@@ -60,7 +60,7 @@ rebuild-contracts:
 license:
 	@echo "  >  \033[32mAdding license headers...\033[0m "
 	GO111MODULE=off go get -u github.com/google/addlicense
-	addlicense -c "ChainSafe Systems" -f ./copyright.txt -y 2020 .
+	addlicense -c "ChainSafe Systems" -f ./scripts/header.txt -y 2020 .
 
 ## license-check: Checks for missing license headers
 license-check:

README.md

@@ -97,10 +97,12 @@ For testing purposes, chainbridge provides 5 test keys. The can be used with `--
 
 ## Metrics
 
-A basic health status check can be enabled with the `--metrics` flag (default port `8001`, use `--metricsPort` to specify).
+Basic metrics and a health status check can be enabled with the `--metrics` flag (default port `8001`, use `--metricsPort` to specify).
 
 The endpoint `/health` will return the current block height and a timestamp of when it was processed. If the timestamp is at least 120 seconds old an error will be returned.
 
+Prometheus metrics are served on `/metrics`.
+
 # Chain Implementations
 
 - Ethereum (Solidity): [chainbridge-solidity](https://github.com/ChainSafe/chainbridge-solidity) 

chains/ethereum/chain.go

@@ -90,7 +90,7 @@ func setupBlockstore(cfg *Config, kp *secp256k1.Keypair) (*blockstore.Blockstore
 	return bs, nil
 }
 
-func InitializeChain(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error) (*Chain, error) {
+func InitializeChain(chainCfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, m *metrics.ChainMetrics) (*Chain, error) {
 	cfg, err := parseChainConfig(chainCfg)
 	if err != nil {
 		return nil, err
@@ -163,10 +163,10 @@ func InitializeChain(chainCfg *core.ChainConfig, logger log15.Logger, sysErr cha
 		cfg.startBlock = curr
 	}
 
-	listener := NewListener(conn, cfg, logger, bs, stop, sysErr)
+	listener := NewListener(conn, cfg, logger, bs, stop, sysErr, m)
 	listener.setContracts(bridgeContract, erc20HandlerContract, erc721HandlerContract, genericHandlerContract)
 
-	writer := NewWriter(conn, cfg, logger, stop, sysErr)
+	writer := NewWriter(conn, cfg, logger, stop, sysErr, m)
 	writer.setContract(bridgeContract)
 
 	return &Chain{

chains/ethereum/chain_test.go

@@ -39,7 +39,7 @@ func TestChain_ListenerShutdownOnFailure(t *testing.T) {
 		},
 	}
 	sysErr := make(chan error)
-	chain, err := InitializeChain(cfg, TestLogger, sysErr)
+	chain, err := InitializeChain(cfg, TestLogger, sysErr, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -98,7 +98,7 @@ func TestChain_WriterShutdownOnFailure(t *testing.T) {
 		},
 	}
 	sysErr := make(chan error)
-	chain, err := InitializeChain(cfg, TestLogger, sysErr)
+	chain, err := InitializeChain(cfg, TestLogger, sysErr, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

chains/ethereum/listener.go

@@ -43,10 +43,11 @@ type listener struct {
 	stop                   <-chan int
 	sysErr                 chan<- error // Reports fatal error to core
 	latestBlock            metrics.LatestBlock
+	metrics                *metrics.ChainMetrics
 }
 
 // NewListener creates and returns a listener
-func NewListener(conn Connection, cfg *Config, log log15.Logger, bs blockstore.Blockstorer, stop <-chan int, sysErr chan<- error) *listener {
+func NewListener(conn Connection, cfg *Config, log log15.Logger, bs blockstore.Blockstorer, stop <-chan int, sysErr chan<- error, m *metrics.ChainMetrics) *listener {
 	return &listener{
 		cfg:         *cfg,
 		conn:        conn,
@@ -55,6 +56,7 @@ func NewListener(conn Connection, cfg *Config, log log15.Logger, bs blockstore.B
 		stop:        stop,
 		sysErr:      sysErr,
 		latestBlock: metrics.LatestBlock{LastUpdated: time.Now()},
+		metrics:     m,
 	}
 }
 
@@ -138,6 +140,9 @@ func (l *listener) pollBlocks() error {
 			l.latestBlock.Height = big.NewInt(0).Set(latestBlock)
 			l.latestBlock.LastUpdated = time.Now()
 			retry = BlockRetryLimit
+			if l.metrics != nil {
+				l.metrics.BlocksProcessed.Inc()
+			}
 		}
 	}
 }

chains/ethereum/listener_test.go

@@ -65,7 +65,7 @@ func createTestListener(t *testing.T, config *Config, contracts *utils.DeployedC
 	}
 
 	router := &MockRouter{msgs: make(chan msg.Message)}
-	listener := NewListener(conn, &newConfig, TestLogger, &blockstore.EmptyStore{}, stop, sysErr)
+	listener := NewListener(conn, &newConfig, TestLogger, &blockstore.EmptyStore{}, stop, sysErr, nil)
 	listener.setContracts(bridgeContract, erc20HandlerContract, erc721HandlerContract, genericHandlerContract)
 	listener.setRouter(router)
 	// Start the listener

chains/ethereum/writer.go

@@ -6,6 +6,7 @@ package ethereum
 import (
 	"github.com/ChainSafe/ChainBridge/bindings/Bridge"
 	"github.com/ChainSafe/ChainBridge/chains"
+	metrics "github.com/ChainSafe/ChainBridge/metrics/types"
 	"github.com/ChainSafe/chainbridge-utils/msg"
 	"github.com/ChainSafe/log15"
 )
@@ -24,16 +25,18 @@ type writer struct {
 	log            log15.Logger
 	stop           <-chan int
 	sysErr         chan<- error // Reports fatal error to core
+	metrics        *metrics.ChainMetrics
 }
 
 // NewWriter creates and returns writer
-func NewWriter(conn Connection, cfg *Config, log log15.Logger, stop <-chan int, sysErr chan<- error) *writer {
+func NewWriter(conn Connection, cfg *Config, log log15.Logger, stop <-chan int, sysErr chan<- error, m *metrics.ChainMetrics) *writer {
 	return &writer{
-		cfg:    *cfg,
-		conn:   conn,
-		log:    log,
-		stop:   stop,
-		sysErr: sysErr,
+		cfg:     *cfg,
+		conn:    conn,
+		log:     log,
+		stop:    stop,
+		sysErr:  sysErr,
+		metrics: m,
 	}
 }
 

chains/ethereum/writer_methods.go

@@ -239,6 +239,9 @@ func (w *writer) voteProposal(m msg.Message, dataHash [32]byte) {
 
 			if err == nil {
 				w.log.Info("Submitted proposal vote", "tx", tx.Hash(), "src", m.Source, "depositNonce", m.DepositNonce)
+				if w.metrics != nil {
+					w.metrics.VotesSubmitted.Inc()
+				}
 				return
 			} else if err.Error() == ErrNonceTooLow.Error() || err.Error() == ErrTxUnderpriced.Error() {
 				w.log.Debug("Nonce too low, will retry")

chains/ethereum/writer_test.go

@@ -33,7 +33,7 @@ func createTestWriter(t *testing.T, cfg *Config, errs chan<- error) (*writer, fu
 
 	conn := newLocalConnection(t, cfg)
 	stop := make(chan int)
-	writer := NewWriter(conn, cfg, newTestLogger(cfg.name), stop, errs)
+	writer := NewWriter(conn, cfg, newTestLogger(cfg.name), stop, errs, nil)
 
 	bridge, err := Bridge.NewBridge(cfg.bridgeContract, conn.Client())
 	if err != nil {
@@ -112,7 +112,7 @@ func TestWriter_start_stop(t *testing.T) {
 	defer conn.Close()
 
 	stop := make(chan int)
-	writer := NewWriter(conn, aliceTestConfig, TestLogger, stop, nil)
+	writer := NewWriter(conn, aliceTestConfig, TestLogger, stop, nil, nil)
 
 	err := writer.start()
 	if err != nil {

chains/substrate/chain.go

@@ -59,7 +59,7 @@ func checkBlockstore(bs *blockstore.Blockstore, startBlock uint64) (uint64, erro
 	}
 }
 
-func InitializeChain(cfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error) (*Chain, error) {
+func InitializeChain(cfg *core.ChainConfig, logger log15.Logger, sysErr chan<- error, m *metrics.ChainMetrics) (*Chain, error) {
 	kp, err := keystore.KeypairFromAddress(cfg.From, keystore.SubChain, cfg.KeystorePath, cfg.Insecure)
 	if err != nil {
 		return nil, err
@@ -102,8 +102,8 @@ func InitializeChain(cfg *core.ChainConfig, logger log15.Logger, sysErr chan<- e
 	}
 
 	// Setup listener & writer
-	l := NewListener(conn, cfg.Name, cfg.Id, startBlock, logger, bs, stop, sysErr)
-	w := NewWriter(conn, logger, sysErr)
+	l := NewListener(conn, cfg.Name, cfg.Id, startBlock, logger, bs, stop, sysErr, m)
+	w := NewWriter(conn, logger, sysErr, m)
 	return &Chain{
 		cfg:      cfg,
 		conn:     conn,

chains/substrate/listener.go

@@ -30,13 +30,14 @@ type listener struct {
 	stop          <-chan int
 	sysErr        chan<- error
 	latestBlock   metrics.LatestBlock
+	metrics       *metrics.ChainMetrics
 }
 
 // Frequency of polling for a new block
 var BlockRetryInterval = time.Second * 5
 var BlockRetryLimit = 5
 
-func NewListener(conn *Connection, name string, id msg.ChainId, startBlock uint64, log log15.Logger, bs blockstore.Blockstorer, stop <-chan int, sysErr chan<- error) *listener {
+func NewListener(conn *Connection, name string, id msg.ChainId, startBlock uint64, log log15.Logger, bs blockstore.Blockstorer, stop <-chan int, sysErr chan<- error, m *metrics.ChainMetrics) *listener {
 	return &listener{
 		name:          name,
 		chainId:       id,
@@ -48,6 +49,7 @@ func NewListener(conn *Connection, name string, id msg.ChainId, startBlock uint6
 		stop:          stop,
 		sysErr:        sysErr,
 		latestBlock:   metrics.LatestBlock{LastUpdated: time.Now()},
+		metrics:       m,
 	}
 }
 
@@ -165,6 +167,9 @@ func (l *listener) pollBlocks() error {
 			l.latestBlock.Height = big.NewInt(0).SetUint64(currentBlock)
 			l.latestBlock.LastUpdated = time.Now()
 			retry = BlockRetryLimit
+			if l.metrics != nil {
+				l.metrics.BlocksProcessed.Inc()
+			}
 		}
 	}
 }

chains/substrate/listener_test.go

@@ -37,7 +37,7 @@ func newTestListener(client *utils.Client, conn *Connection) (*listener, chan er
 	}
 
 	errs := make(chan error)
-	l := NewListener(conn, "Alice", 1, startBlock, AliceTestLogger, &blockstore.EmptyStore{}, make(chan int), errs)
+	l := NewListener(conn, "Alice", 1, startBlock, AliceTestLogger, &blockstore.EmptyStore{}, make(chan int), errs, nil)
 	l.setRouter(r)
 	err = l.start()
 	if err != nil {

chains/substrate/test_helper_test.go

@@ -93,8 +93,8 @@ func TestMain(m *testing.M) {
 	if err != nil {
 		panic(err)
 	}
-	alice := NewWriter(aliceConn, AliceTestLogger, wSysErr)
-	bob := NewWriter(bobConn, BobTestLogger, wSysErr)
+	alice := NewWriter(aliceConn, AliceTestLogger, wSysErr, nil)
+	bob := NewWriter(bobConn, BobTestLogger, wSysErr, nil)
 	context = testContext{
 		client:         client,
 		listener:       l,

chains/substrate/writer.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/ChainSafe/ChainBridge/chains"
+	metrics "github.com/ChainSafe/ChainBridge/metrics/types"
 	utils "github.com/ChainSafe/ChainBridge/shared/substrate"
 	"github.com/ChainSafe/chainbridge-utils/msg"
 	"github.com/ChainSafe/log15"
@@ -22,16 +23,18 @@ var AcknowledgeProposal utils.Method = utils.BridgePalletName + ".acknowledge_pr
 var TerminatedError = errors.New("terminated")
 
 type writer struct {
-	conn   *Connection
-	log    log15.Logger
-	sysErr chan<- error
+	conn    *Connection
+	log     log15.Logger
+	sysErr  chan<- error
+	metrics *metrics.ChainMetrics
 }
 
-func NewWriter(conn *Connection, log log15.Logger, sysErr chan<- error) *writer {
+func NewWriter(conn *Connection, log log15.Logger, sysErr chan<- error, m *metrics.ChainMetrics) *writer {
 	return &writer{
-		conn:   conn,
-		log:    log,
-		sysErr: sysErr,
+		conn:    conn,
+		log:     log,
+		sysErr:  sysErr,
+		metrics: m,
 	}
 }
 
@@ -81,6 +84,9 @@ func (w *writer) ResolveMessage(m msg.Message) bool {
 				time.Sleep(BlockRetryInterval)
 				continue
 			}
+			if w.metrics != nil {
+				w.metrics.VotesSubmitted.Inc()
+			}
 			return true
 		} else {
 			w.log.Debug("Ignoring proposal", "reason", reason, "nonce", prop.depositNonce, "source", prop.sourceId, "resource", prop.resourceId)

cmd/chainbridge/main.go

@@ -9,6 +9,8 @@ package main
 
 import (
 	"errors"
+	"fmt"
+	"net/http"
 	"os"
 	"strconv"
 
@@ -17,8 +19,10 @@ import (
 	"github.com/ChainSafe/ChainBridge/config"
 	"github.com/ChainSafe/ChainBridge/core"
 	"github.com/ChainSafe/ChainBridge/metrics/health"
+	metrics "github.com/ChainSafe/ChainBridge/metrics/types"
 	"github.com/ChainSafe/chainbridge-utils/msg"
 	log "github.com/ChainSafe/log15"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/urfave/cli/v2"
 )
 
@@ -176,11 +180,18 @@ func run(ctx *cli.Context) error {
 			Opts:           chain.Opts,
 		}
 		var newChain core.Chain
+		var m *metrics.ChainMetrics
+
 		logger := log.Root().New("chain", chainConfig.Name)
+
+		if ctx.Bool(config.MetricsFlag.Name) {
+			m = metrics.NewChainMetrics(chain.Name)
+		}
+
 		if chain.Type == "ethereum" {
-			newChain, err = ethereum.InitializeChain(chainConfig, logger, sysErr)
+			newChain, err = ethereum.InitializeChain(chainConfig, logger, sysErr, m)
 		} else if chain.Type == "substrate" {
-			newChain, err = substrate.InitializeChain(chainConfig, logger, sysErr)
+			newChain, err = substrate.InitializeChain(chainConfig, logger, sysErr, m)
 		} else {
 			return errors.New("unrecognized Chain Type")
 		}
@@ -189,12 +200,24 @@ func run(ctx *cli.Context) error {
 			return err
 		}
 		c.AddChain(newChain)
+
 	}
 
-	// Start metrics server
+	// Start prometheus and health server
 	if ctx.Bool(config.MetricsFlag.Name) {
 		port := ctx.Int(config.MetricsPort.Name)
-		go health.Start(port, c.Registry)
+		h := health.NewHealthServer(port, c.Registry)
+
+		go func() {
+			http.Handle("/metrics", promhttp.Handler())
+			http.HandleFunc("/health", h.HealthStatus)
+			err := http.ListenAndServe(fmt.Sprintf(":%d", port), nil)
+			if err == http.ErrServerClosed {
+				log.Info("Health status server is shutting down", err)
+			} else {
+				log.Error("Error serving metrics", "err", err)
+			}
+		}()
 	}
 
 	c.Start()

e2e/e2e_test.go

@@ -71,19 +71,19 @@ func createAndStartBridge(t *testing.T, name string, contractsA, contractsB *eth
 	logger := log.Root().New()
 	sysErr := make(chan error)
 	ethACfg := eth.CreateConfig(name, EthAChainId, contractsA, eth.EthAEndpoint)
-	ethA, err := ethChain.InitializeChain(ethACfg, logger.New("relayer", name, "chain", "ethA"), sysErr)
+	ethA, err := ethChain.InitializeChain(ethACfg, logger.New("relayer", name, "chain", "ethA"), sysErr, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	subCfg := sub.CreateConfig(name, SubChainId)
-	subA, err := subChain.InitializeChain(subCfg, logger.New("relayer", name, "chain", "sub"), sysErr)
+	subA, err := subChain.InitializeChain(subCfg, logger.New("relayer", name, "chain", "sub"), sysErr, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	ethBCfg := eth.CreateConfig(name, EthBChainId, contractsB, eth.EthBEndpoint)
-	ethB, err := ethChain.InitializeChain(ethBCfg, logger.New("relayer", name, "chain", "ethB"), sysErr)
+	ethB, err := ethChain.InitializeChain(ethBCfg, logger.New("relayer", name, "chain", "ethB"), sysErr, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/deckarep/golang-set v1.7.1 // indirect
 	github.com/ethereum/go-ethereum v1.9.17
 	github.com/gorilla/websocket v1.4.2 // indirect
-	github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416
+	github.com/prometheus/client_golang v1.4.1
 	github.com/rs/cors v1.7.0 // indirect
 	github.com/stretchr/testify v1.4.0
 	github.com/urfave/cli/v2 v2.2.0