Add prometheus metrics collection (#33)

* Upgrade geth

* Add prometheus metrics collection

* Update README

pkg/prom/db_stats_collector.go

@@ -0,0 +1,157 @@
+// VulcanizeDB
+// Copyright © 2022 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package prom
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
+)
+
+// DBStatsGetter is an interface that gets sql.DBStats.
+type DBStatsGetter interface {
+	Stats() sql.Stats
+}
+
+// DBStatsCollector implements the prometheus.Collector interface.
+type DBStatsCollector struct {
+	sg DBStatsGetter
+
+	// descriptions of exported metrics
+	maxOpenDesc           *prometheus.Desc
+	openDesc              *prometheus.Desc
+	inUseDesc             *prometheus.Desc
+	idleDesc              *prometheus.Desc
+	waitedForDesc         *prometheus.Desc
+	blockedSecondsDesc    *prometheus.Desc
+	closedMaxIdleDesc     *prometheus.Desc
+	closedMaxLifetimeDesc *prometheus.Desc
+}
+
+// NewDBStatsCollector creates a new DBStatsCollector.
+func NewDBStatsCollector(dbName string, sg DBStatsGetter) *DBStatsCollector {
+	labels := prometheus.Labels{"db_name": dbName}
+	return &DBStatsCollector{
+		sg: sg,
+		maxOpenDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "max_open"),
+			"Maximum number of open connections to the database.",
+			nil,
+			labels,
+		),
+		openDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "open"),
+			"The number of established connections both in use and idle.",
+			nil,
+			labels,
+		),
+		inUseDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "in_use"),
+			"The number of connections currently in use.",
+			nil,
+			labels,
+		),
+		idleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "idle"),
+			"The number of idle connections.",
+			nil,
+			labels,
+		),
+		waitedForDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "waited_for"),
+			"The total number of connections waited for.",
+			nil,
+			labels,
+		),
+		blockedSecondsDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "blocked_seconds"),
+			"The total time blocked waiting for a new connection.",
+			nil,
+			labels,
+		),
+		closedMaxIdleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "closed_max_idle"),
+			"The total number of connections closed due to SetMaxIdleConns.",
+			nil,
+			labels,
+		),
+		closedMaxLifetimeDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, connSubsystem, "closed_max_lifetime"),
+			"The total number of connections closed due to SetConnMaxLifetime.",
+			nil,
+			labels,
+		),
+	}
+}
+
+// Describe implements the prometheus.Collector interface.
+func (c DBStatsCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.maxOpenDesc
+	ch <- c.openDesc
+	ch <- c.inUseDesc
+	ch <- c.idleDesc
+	ch <- c.waitedForDesc
+	ch <- c.blockedSecondsDesc
+	ch <- c.closedMaxIdleDesc
+	ch <- c.closedMaxLifetimeDesc
+}
+
+// Collect implements the prometheus.Collector interface.
+func (c DBStatsCollector) Collect(ch chan<- prometheus.Metric) {
+	stats := c.sg.Stats()
+
+	ch <- prometheus.MustNewConstMetric(
+		c.maxOpenDesc,
+		prometheus.GaugeValue,
+		float64(stats.MaxOpen()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.openDesc,
+		prometheus.GaugeValue,
+		float64(stats.Open()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.inUseDesc,
+		prometheus.GaugeValue,
+		float64(stats.InUse()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.idleDesc,
+		prometheus.GaugeValue,
+		float64(stats.Idle()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.waitedForDesc,
+		prometheus.CounterValue,
+		float64(stats.WaitCount()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.blockedSecondsDesc,
+		prometheus.CounterValue,
+		stats.WaitDuration().Seconds(),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.closedMaxIdleDesc,
+		prometheus.CounterValue,
+		float64(stats.MaxIdleClosed()),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		c.closedMaxLifetimeDesc,
+		prometheus.CounterValue,
+		float64(stats.MaxLifetimeClosed()),
+	)
+}

pkg/prom/prom.go

@@ -0,0 +1,90 @@
+// VulcanizeDB
+// Copyright © 2022 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package prom
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	namespace = "ipld_eth_state_snapshot"
+
+	connSubsystem  = "connections"
+	statsSubsystem = "stats"
+)
+
+var (
+	metrics bool
+
+	stateNodeCount   prometheus.Counter
+	storageNodeCount prometheus.Counter
+	codeNodeCount    prometheus.Counter
+)
+
+func Init() {
+	metrics = true
+
+	stateNodeCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      "state_node_count",
+		Help:      "Number of state nodes processed",
+	})
+
+	storageNodeCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      "storage_node_count",
+		Help:      "Number of storage nodes processed",
+	})
+
+	codeNodeCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: statsSubsystem,
+		Name:      "code_node_count",
+		Help:      "Number of code nodes processed",
+	})
+}
+
+// RegisterDBCollector create metric collector for given connection
+func RegisterDBCollector(name string, db DBStatsGetter) {
+	if metrics {
+		prometheus.Register(NewDBStatsCollector(name, db))
+	}
+}
+
+// IncStateNodeCount increments the number of state nodes processed
+func IncStateNodeCount() {
+	if metrics {
+		stateNodeCount.Inc()
+	}
+}
+
+// IncStorageNodeCount increments the number of storage nodes processed
+func IncStorageNodeCount() {
+	if metrics {
+		storageNodeCount.Inc()
+	}
+}
+
+// IncCodeNodeCount increments the number of code nodes processed
+func IncCodeNodeCount() {
+	if metrics {
+		codeNodeCount.Inc()
+	}
+}

pkg/prom/serve.go

@@ -0,0 +1,47 @@
+// VulcanizeDB
+// Copyright © 2022 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package prom
+
+import (
+	"errors"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/sirupsen/logrus"
+)
+
+var errPromHTTP = errors.New("can't start http server for prometheus")
+
+// Serve start listening http
+func Serve(addr string) *http.Server {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+	srv := http.Server{
+		Addr:    addr,
+		Handler: mux,
+	}
+	go func() {
+		if err := srv.ListenAndServe(); err != nil {
+			logrus.
+				WithError(err).
+				WithField("module", "prom").
+				WithField("addr", addr).
+				Fatal(errPromHTTP)
+		}
+	}()
+	return &srv
+}

pkg/snapshot/env.go

@@ -25,6 +25,12 @@ const (
 	LOGRUS_LEVEL = "LOGRUS_LEVEL"
 	LOGRUS_FILE  = "LOGRUS_FILE"
 
+	PROM_METRICS   = "PROM_METRICS"
+	PROM_HTTP      = "PROM_HTTP"
+	PROM_HTTP_ADDR = "PROM_HTTP_ADDR"
+	PROM_HTTP_PORT = "PROM_HTTP_PORT"
+	PROM_DB_STATS  = "PROM_DB_STATS"
+
 	FILE_OUTPUT_DIR = "FILE_OUTPUT_DIR"
 
 	ANCIENT_DB_PATH = "ANCIENT_DB_PATH"
@@ -56,6 +62,12 @@ const (
 	LOGRUS_LEVEL_TOML = "log.level"
 	LOGRUS_FILE_TOML  = "log.file"
 
+	PROM_METRICS_TOML   = "prom.metrics"
+	PROM_HTTP_TOML      = "prom.http"
+	PROM_HTTP_ADDR_TOML = "prom.httpAddr"
+	PROM_HTTP_PORT_TOML = "prom.httpPort"
+	PROM_DB_STATS_TOML  = "prom.dbStats"
+
 	FILE_OUTPUT_DIR_TOML = "file.outputDir"
 
 	ANCIENT_DB_PATH_TOML = "leveldb.ancient"
@@ -87,6 +99,12 @@ const (
 	LOGRUS_LEVEL_CLI = "log-level"
 	LOGRUS_FILE_CLI  = "log-file"
 
+	PROM_METRICS_CLI   = "prom-metrics"
+	PROM_HTTP_CLI      = "prom-http"
+	PROM_HTTP_ADDR_CLI = "prom-httpAddr"
+	PROM_HTTP_PORT_CLI = "prom-httpPort"
+	PROM_DB_STATS_CLI  = "prom-dbStats"
+
 	FILE_OUTPUT_DIR_CLI = "output-dir"
 
 	ANCIENT_DB_PATH_CLI = "ancient-path"

pkg/snapshot/file/publisher.go

@@ -34,6 +34,7 @@ import (
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
 	nodeinfo "github.com/ethereum/go-ethereum/statediff/indexer/node"
 	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/prom"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
 )
 
@@ -223,6 +224,8 @@ func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, snapTx s
 	}
 	// increment state node counter.
 	atomic.AddUint64(&p.stateNodeCounter, 1)
+	prom.IncStateNodeCount()
+
 	// increment current batch size counter
 	p.currBatchSize += 2
 	return err
@@ -249,6 +252,8 @@ func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, stateP
 	}
 	// increment storage node counter.
 	atomic.AddUint64(&p.storageNodeCounter, 1)
+	prom.IncStorageNodeCount()
+
 	// increment current batch size counter
 	p.currBatchSize += 2
 	return nil
@@ -268,6 +273,8 @@ func (p *publisher) PublishCode(codeHash common.Hash, codeBytes []byte, snapTx s
 	}
 	// increment code node counter.
 	atomic.AddUint64(&p.codeNodeCounter, 1)
+	prom.IncCodeNodeCount()
+
 	p.currBatchSize++
 	return nil
 }

pkg/snapshot/file/publisher_test.go

@@ -127,6 +127,7 @@ func TestPgCopy(t *testing.T) {
 	test.NoError(t, err)
 
 	headerNode, err := ipld.NewEthHeader(&fixt.Block1_Header)
+	test.NoError(t, err)
 	test.ExpectEqual(t, headerNode.Cid().String(), header.CID)
 	test.ExpectEqual(t, fixt.Block1_Header.Hash().String(), header.BlockHash)
 }

pkg/snapshot/pg/publisher.go

@@ -33,6 +33,7 @@ import (
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 	"github.com/ethereum/go-ethereum/statediff/indexer/ipld"
 	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/prom"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
 )
 
@@ -150,6 +151,7 @@ func (p *publisher) PublishStateNode(node *snapt.Node, headerID string, snapTx s
 	}
 	// increment state node counter.
 	atomic.AddUint64(&p.stateNodeCounter, 1)
+	prom.IncStateNodeCount()
 
 	// increment current batch size counter
 	p.currBatchSize += 2
@@ -176,6 +178,7 @@ func (p *publisher) PublishStorageNode(node *snapt.Node, headerID string, stateP
 	}
 	// increment storage node counter.
 	atomic.AddUint64(&p.storageNodeCounter, 1)
+	prom.IncStorageNodeCount()
 
 	// increment current batch size counter
 	p.currBatchSize += 2
@@ -197,6 +200,7 @@ func (p *publisher) PublishCode(codeHash common.Hash, codeBytes []byte, snapTx s
 
 	// increment code node counter.
 	atomic.AddUint64(&p.codeNodeCounter, 1)
+	prom.IncCodeNodeCount()
 
 	p.currBatchSize++
 	return nil

pkg/snapshot/service.go

@@ -36,7 +36,7 @@ import (
 )
 
 var (
-	emptyNode, _      = rlp.EncodeToBytes([]byte{})
+	emptyNode, _      = rlp.EncodeToBytes(&[]byte{})
 	emptyCodeHash     = crypto.Keccak256([]byte{})
 	emptyContractRoot = crypto.Keccak256Hash(emptyNode)
 
@@ -142,7 +142,6 @@ func (s *Service) CreateSnapshot(params SnapshotParams) error {
 	} else {
 		return s.createSnapshot(iters[0], headerID)
 	}
-	return nil
 }
 
 // Create snapshot up to head (ignores height param)

pkg/snapshot/util.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
 
+	"github.com/vulcanize/ipld-eth-state-snapshot/pkg/prom"
 	file "github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot/file"
 	pg "github.com/vulcanize/ipld-eth-state-snapshot/pkg/snapshot/pg"
 	snapt "github.com/vulcanize/ipld-eth-state-snapshot/pkg/types"
@@ -18,6 +19,9 @@ func NewPublisher(mode SnapshotMode, config *Config) (snapt.Publisher, error) {
 		if err != nil {
 			return nil, err
 		}
+
+		prom.RegisterDBCollector(config.DB.ConnConfig.DatabaseName, driver)
+
 		return pg.NewPublisher(postgres.NewPostgresDB(driver)), nil
 	case FileSnapshot:
 		return file.NewPublisher(config.File.OutputDir, config.Eth.NodeInfo)