feat: add support for metrics collection (#129)

## Summary
* Add generic `MetricsSink` interface that fans out to enabled metrics
collectors
* Add support for DogstatsD (DataDog's Statsd client)
* Add support for Prometheus server

## Deprecations

* Removes old `--statsd.url` runtime flag

README.md

@@ -122,17 +122,21 @@ Global Flags:
       --database.port int                         PostgreSQL port (default 5432)
       --database.schema_name string               PostgreSQL schema name (default "public")
       --database.user string                      PostgreSQL username (default "sidecar")
+      --datadog.statsd.enabled                    e.g. "true" or "false"
+      --datadog.statsd.url string                 e.g. "localhost:8125"
       --debug                                     "true" or "false"
       --ethereum.chunked_batch_call_size int      The number of calls to make in parallel when using the chunked batch call method (default 10)
       --ethereum.contract_call_batch_size int     The number of contract calls to batch together when fetching data from the Ethereum node (default 25)
       --ethereum.native_batch_call_size int       The number of calls to batch together when using the native eth_call method (default 500)
       --ethereum.rpc-url string                   e.g. "http://34.229.43.36:8545"
       --ethereum.use_native_batch_call            Use the native eth_call method for batch calls (default true)
+      --prometheus.enabled                        e.g. "true" or "false"
+      --prometheus.port int                       The port to run the prometheus server on (default 2112)
       --rewards.generate_staker_operators_table   Generate staker operators table while indexing
       --rewards.validate_rewards_root             Validate rewards roots while indexing (default true)
       --rpc.grpc-port int                         gRPC port (default 7100)
       --rpc.http-port int                         http rpc port (default 7101)
-      --statsd.url string                         e.g. "localhost:8125"
+
 
 ```
 

cmd/database.go

@@ -41,9 +41,14 @@ var runDatabaseCmd = &cobra.Command{
 			zap.String("commit", version.GetCommit()),
 		)
 
-		sdc, err := metrics.InitStatsdClient(cfg.StatsdUrl)
+		metricsClients, err := metrics.InitMetricsSinksFromConfig(cfg, l)
 		if err != nil {
-			l.Sugar().Fatal("Failed to setup statsd client", zap.Error(err))
+			l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
+		}
+
+		sdc, err := metrics.NewMetricsSink(&metrics.MetricsSinkConfig{}, metricsClients)
+		if err != nil {
+			l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
 		}
 
 		client := ethereum.NewClient(ethereum.ConvertGlobalConfigToEthereumConfig(&cfg.EthereumRpcConfig), l)
@@ -96,7 +101,7 @@ var runDatabaseCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to create rewards calculator", zap.Error(err))
 		}
 
-		_ = pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, cfg, l)
+		_ = pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, cfg, sdc, l)
 
 		l.Sugar().Infow("Done")
 	},

cmd/debugger/main.go

@@ -32,9 +32,14 @@ func main() {
 
 	l, _ := logger.NewLogger(&logger.LoggerConfig{Debug: cfg.Debug})
 
-	sdc, err := metrics.InitStatsdClient(cfg.StatsdUrl)
+	metricsClients, err := metrics.InitMetricsSinksFromConfig(cfg, l)
 	if err != nil {
-		l.Sugar().Fatal("Failed to setup statsd client", zap.Error(err))
+		l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
+	}
+
+	sdc, err := metrics.NewMetricsSink(&metrics.MetricsSinkConfig{}, metricsClients)
+	if err != nil {
+		l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
 	}
 
 	client := ethereum.NewClient(ethereum.ConvertGlobalConfigToEthereumConfig(&cfg.EthereumRpcConfig), l)
@@ -87,7 +92,7 @@ func main() {
 		l.Sugar().Fatalw("Failed to create rewards calculator", zap.Error(err))
 	}
 
-	p := pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, cfg, l)
+	p := pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, cfg, sdc, l)
 
 	// Create new sidecar instance
 	sidecar := sidecar.NewSidecar(&sidecar.SidecarConfig{

cmd/operatorRestakedStrategies.go

@@ -34,9 +34,14 @@ var runOperatorRestakedStrategiesCmd = &cobra.Command{
 
 		l, _ := logger.NewLogger(&logger.LoggerConfig{Debug: cfg.Debug})
 
-		sdc, err := metrics.InitStatsdClient(cfg.StatsdUrl)
+		metricsClients, err := metrics.InitMetricsSinksFromConfig(cfg, l)
 		if err != nil {
-			l.Sugar().Fatal("Failed to setup statsd client", zap.Error(err))
+			l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
+		}
+
+		sdc, err := metrics.NewMetricsSink(&metrics.MetricsSinkConfig{}, metricsClients)
+		if err != nil {
+			l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
 		}
 
 		client := ethereum.NewClient(ethereum.ConvertGlobalConfigToEthereumConfig(&cfg.EthereumRpcConfig), l)

cmd/root.go

@@ -26,7 +26,6 @@ func init() {
 
 	rootCmd.PersistentFlags().Bool("debug", false, `"true" or "false"`)
 	rootCmd.PersistentFlags().StringP("chain", "c", "mainnet", "The chain to use (mainnet, holesky, preprod")
-	rootCmd.PersistentFlags().String("statsd.url", "", `e.g. "localhost:8125"`)
 
 	rootCmd.PersistentFlags().String("ethereum.rpc-url", "", `e.g. "http://34.229.43.36:8545"`)
 	rootCmd.PersistentFlags().Int(config.EthereumRpcContractCallBatchSize, 25, `The number of contract calls to batch together when fetching data from the Ethereum node`)
@@ -47,6 +46,12 @@ func init() {
 	rootCmd.PersistentFlags().Int("rpc.grpc-port", 7100, `gRPC port`)
 	rootCmd.PersistentFlags().Int("rpc.http-port", 7101, `http rpc port`)
 
+	rootCmd.PersistentFlags().Bool("datadog.statsd.enabled", false, `e.g. "true" or "false"`)
+	rootCmd.PersistentFlags().String("datadog.statsd.url", "", `e.g. "localhost:8125"`)
+
+	rootCmd.PersistentFlags().Bool("prometheus.enabled", false, `e.g. "true" or "false"`)
+	rootCmd.PersistentFlags().Int("prometheus.port", 2112, `The port to run the prometheus server on`)
+
 	rootCmd.PersistentFlags().VisitAll(func(f *pflag.Flag) {
 		key := config.KebabToSnakeCase(f.Name)
 		viper.BindPFlag(key, f) //nolint:errcheck

cmd/run.go

@@ -48,9 +48,14 @@ var runCmd = &cobra.Command{
 			zap.String("chain", cfg.Chain.String()),
 		)
 
-		sdc, err := metrics.InitStatsdClient(cfg.StatsdUrl)
+		metricsClients, err := metrics.InitMetricsSinksFromConfig(cfg, l)
 		if err != nil {
-			l.Sugar().Fatal("Failed to setup statsd client", zap.Error(err))
+			l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
+		}
+
+		sdc, err := metrics.NewMetricsSink(&metrics.MetricsSinkConfig{}, metricsClients)
+		if err != nil {
+			l.Sugar().Fatal("Failed to setup metrics sink", zap.Error(err))
 		}
 
 		client := ethereum.NewClient(ethereum.ConvertGlobalConfigToEthereumConfig(&cfg.EthereumRpcConfig), l)
@@ -103,7 +108,7 @@ var runCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to create rewards calculator", zap.Error(err))
 		}
 
-		p := pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, cfg, l)
+		p := pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, cfg, sdc, l)
 
 		// Create new sidecar instance
 		sidecar := sidecar.NewSidecar(&sidecar.SidecarConfig{
@@ -117,6 +122,11 @@ var runCmd = &cobra.Command{
 			l.Sugar().Fatalw("Failed to start RPC server", zap.Error(err))
 		}
 
+		promChan := make(chan bool)
+		if cfg.PrometheusConfig.Enabled {
+			sidecar.WithPrometheusServer(promChan)
+		}
+
 		// Start the sidecar main process in a goroutine so that we can listen for a shutdown signal
 		go sidecar.Start(ctx)
 

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0
 	github.com/lib/pq v1.10.9
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.12.0
 	github.com/rs/cors v1.7.0
 	github.com/shopspring/decimal v1.4.0
 	github.com/spf13/cobra v1.8.1
@@ -34,9 +35,11 @@ require (
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/StackExchange/wmi v1.2.1 // indirect
 	github.com/bahlo/generic-list-go v0.2.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.13.0 // indirect
 	github.com/btcsuite/btcd/btcec/v2 v2.3.4 // indirect
 	github.com/buger/jsonparser v1.1.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/consensys/bavard v0.1.13 // indirect
 	github.com/consensys/gnark-crypto v0.12.1 // indirect
 	github.com/crate-crypto/go-ipa v0.0.0-20240223125850-b1e8a79f509c // indirect
@@ -64,10 +67,14 @@ require (
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mmcloughlin/addchain v0.4.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/prometheus/client_model v0.2.1-0.20210607210712-147c58e9608a // indirect
+	github.com/prometheus/common v0.32.1 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/rs/zerolog v1.33.0 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect