new prometheus metrics for client-stats metrics (#8834)

Author: kasey

beacon-chain/db/db.go

@@ -14,6 +14,13 @@ func NewDB(ctx context.Context, dirPath string, config *kv.Config) (Database, er
 	return kv.NewKVStore(ctx, dirPath, config)
 }
 
+// NewDBFilename uses the KVStoreDatafilePath so that if this layer of
+// indirection between db.NewDB->kv.NewKVStore ever changes, it will be easy to remember
+// to also change this filename indirection at the same time.
+func NewDBFilename(dirPath string) string {
+	return kv.KVStoreDatafilePath(dirPath)
+}
+
 // NewSlasherDB initializes a new DB for slasher.
 func NewSlasherDB(ctx context.Context, dirPath string, config *slasherkv.Config) (SlasherDatabase, error) {
 	return slasherkv.NewKVStore(ctx, dirPath, config)

beacon-chain/db/kv/kv.go

@@ -64,6 +64,13 @@ type Store struct {
 	ctx                 context.Context
 }
 
+// KVStoreDatafilePath is the canonical construction of a full
+// database file path from the directory path, so that code outside
+// this package can find the full path in a consistent way.
+func KVStoreDatafilePath(dirPath string) string {
+	return path.Join(dirPath, DatabaseFileName)
+}
+
 // NewKVStore initializes a new boltDB key-value store at the directory
 // path specified, creates the kv-buckets based on the schema, and stores
 // an open connection db object as a property of the Store struct.
@@ -77,7 +84,7 @@ func NewKVStore(ctx context.Context, dirPath string, config *Config) (*Store, er
 			return nil, err
 		}
 	}
-	datafile := path.Join(dirPath, DatabaseFileName)
+	datafile := KVStoreDatafilePath(dirPath)
 	boltDB, err := bolt.Open(
 		datafile,
 		params.BeaconIoConfig().ReadWritePermissions,

beacon-chain/node/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "config.go",
         "log.go",
         "node.go",
+        "prometheus.go",
     ],
     importpath = "github.com/prysmaticlabs/prysm/beacon-chain/node",
     visibility = [
@@ -48,6 +49,7 @@ go_library(
         "//shared/version:go_default_library",
         "@com_github_ethereum_go_ethereum//common:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prysmaticlabs_eth2_types//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@com_github_urfave_cli_v2//:go_default_library",

beacon-chain/node/node.go

@@ -73,6 +73,7 @@ type BeaconNode struct {
 	opFeed          *event.Feed
 	forkChoiceStore forkchoice.ForkChoicer
 	stateGen        *stategen.State
+	collector       *bcnodeCollector
 }
 
 // New creates a new node instance, sets up configuration options, and registers
@@ -162,6 +163,15 @@ func New(cliCtx *cli.Context) (*BeaconNode, error) {
 		}
 	}
 
+	// db.DatabasePath is the path to the containing directory
+	// db.NewDBFilename expands that to the canonical full path using
+	// the same constuction as NewDB()
+	c, err := newBeaconNodePromCollector(db.NewDBFilename(beacon.db.DatabasePath()))
+	if err != nil {
+		return nil, err
+	}
+	beacon.collector = c
+
 	return beacon, nil
 }
 
@@ -224,6 +234,7 @@ func (b *BeaconNode) Close() {
 	if err := b.db.Close(); err != nil {
 		log.Errorf("Failed to close database: %v", err)
 	}
+	b.collector.unregister()
 	b.cancel()
 	close(b.stop)
 }
@@ -435,15 +446,22 @@ func (b *BeaconNode) registerPOWChainService() error {
 		return err
 	}
 
+	bs, err := powchain.NewPowchainCollector(b.ctx)
+	if err != nil {
+		return err
+	}
+
 	cfg := &powchain.Web3ServiceConfig{
-		HttpEndpoints:      endpoints,
-		DepositContract:    common.HexToAddress(depAddress),
-		BeaconDB:           b.db,
-		DepositCache:       b.depositCache,
-		StateNotifier:      b,
-		StateGen:           b.stateGen,
-		Eth1HeaderReqLimit: b.cliCtx.Uint64(flags.Eth1HeaderReqLimit.Name),
+		HttpEndpoints:          endpoints,
+		DepositContract:        common.HexToAddress(depAddress),
+		BeaconDB:               b.db,
+		DepositCache:           b.depositCache,
+		StateNotifier:          b,
+		StateGen:               b.stateGen,
+		Eth1HeaderReqLimit:     b.cliCtx.Uint64(flags.Eth1HeaderReqLimit.Name),
+		BeaconNodeStatsUpdater: bs,
 	}
+
 	web3Service, err := powchain.NewService(b.ctx, cfg)
 	if err != nil {
 		return errors.Wrap(err, "could not register proof-of-work chain web3Service")

beacon-chain/node/prometheus.go

@@ -0,0 +1,61 @@
+package node
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type bcnodeCollector struct {
+	DiskBeaconchainBytesTotal *prometheus.Desc
+	dbPath                    string
+}
+
+func newBeaconNodePromCollector(dbPath string) (*bcnodeCollector, error) {
+	namespace := "bcnode"
+	c := &bcnodeCollector{
+		DiskBeaconchainBytesTotal: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "disk_beaconchain_bytes_total"),
+			"Total hard disk space used by the beaconchain database, in bytes.",
+			nil,
+			nil,
+		),
+		dbPath: dbPath,
+	}
+	_, err := c.getCurrentDbBytes()
+	if err != nil {
+		return nil, err
+	}
+	return c, prometheus.Register(c)
+}
+
+func (bc *bcnodeCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- bc.DiskBeaconchainBytesTotal
+}
+
+func (bc *bcnodeCollector) Collect(ch chan<- prometheus.Metric) {
+	dbBytes, err := bc.getCurrentDbBytes()
+	if err != nil {
+		log.Warn(err)
+		return
+	}
+
+	ch <- prometheus.MustNewConstMetric(
+		bc.DiskBeaconchainBytesTotal,
+		prometheus.GaugeValue,
+		dbBytes,
+	)
+}
+
+func (bc *bcnodeCollector) getCurrentDbBytes() (float64, error) {
+	fs, err := os.Stat(bc.dbPath)
+	if err != nil {
+		return 0, fmt.Errorf("could not collect database file size for prometheus, path=%s, err=%s", bc.dbPath, err)
+	}
+	return float64(fs.Size()), nil
+}
+
+func (bc *bcnodeCollector) unregister() {
+	prometheus.Unregister(bc)
+}

beacon-chain/powchain/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "deposit.go",
         "log.go",
         "log_processing.go",
+        "prometheus.go",
         "provider.go",
         "service.go",
     ],
@@ -32,6 +33,7 @@ go_library(
         "//contracts/deposit-contract:go_default_library",
         "//proto/beacon/db:go_default_library",
         "//shared/bytesutil:go_default_library",
+        "//shared/clientstats:go_default_library",
         "//shared/hashutil:go_default_library",
         "//shared/httputils:go_default_library",
         "//shared/httputils/authorizationmethod:go_default_library",
@@ -67,6 +69,7 @@ go_test(
         "init_test.go",
         "log_processing_test.go",
         "powchain_test.go",
+        "prometheus_test.go",
         "provider_test.go",
         "service_test.go",
     ],
@@ -84,6 +87,7 @@ go_test(
         "//proto/beacon/db:go_default_library",
         "//shared/bls:go_default_library",
         "//shared/bytesutil:go_default_library",
+        "//shared/clientstats:go_default_library",
         "//shared/event:go_default_library",
         "//shared/httputils:go_default_library",
         "//shared/httputils/authorizationmethod:go_default_library",
@@ -98,6 +102,7 @@ go_test(
         "@com_github_ethereum_go_ethereum//common/hexutil:go_default_library",
         "@com_github_ethereum_go_ethereum//core/types:go_default_library",
         "@com_github_ethereum_go_ethereum//trie:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prysmaticlabs_eth2_types//:go_default_library",
         "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",

beacon-chain/powchain/prometheus.go

@@ -0,0 +1,153 @@
+package powchain
+
+import (
+	"context"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prysmaticlabs/prysm/shared/clientstats"
+)
+
+type BeaconNodeStatsUpdater interface {
+	Update(stats clientstats.BeaconNodeStats)
+}
+
+type PowchainCollector struct {
+	SyncEth1Connected          *prometheus.Desc
+	SyncEth1FallbackConnected  *prometheus.Desc
+	SyncEth1FallbackConfigured *prometheus.Desc // true if flag specified: --fallback-web3provider
+	updateChan                 chan clientstats.BeaconNodeStats
+	latestStats                clientstats.BeaconNodeStats
+	sync.Mutex
+	ctx        context.Context
+	finishChan chan struct{}
+}
+
+var _ BeaconNodeStatsUpdater = &PowchainCollector{}
+var _ prometheus.Collector = &PowchainCollector{}
+
+// Update satisfies the BeaconNodeStatsUpdater
+func (pc *PowchainCollector) Update(update clientstats.BeaconNodeStats) {
+	pc.updateChan <- update
+}
+
+// Describe is invoked by the prometheus collection loop.
+// It returns a set of metric Descriptor references which
+// are also used in Collect to group collected metrics into
+// a family. Describe and Collect together satisfy the
+// prometheus.Collector interface.
+func (pc *PowchainCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- pc.SyncEth1Connected
+	ch <- pc.SyncEth1FallbackConfigured
+	ch <- pc.SyncEth1FallbackConnected
+}
+
+// Collect is invoked by the prometheus collection loop.
+// It returns a set of Metrics representing the observation
+// for the current collection period. In the case of this
+// collector, we use values from the latest BeaconNodeStats
+// value sent by the powchain Service, which updates this value
+// whenever an internal event could change the state of one of
+// the metrics.
+// Describe and Collect together satisfy the
+// prometheus.Collector interface.
+func (pc *PowchainCollector) Collect(ch chan<- prometheus.Metric) {
+	bs := pc.getLatestStats()
+
+	var syncEth1FallbackConfigured float64 = 0
+	if bs.SyncEth1FallbackConfigured {
+		syncEth1FallbackConfigured = 1
+	}
+	ch <- prometheus.MustNewConstMetric(
+		pc.SyncEth1FallbackConfigured,
+		prometheus.GaugeValue,
+		syncEth1FallbackConfigured,
+	)
+
+	var syncEth1FallbackConnected float64 = 0
+	if bs.SyncEth1FallbackConnected {
+		syncEth1FallbackConnected = 1
+	}
+	ch <- prometheus.MustNewConstMetric(
+		pc.SyncEth1FallbackConnected,
+		prometheus.GaugeValue,
+		syncEth1FallbackConnected,
+	)
+
+	var syncEth1Connected float64 = 0
+	if bs.SyncEth1Connected {
+		syncEth1Connected = 1
+	}
+	ch <- prometheus.MustNewConstMetric(
+		pc.SyncEth1Connected,
+		prometheus.GaugeValue,
+		syncEth1Connected,
+	)
+}
+
+func (pc *PowchainCollector) getLatestStats() clientstats.BeaconNodeStats {
+	pc.Lock()
+	defer pc.Unlock()
+	return pc.latestStats
+}
+
+func (pc *PowchainCollector) setLatestStats(bs clientstats.BeaconNodeStats) {
+	pc.Lock()
+	pc.latestStats = bs
+	pc.Unlock()
+}
+
+// unregister returns true if the prometheus DefaultRegistry
+// confirms that it was removed.
+func (pc *PowchainCollector) unregister() bool {
+	return prometheus.Unregister(pc)
+}
+
+func (pc *PowchainCollector) latestStatsUpdateLoop() {
+	for {
+		select {
+		case <-pc.ctx.Done():
+			pc.unregister()
+			pc.finishChan <- struct{}{}
+			return
+		case bs := <-pc.updateChan:
+			pc.setLatestStats(bs)
+		}
+	}
+}
+
+func NewPowchainCollector(ctx context.Context) (*PowchainCollector, error) {
+	namespace := "powchain"
+	updateChan := make(chan clientstats.BeaconNodeStats, 2)
+	c := &PowchainCollector{
+		SyncEth1FallbackConfigured: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "sync_eth1_fallback_configured"),
+			"Boolean recording whether a fallback eth1 endpoint was configured: 0=false, 1=true.",
+			nil,
+			nil,
+		),
+		SyncEth1FallbackConnected: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "sync_eth1_fallback_connected"),
+			"Boolean indicating whether a fallback eth1 endpoint is currently connected: 0=false, 1=true.",
+			nil,
+			nil,
+		),
+		SyncEth1Connected: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "sync_eth1_connected"),
+			"Boolean indicating whether a fallback eth1 endpoint is currently connected: 0=false, 1=true.",
+			nil,
+			nil,
+		),
+		updateChan: updateChan,
+		ctx:        ctx,
+		finishChan: make(chan struct{}, 1),
+	}
+	go c.latestStatsUpdateLoop()
+	return c, prometheus.Register(c)
+}
+
+type NopBeaconNodeStatsUpdater struct{}
+
+func (nop *NopBeaconNodeStatsUpdater) Update(stats clientstats.BeaconNodeStats) {}
+
+var _ BeaconNodeStatsUpdater = &NopBeaconNodeStatsUpdater{}