RFQ Relayer: add request status count metric

services/rfq/relayer/reldb/base/quote.go

@@ -138,3 +138,35 @@
 	}
 	return status >= prevStatus
 }
+
+type statusCount struct {
+	Status int
+	Count  int64
+}
+
+// GetStatusCounts gets the counts of quote requests by status.
+func (s Store) GetStatusCounts(ctx context.Context, matchStatuses ...reldb.QuoteRequestStatus) (map[reldb.QuoteRequestStatus]int, error) {
+	inArgs := make([]int, len(matchStatuses))
+	for i := range matchStatuses {
+		inArgs[i] = int(matchStatuses[i].Int())
+	}
+
+	var results []statusCount
+	tx := s.DB().
+		WithContext(ctx).
+		Model(&RequestForQuote{}).
+		Select(fmt.Sprintf("%s, COUNT(*) as count", statusFieldName)).
+		Where(fmt.Sprintf("%s IN ?", statusFieldName), inArgs).
+		Group(statusFieldName).
+		Scan(&results)
+	if tx.Error != nil {
+		return nil, fmt.Errorf("could not get db results: %w", tx.Error)
+	}
+
+	statuses := make(map[reldb.QuoteRequestStatus]int)
+	for _, result := range results {
+		statuses[reldb.QuoteRequestStatus(result.Status)] = int(result.Count)
+	}
+
+	return statuses, nil
+}

services/rfq/relayer/reldb/db.go

@@ -48,6 +48,8 @@ type Reader interface {
 	GetRebalanceByID(ctx context.Context, rebalanceID string) (*Rebalance, error)
 	// GetDBStats gets the database stats.
 	GetDBStats(ctx context.Context) (*sql.DBStats, error)
+	// GetStatusCounts gets the counts of quote requests by status.
+	GetStatusCounts(ctx context.Context, matchStatuses ...QuoteRequestStatus) (map[QuoteRequestStatus]int, error)
 }
 
 // Service is the interface for the database service.

services/rfq/relayer/service/otel.go

@@ -0,0 +1,82 @@
+package service
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cornelk/hashmap"
+	"github.com/synapsecns/sanguine/core/metrics"
+	"github.com/synapsecns/sanguine/ethergo/signer/signer"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const meterName = "github.com/synapsecns/sanguine/services/rfq/relayer/service"
+
+// generate an interface for otelRecorder that exports the public method.
+// this allows us to avoid using recordX externally anad makes the package less confusing.
+//
+// =============================================================================
+// =============================================================================
+// IMPORTANT: DO NOT REMOVE THIS COMMENT.
+// NOTICE: PLEASE MAKE SURE YOU UPDATE BOTH THE DOCS AND THE GRAFANA DASHBOARD (IF NEEDED) AFTER UPDATING METRICS.
+// =============================================================================
+// =============================================================================
+//
+//go:generate go run github.com/vburenin/ifacemaker -f otel.go -s otelRecorder -i iOtelRecorder -p service -o otel_generated.go -c "autogenerated file"
+type otelRecorder struct {
+	metrics metrics.Handler
+	// meter is the metrics meter.
+	meter metric.Meter
+	// statusCountGauge is the gauge for the status.
+	statusCountGauge metric.Int64ObservableGauge
+	// statusCounts is used for metrics.
+	// status -> count
+	statusCounts *hashmap.Map[int, int]
+	// signer is the signer for signing transactions.
+	signer signer.Signer
+}
+
+func newOtelRecorder(meterHandler metrics.Handler, signer signer.Signer) (_ iOtelRecorder, err error) {
+	or := otelRecorder{
+		metrics:      meterHandler,
+		meter:        meterHandler.Meter(meterName),
+		statusCounts: hashmap.New[int, int](),
+		signer:       signer,
+	}
+
+	or.statusCountGauge, err = or.meter.Int64ObservableGauge("status_count")
+	if err != nil {
+		return nil, fmt.Errorf("could not create last block gauge")
+	}
+
+	_, err = or.meter.RegisterCallback(or.recordStatusCounts, or.statusCountGauge)
+	if err != nil {
+		return nil, fmt.Errorf("could not register callback for status gauge")
+	}
+
+	return &or, nil
+}
+
+func (o *otelRecorder) recordStatusCounts(_ context.Context, observer metric.Observer) (err error) {
+	if o.metrics == nil || o.statusCountGauge == nil || o.statusCounts == nil {
+		return nil
+	}
+
+	o.statusCounts.Range(func(status int, count int) bool {
+		opts := metric.WithAttributes(
+			attribute.Int("status", status),
+			attribute.String("wallet", o.signer.Address().Hex()),
+		)
+		observer.ObserveInt64(o.statusCountGauge, int64(count), opts)
+
+		return true
+	})
+
+	return nil
+}
+
+// RecordStatusCounts records the request status count.
+func (o *otelRecorder) RecordStatusCount(status, count int) {
+	o.statusCounts.Set(status, count)
+}

services/rfq/relayer/service/relayer.go

@@ -62,7 +62,8 @@
 	// semaphore is used to limit the number of concurrent requests
 	semaphore *semaphore.Weighted
 	// handlerMtx is used to synchronize handling of relay requests
-	handlerMtx mapmutex.StringMapMutex
+	handlerMtx   mapmutex.StringMapMutex
+	otelRecorder iOtelRecorder
 }
 
 var logger = log.Logger("relayer")
@@ -142,6 +143,11 @@
 		return nil, fmt.Errorf("could not get api server: %w", err)
 	}
 
+	otelRecorder, err := newOtelRecorder(metricHandler, sg)
+	if err != nil {
+		return nil, fmt.Errorf("could not get otel recorder: %w", err)
+	}
+
 	cache := ttlcache.New[common.Hash, bool](ttlcache.WithTTL[common.Hash, bool](time.Second * 30))
 	rel := Relayer{
 		db:             store,
@@ -159,6 +165,7 @@
 		apiClient:      apiClient,
 		semaphore:      semaphore.NewWeighted(maxConcurrentRequests),
 		handlerMtx:     mapmutex.NewStringMapMutex(),
+		otelRecorder:   otelRecorder,
 	}
 	return &rel, nil
 }
@@ -280,6 +287,14 @@
 		return nil
 	})
 
+	g.Go(func() error {
+		err = r.recordMetrics(ctx)
+		if err != nil {
+			return fmt.Errorf("could not record metrics: %w", err)
+		}
+		return nil
+	})
+
 	err = g.Wait()
 	if err != nil {
 		return fmt.Errorf("could not start: %w", err)
@@ -360,6 +375,25 @@
 	return nil
 }
 
+const defaultMetricsInterval = 10
+
+func (r *Relayer) recordMetrics(ctx context.Context) (err error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("could not record metrics: %w", ctx.Err())
+		case <-time.After(defaultMetricsInterval * time.Second):
+			statusCounts, err := r.db.GetStatusCounts(ctx, reldb.Seen, reldb.NotEnoughInventory, reldb.CommittedPending, reldb.CommittedConfirmed, reldb.RelayStarted, reldb.RelayCompleted, reldb.ProvePosting, reldb.ProvePosted, reldb.ClaimPending)
+			if err != nil {
+				return fmt.Errorf("could not get status counts: %w", err)
+			}
+			for status, count := range statusCounts {
+				r.otelRecorder.RecordStatusCount(int(status.Int()), count)
+			}
+		}
+	}
+}
+
 func (r *Relayer) processDB(ctx context.Context, serial bool, matchStatuses ...reldb.QuoteRequestStatus) (err error) {
 	ctx, span := r.metrics.Tracer().Start(ctx, "processDB", trace.WithAttributes(
 		attribute.Bool("serial", serial),