Relay metrics (Layr-Labs#938)

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

common/metrics/metrics.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_server.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/mock_metrics.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/test/main.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",

metrics.md

@@ -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)
 

operators/churner/metrics.go

@@ -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

relay/blob_provider.go

@@ -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)
 	}

relay/blob_provider_test.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++ {

relay/cache/cache_accessor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"golang.org/x/sync/semaphore"
 	"sync"
+	"time"
 )
 
 // CacheAccessor is an interface for accessing a resource that is cached. It assumes that cache misses
@@ -55,17 +56,24 @@ type cacheAccessor[K comparable, V any] struct {
 
 	// accessor is the function used to fetch values that are not in the cache.
 	accessor Accessor[K, V]
+
+	// metrics is used to record metrics about the cache accessor's performance.
+	metrics *CacheAccessorMetrics
 }
 
-// NewCacheAccessor creates a new CacheAccessor. The cacheSize parameter specifies the maximum number of items
-// that can be stored in the cache. The concurrencyLimit parameter specifies the maximum number of concurrent
-// lookups that can be in progress at any given time. If a greater number of lookups are requested, the excess
-// lookups will block until a lookup completes. If concurrencyLimit is zero, then no limits are imposed. The accessor
-// parameter is the function used to fetch values that are not in the cache.
+// NewCacheAccessor creates a new CacheAccessor.
+//
+// The concurrencyLimit parameter specifies the maximum number of concurrent lookups that can be in progress at any
+// given time. If a greater number of lookups are requested, the excess lookups will block until a lookup completes.
+// If concurrencyLimit is zero, then no limits are imposed. The accessor parameter is the function used to fetch values that are not in the cache.
+//
+// If metrics is not nil, it will be used to record metrics about the cache accessor's performance.
+// If nil, no metrics will be recorded.
 func NewCacheAccessor[K comparable, V any](
 	cache Cache[K, V],
 	concurrencyLimit int,
-	accessor Accessor[K, V]) (CacheAccessor[K, V], error) {
+	accessor Accessor[K, V],
+	metrics *CacheAccessorMetrics) (CacheAccessor[K, V], error) {
 
 	lookupsInProgress := make(map[K]*accessResult[V])
 
@@ -79,6 +87,7 @@ func NewCacheAccessor[K comparable, V any](
 		concurrencyLimiter: concurrencyLimiter,
 		accessor:           accessor,
 		lookupsInProgress:  lookupsInProgress,
+		metrics:            metrics,
 	}, nil
 }
 
@@ -97,6 +106,10 @@ func (c *cacheAccessor[K, V]) Get(ctx context.Context, key K) (V, error) {
 	v, ok := c.cache.Get(key)
 	if ok {
 		c.cacheLock.Unlock()
+
+		if c.metrics != nil {
+			c.metrics.cacheHits.Increment()
+		}
 		return v, nil
 	}
 
@@ -109,6 +122,10 @@ func (c *cacheAccessor[K, V]) Get(ctx context.Context, key K) (V, error) {
 
 	c.cacheLock.Unlock()
 
+	if c.metrics != nil {
+		c.metrics.cacheMisses.Increment()
+	}
+
 	if alreadyLoading {
 		// The result is being fetched on another goroutine. Wait for it to finish.
 		return c.waitForResult(ctx, result)
@@ -150,11 +167,30 @@ func (c *cacheAccessor[K, V]) fetchResult(ctx context.Context, key K, result *ac
 			<-c.concurrencyLimiter
 		}
 
+		if c.metrics != nil {
+			start := time.Now()
+			defer func() {
+				c.metrics.cacheMissLatency.ReportLatency(time.Since(start))
+			}()
+		}
+
 		c.cacheLock.Lock()
 
 		// Update the cache if the fetch was successful.
 		if err == nil {
 			c.cache.Put(key, value)
+
+			if c.metrics != nil {
+				size := c.cache.Size()
+				weight := c.cache.Weight()
+				c.metrics.size.Set(float64(size))
+				c.metrics.weight.Set(float64(weight))
+				var averageWeight float64
+				if size > 0 {
+					averageWeight = float64(weight) / float64(size)
+				}
+				c.metrics.averageWeight.Set(averageWeight)
+			}
 		}
 
 		// Provide the result to all other goroutines that may be waiting for it.