Relay metrics (#938)

Signed-off-by: Cody Littley <[email protected]>

Author: cody-littley

common/metrics/config.go

@@ -1,6 +1,9 @@
 package metrics
 
-import "time"
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"time"
+)
 
 // Metrics provides a convenient interface for reporting metrics.
 type Metrics interface {
@@ -62,6 +65,9 @@ type Metrics interface {
 		pollPeriod time.Duration,
 		source func() float64,
 		label ...any) error
+
+	// RegisterExternalMetrics registers prometheus collectors created outside the metrics framework.
+	RegisterExternalMetrics(collectors ...prometheus.Collector)
 }
 
 // Metric represents a metric that can be reported.

common/metrics/metrics.go

@@ -24,8 +24,8 @@ type metrics struct {
 	// logger is the logger used to log messages.
 	logger logging.Logger
 
-	// config is the configuration for the metrics.
-	config *Config
+	// namespace is prepended to all metric names.
+	namespace string
 
 	// registry is the prometheus registry used to report metrics.
 	registry *prometheus.Registry
@@ -55,13 +55,13 @@ type metrics struct {
 }
 
 // NewMetrics creates a new Metrics instance.
-func NewMetrics(logger logging.Logger, config *Config) Metrics {
+func NewMetrics(logger logging.Logger, namespace string, port int) Metrics {
 	reg := prometheus.NewRegistry()
 	reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
 	reg.MustRegister(collectors.NewGoCollector())
 
-	logger.Infof("Starting metrics server at port %d", config.HTTPPort)
-	addr := fmt.Sprintf(":%d", config.HTTPPort)
+	logger.Infof("Starting metrics server at port %d", port)
+	addr := fmt.Sprintf(":%d", port)
 	mux := http.NewServeMux()
 	mux.Handle("/metrics", promhttp.HandlerFor(
 		reg,
@@ -74,7 +74,7 @@ func NewMetrics(logger logging.Logger, config *Config) Metrics {
 
 	m := &metrics{
 		logger:       logger,
-		config:       config,
+		namespace:    namespace,
 		registry:     reg,
 		metricMap:    make(map[metricID]Metric),
 		isAlive:      atomic.Bool{},
@@ -198,7 +198,7 @@ func (m *metrics) NewLatencyMetric(
 	metric, err := newLatencyMetric(
 		m.logger,
 		m.registry,
-		m.config.Namespace,
+		m.namespace,
 		name,
 		description,
 		objectives,
@@ -238,7 +238,7 @@ func (m *metrics) NewCountMetric(
 	metric, err := newCountMetric(
 		m.logger,
 		m.registry,
-		m.config.Namespace,
+		m.namespace,
 		name, description,
 		labelTemplate)
 
@@ -287,7 +287,7 @@ func (m *metrics) newGaugeMetricUnsafe(
 	metric, err := newGaugeMetric(
 		m.logger,
 		m.registry,
-		m.config.Namespace,
+		m.namespace,
 		name,
 		unit,
 		description,
@@ -368,7 +368,7 @@ func (m *metrics) GenerateMetricsDocumentation() string {
 	}
 	slices.SortFunc(metricIDs, sortFunc)
 
-	sb.Write([]byte(fmt.Sprintf("# Metrics Documentation for namespace '%s'\n\n", m.config.Namespace)))
+	sb.Write([]byte(fmt.Sprintf("# Metrics Documentation for namespace '%s'\n\n", m.namespace)))
 	sb.Write([]byte(fmt.Sprintf("This documentation was automatically generated at time `%s`\n\n",
 		time.Now().Format(time.RFC3339))))
 
@@ -402,7 +402,7 @@ func (m *metrics) GenerateMetricsDocumentation() string {
 			sb.Write([]byte(fmt.Sprintf("| **Quantiles** | %s |\n", m.quantilesMap[*id])))
 		}
 		sb.Write([]byte(fmt.Sprintf("| **Fully Qualified Name** | `%s_%s_%s` |\n",
-			m.config.Namespace, id.name, id.unit)))
+			m.namespace, id.name, id.unit)))
 	}
 
 	return sb.String()
@@ -428,3 +428,7 @@ func (m *metrics) WriteMetricsDocumentation(fileName string) error {
 
 	return nil
 }
+
+func (m *metrics) RegisterExternalMetrics(collectors ...prometheus.Collector) {
+	m.registry.MustRegister(collectors...)
+}

common/metrics/metrics_server.go

@@ -1,6 +1,7 @@
 package metrics
 
 import (
+	"github.com/prometheus/client_golang/prometheus"
 	"time"
 )
 
@@ -62,6 +63,10 @@ func (m *mockMetrics) NewAutoGauge(
 	return nil
 }
 
+func (m *mockMetrics) RegisterExternalMetrics(collectors ...prometheus.Collector) {
+
+}
+
 var _ CountMetric = &mockCountMetric{}
 
 type mockCountMetric struct {

common/metrics/mock_metrics.go

@@ -24,18 +24,12 @@ type LabelType2 struct {
 }
 
 func main() {
-
-	metricsConfig := &metrics.Config{
-		Namespace: "test",
-		HTTPPort:  9101,
-	}
-
 	logger, err := common.NewLogger(common.DefaultLoggerConfig())
 	if err != nil {
 		panic(err)
 	}
 
-	metricsServer := metrics.NewMetrics(logger, metricsConfig)
+	metricsServer := metrics.NewMetrics(logger, "test", 9101)
 
 	l1, err := metricsServer.NewLatencyMetric(
 		"l1",

common/metrics/test/main.go

@@ -1,4 +1,5 @@
 # EigenDA Metrics Documentation
 
-- [churner](operators/churner/churner-metrics.md)
+- [churner](operators/churner/mdoc/churner-metrics.md)
+- [relay](relay/mdoc/relay-metrics.md)
 

metrics.md

@@ -65,10 +65,7 @@ func NewMetrics(httpPort int, logger logging.Logger) (*Metrics, error) {
 	reg.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
 	reg.MustRegister(collectors.NewGoCollector())
 
-	metricsServer := metrics.NewMetrics(logger, &metrics.Config{
-		Namespace: "eigenda_churner",
-		HTTPPort:  httpPort,
-	})
+	metricsServer := metrics.NewMetrics(logger, "eigenda_churner", httpPort)
 
 	numRequests, err := metricsServer.NewCountMetric(
 		"request",
@@ -100,7 +97,7 @@ func NewMetrics(httpPort int, logger logging.Logger) (*Metrics, error) {
 
 // WriteMetricsDocumentation writes the metrics for the churner to a markdown file.
 func (g *Metrics) WriteMetricsDocumentation() error {
-	return g.metricsServer.WriteMetricsDocumentation("operators/churner/churner-metrics.md")
+	return g.metricsServer.WriteMetricsDocumentation("operators/churner/mdoc/churner-metrics.md")
 }
 
 // ObserveLatency observes the latency of a stage

operators/churner/churner-metrics.md renamed to operators/churner/mdoc/churner-metrics.md

@@ -33,7 +33,8 @@ func newBlobProvider(
 	blobStore *blobstore.BlobStore,
 	blobCacheSize uint64,
 	maxIOConcurrency int,
-	fetchTimeout time.Duration) (*blobProvider, error) {
+	fetchTimeout time.Duration,
+	metrics *cache.CacheAccessorMetrics) (*blobProvider, error) {
 
 	server := &blobProvider{
 		ctx:          ctx,
@@ -42,9 +43,12 @@ func newBlobProvider(
 		fetchTimeout: fetchTimeout,
 	}
 
-	c := cache.NewFIFOCache[v2.BlobKey, []byte](blobCacheSize, computeBlobCacheWeight)
+	cacheAccessor, err := cache.NewCacheAccessor[v2.BlobKey, []byte](
+		cache.NewFIFOCache[v2.BlobKey, []byte](blobCacheSize, computeBlobCacheWeight),
+		maxIOConcurrency,
+		server.fetchBlob,
+		metrics)
 
-	cacheAccessor, err := cache.NewCacheAccessor[v2.BlobKey, []byte](c, maxIOConcurrency, server.fetchBlob)
 	if err != nil {
 		return nil, fmt.Errorf("error creating blob cache: %w", err)
 	}

operators/churner/metrics.go

@@ -41,7 +41,8 @@ func TestReadWrite(t *testing.T) {
 		blobStore,
 		1024*1024*32,
 		32,
-		10*time.Second)
+		10*time.Second,
+		nil)
 	require.NoError(t, err)
 
 	// Read the blobs back.
@@ -78,7 +79,8 @@ func TestNonExistentBlob(t *testing.T) {
 		blobStore,
 		1024*1024*32,
 		32,
-		10*time.Second)
+		10*time.Second,
+		nil)
 	require.NoError(t, err)
 
 	for i := 0; i < 10; i++ {