[rfq] metrics, hotfixes

ethergo/listener/listener.go

@@ -51,6 +51,8 @@
 	newBlockHandler NewBlockHandler
 	finalityMode    rpc.BlockNumber
 	blockWait       uint64
+	// otelRecorder is the recorder for the otel metrics.
+	otelRecorder iOtelRecorder
 }
 
 var (
@@ -77,6 +79,12 @@
 		option(c)
 	}
 
+	var err error
+	c.otelRecorder, err = newOtelRecorder(handler, int(c.chainID))
+	if err != nil {
+		return nil, fmt.Errorf("could not create otel recorder: %w", err)
+	}
+
 	return c, nil
 }
 
@@ -183,6 +191,7 @@
 	if err != nil {
 		return fmt.Errorf("could not put latest block: %w", err)
 	}
+	c.otelRecorder.RecordLastBlock(endBlock)
 
 	c.startBlock = lastUnconfirmedBlock
 	return nil

ethergo/listener/otel.go

@@ -0,0 +1,118 @@
+package listener
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/synapsecns/sanguine/core/metrics"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const meterName = "github.com/synapsecns/sanguine/ethergo/listener"
+
+// 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 listener -o otel_generated.go -c "autogenerated file"
+type otelRecorder struct {
+	metrics metrics.Handler
+	// meter is the metrics meter.
+	meter metric.Meter
+	// lastBlockGauge is the gauge for the last block.
+	lastBlockGauge metric.Int64ObservableGauge
+	// lastBlockAgeGauge is the gauge for the last block age.
+	lastBlockAgeGauge metric.Float64ObservableGauge
+	// lastBlock is the last block processed by the listener.
+	lastBlock *uint64
+	// lastBlockFetchTime is the time the last block was fetched (used to calculate last block age).
+	lastBlockFetchTime *time.Time
+	// chainID is the chain ID for the listener.
+	chainID int
+}
+
+func newOtelRecorder(meterHandler metrics.Handler, chainID int) (_ iOtelRecorder, err error) {
+	or := otelRecorder{
+		metrics:            meterHandler,
+		meter:              meterHandler.Meter(meterName),
+		lastBlock:          nil,
+		lastBlockFetchTime: nil,
+		chainID:            chainID,
+	}
+
+	or.lastBlockGauge, err = or.meter.Int64ObservableGauge("last_block")
+	if err != nil {
+		return nil, fmt.Errorf("could not create last block gauge")
+	}
+
+	or.lastBlockAgeGauge, err = or.meter.Float64ObservableGauge("last_block_age")
+	if err != nil {
+		return nil, fmt.Errorf("could not create last block age gauge")
+	}
+
+	_, err = or.meter.RegisterCallback(or.recordLastBlock, or.lastBlockGauge)
+	if err != nil {
+		return nil, fmt.Errorf("could not register callback for last block gauge")
+	}
+
+	_, err = or.meter.RegisterCallback(or.recordLastBlockAge, or.lastBlockAgeGauge)
+	if err != nil {
+		return nil, fmt.Errorf("could not register callback for last block age gauge")
+	}
+
+	return &or, nil
+}
+
+func (o *otelRecorder) recordLastBlock(_ context.Context, observer metric.Observer) (err error) {
+	if o.metrics == nil || o.lastBlockGauge == nil || o.lastBlock == nil {
+		return nil
+	}
+
+	opts := metric.WithAttributes(
+		attribute.Int(metrics.ChainID, o.chainID),
+	)
+	observer.ObserveInt64(o.lastBlockGauge, int64(*o.lastBlock), opts)
+
+	return nil
+}
+
+func (o *otelRecorder) recordLastBlockAge(_ context.Context, observer metric.Observer) (err error) {
+	if o.metrics == nil || o.lastBlockAgeGauge == nil || o.lastBlockFetchTime == nil {
+		return nil
+	}
+
+	age := time.Since(*o.lastBlockFetchTime).Seconds()
+	opts := metric.WithAttributes(
+		attribute.Int(metrics.ChainID, o.chainID),
+	)
+	observer.ObserveFloat64(o.lastBlockAgeGauge, age, opts)
+
+	return nil
+}
+
+// RecordLastBlock records the last block processed by the listener.
+func (o *otelRecorder) RecordLastBlock(lastBlock uint64) {
+	// verify if the last block has changed
+	var hasChanged bool
+	if o.lastBlock == nil {
+		hasChanged = true
+	} else {
+		hasChanged = *o.lastBlock != lastBlock
+	}
+	if !hasChanged {
+		return
+	}
+
+	// record the block
+	o.lastBlock = &lastBlock
+	fetchTime := time.Now()
+	o.lastBlockFetchTime = &fetchTime
+}

services/rfq/relayer/service/relayer.go

@@ -387,13 +387,7 @@
 		} else {
 			// process in parallel (new goroutine)
 			request := req // capture func literal
-			ok := r.semaphore.TryAcquire(1)
-			if !ok {
-				span.AddEvent("could not acquire semaphore", trace.WithAttributes(
-					attribute.String("transaction_id", hexutil.Encode(request.TransactionID[:])),
-				))
-				continue
-			}
+			err = r.semaphore.Acquire(ctx, 1)
 			if err != nil {
 				return fmt.Errorf("could not acquire semaphore: %w", err)
 			}