Node metrics

node/cmd/main.go

@@ -85,8 +85,14 @@ func NodeMain(ctx *cli.Context) error {
 	}
 
 	// Creates the GRPC server.
+
+	// TODO(cody-littley): the metrics server is currently started by eigenmetrics, which is in another repo.
+	//  When we fully remove v1 support, we need to start the metrics server inside the v2 metrics code.
 	server := nodegrpc.NewServer(config, node, logger, ratelimiter)
-	serverV2 := nodegrpc.NewServerV2(config, node, logger, ratelimiter)
+	serverV2, err := nodegrpc.NewServerV2(config, node, logger, ratelimiter, reg)
+	if err != nil {
+		return fmt.Errorf("failed to create server v2: %v", err)
+	}
 	err = nodegrpc.RunServers(server, serverV2, config, logger)
 
 	return err

node/config.go

@@ -52,7 +52,7 @@ type Config struct {
 	EnableNodeApi                  bool
 	NodeApiPort                    string
 	EnableMetrics                  bool
-	MetricsPort                    string
+	MetricsPort                    int
 	OnchainMetricsInterval         int64
 	Timeout                        time.Duration
 	RegisterNodeAtStart            bool
@@ -62,6 +62,7 @@ type Config struct {
 	OverrideStoreDurationBlocks    int64
 	QuorumIDList                   []core.QuorumID
 	DbPath                         string
+	DBSizePollPeriod               time.Duration
 	LogPath                        string
 	PrivateBls                     string
 	ID                             core.OperatorID
@@ -207,7 +208,7 @@ func NewConfig(ctx *cli.Context) (*Config, error) {
 		EnableNodeApi:                  ctx.GlobalBool(flags.EnableNodeApiFlag.Name),
 		NodeApiPort:                    ctx.GlobalString(flags.NodeApiPortFlag.Name),
 		EnableMetrics:                  ctx.GlobalBool(flags.EnableMetricsFlag.Name),
-		MetricsPort:                    ctx.GlobalString(flags.MetricsPortFlag.Name),
+		MetricsPort:                    ctx.GlobalInt(flags.MetricsPortFlag.Name),
 		OnchainMetricsInterval:         ctx.GlobalInt64(flags.OnchainMetricsIntervalFlag.Name),
 		Timeout:                        timeout,
 		RegisterNodeAtStart:            registerNodeAtStart,
@@ -218,6 +219,7 @@ func NewConfig(ctx *cli.Context) (*Config, error) {
 		OverrideStoreDurationBlocks:    ctx.GlobalInt64(flags.OverrideStoreDurationBlocksFlag.Name),
 		QuorumIDList:                   ids,
 		DbPath:                         ctx.GlobalString(flags.DbPathFlag.Name),
+		DBSizePollPeriod:               ctx.GlobalDuration(flags.MetricsDBSizePollPeriodFlag.Name),
 		PrivateBls:                     privateBls,
 		EthClientConfig:                ethClientConfig,
 		EncoderConfig:                  kzg.ReadCLIConfig(ctx),

node/flags/flags.go

@@ -66,11 +66,11 @@ var (
 		Required: true,
 		EnvVar:   common.PrefixEnvVar(EnvVarPrefix, "ENABLE_METRICS"),
 	}
-	MetricsPortFlag = cli.StringFlag{
+	MetricsPortFlag = cli.IntFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "metrics-port"),
 		Usage:    "Port at which node listens for metrics calls",
 		Required: false,
-		Value:    "9091",
+		Value:    9091,
 		EnvVar:   common.PrefixEnvVar(EnvVarPrefix, "METRICS_PORT"),
 	}
 	OnchainMetricsIntervalFlag = cli.StringFlag{
@@ -98,6 +98,14 @@ var (
 		Required: true,
 		EnvVar:   common.PrefixEnvVar(EnvVarPrefix, "DB_PATH"),
 	}
+	MetricsDBSizePollPeriodFlag = cli.DurationFlag{
+		Name: common.PrefixFlag(FlagPrefix, "metrics-db-size-poll-period"),
+		Usage: "The period at which the database size is polled for metrics. " +
+			"If set to 0, the database size is not polled.",
+		Required: false,
+		Value:    10 * time.Minute,
+		EnvVar:   common.PrefixEnvVar(EnvVarPrefix, "METRICS_DB_SIZE_POLL_PERIOD"),
+	}
 	// The files for encrypted private keys.
 	BlsKeyFileFlag = cli.StringFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "bls-key-file"),
@@ -384,6 +392,7 @@ var optionalFlags = []cli.Flag{
 	ChunkDownloadTimeoutFlag,
 	PprofHttpPort,
 	EnablePprof,
+	MetricsDBSizePollPeriodFlag,
 }
 
 func init() {

node/grpc/metrics_v2.go

@@ -0,0 +1,175 @@
+package grpc
+
+import (
+	"github.com/Layr-Labs/eigensdk-go/logging"
+	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"google.golang.org/grpc"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"time"
+)
+
+const namespace = "eigenda_node"
+
+// MetricsV2 encapsulates metrics for the v2 DA node.
+type MetricsV2 struct {
+	logger logging.Logger
+
+	registry         *prometheus.Registry
+	grpcServerOption grpc.ServerOption
+
+	storeChunksLatency  *prometheus.SummaryVec
+	storeChunksDataSize *prometheus.GaugeVec
+
+	getChunksLatency  *prometheus.SummaryVec
+	getChunksDataSize *prometheus.GaugeVec
+
+	dbSize           *prometheus.GaugeVec
+	dbSizePollPeriod time.Duration
+	dbDir            string
+	isAlive          *atomic.Bool
+}
+
+// NewV2Metrics creates a new MetricsV2 instance. dbSizePollPeriod is the period at which the database size is polled.
+// If set to 0, the database size is not polled.
+func NewV2Metrics(
+	logger logging.Logger,
+	registry *prometheus.Registry,
+	dbDir string,
+	dbSizePollPeriod time.Duration) (*MetricsV2, error) {
+
+	// These should be re-enabled once the legacy v1 metrics are removed.
+	//registry.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
+	//registry.MustRegister(collectors.NewGoCollector())
+
+	grpcMetrics := grpcprom.NewServerMetrics()
+	registry.MustRegister(grpcMetrics)
+	grpcServerOption := grpc.UnaryInterceptor(
+		grpcMetrics.UnaryServerInterceptor(),
+	)
+
+	storeChunksLatency := promauto.With(registry).NewSummaryVec(
+		prometheus.SummaryOpts{
+			Namespace:  namespace,
+			Name:       "store_chunks_latency_ms",
+			Help:       "The latency of a StoreChunks() RPC call.",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{},
+	)
+
+	storeChunksDataSize := promauto.With(registry).NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "store_chunks_data_size_bytes",
+			Help:      "The size of the data requested to be stored by StoreChunks() RPC calls.",
+		},
+		[]string{},
+	)
+
+	getChunksLatency := promauto.With(registry).NewSummaryVec(
+		prometheus.SummaryOpts{
+			Namespace:  namespace,
+			Name:       "get_chunks_latency_ms",
+			Help:       "The latency of a GetChunks() RPC call.",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{},
+	)
+
+	getChunksDataSize := promauto.With(registry).NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "get_chunks_data_size_bytes",
+			Help:      "The size of the data requested to be retrieved by GetChunks() RPC calls.",
+		},
+		[]string{},
+	)
+
+	dbSize := promauto.With(registry).NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      "db_size_bytes",
+			Help:      "The size of the leveldb database.",
+		},
+		[]string{},
+	)
+	isAlive := &atomic.Bool{}
+	isAlive.Store(true)
+
+	return &MetricsV2{
+		logger:              logger,
+		registry:            registry,
+		grpcServerOption:    grpcServerOption,
+		storeChunksLatency:  storeChunksLatency,
+		storeChunksDataSize: storeChunksDataSize,
+		getChunksLatency:    getChunksLatency,
+		getChunksDataSize:   getChunksDataSize,
+		dbSize:              dbSize,
+		dbSizePollPeriod:    dbSizePollPeriod,
+		dbDir:               dbDir,
+		isAlive:             isAlive,
+	}, nil
+}
+
+// Start starts the metrics server.
+func (m *MetricsV2) Start() {
+	if m.dbSizePollPeriod.Nanoseconds() == 0 {
+		return
+	}
+	go func() {
+		ticker := time.NewTicker(m.dbSizePollPeriod)
+
+		for m.isAlive.Load() {
+			var size int64
+			err := filepath.Walk(m.dbDir, func(_ string, info os.FileInfo, err error) error {
+				if err != nil {
+					return err
+				}
+				if !info.IsDir() {
+					size += info.Size()
+				}
+				return err
+			})
+
+			if err != nil {
+				m.logger.Errorf("failed to get database size: %v", err)
+			} else {
+				m.dbSize.WithLabelValues().Set(float64(size))
+			}
+			<-ticker.C
+		}
+	}()
+
+}
+
+// Stop stops the metrics server.
+func (m *MetricsV2) Stop() {
+	m.isAlive.Store(false)
+}
+
+// GetGRPCServerOption returns the gRPC server option that enables automatic GRPC metrics collection.
+func (m *MetricsV2) GetGRPCServerOption() grpc.ServerOption {
+	return m.grpcServerOption
+}
+
+func (m *MetricsV2) ReportStoreChunksLatency(latency time.Duration) {
+	m.storeChunksLatency.WithLabelValues().Observe(
+		float64(latency.Nanoseconds()) / float64(time.Millisecond))
+}
+
+func (m *MetricsV2) ReportStoreChunksDataSize(size uint64) {
+	m.storeChunksDataSize.WithLabelValues().Set(float64(size))
+}
+
+func (m *MetricsV2) ReportGetChunksLatency(latency time.Duration) {
+	m.getChunksLatency.WithLabelValues().Observe(
+		float64(latency.Nanoseconds()) / float64(time.Millisecond))
+}
+
+func (m *MetricsV2) ReportGetChunksDataSize(size int) {
+	m.getChunksDataSize.WithLabelValues().Set(float64(size))
+}

node/grpc/run.go

@@ -24,6 +24,8 @@ func RunServers(serverV1 *Server, serverV2 *ServerV2, config *node.Config, logge
 		return errors.New("node V2 server is not configured")
 	}
 
+	serverV2.metrics.Start()
+
 	go func() {
 		for {
 			addr := fmt.Sprintf("%s:%s", localhost, config.InternalDispersalPort)
@@ -33,7 +35,7 @@ func RunServers(serverV1 *Server, serverV2 *ServerV2, config *node.Config, logge
 			}
 
 			opt := grpc.MaxRecvMsgSize(60 * 1024 * 1024 * 1024) // 60 GiB
-			gs := grpc.NewServer(opt)
+			gs := grpc.NewServer(opt, serverV2.metrics.GetGRPCServerOption())
 
 			// Register reflection service on gRPC server
 			// This makes "grpcurl -plaintext localhost:9000 list" command work
@@ -60,7 +62,7 @@ func RunServers(serverV1 *Server, serverV2 *ServerV2, config *node.Config, logge
 			}
 
 			opt := grpc.MaxRecvMsgSize(1024 * 1024 * 300) // 300 MiB
-			gs := grpc.NewServer(opt)
+			gs := grpc.NewServer(opt, serverV2.metrics.GetGRPCServerOption())
 
 			// Register reflection service on gRPC server
 			// This makes "grpcurl -plaintext localhost:9000 list" command work