metrics.go

package submitter

import (
	"context"
	"fmt"
	"github.com/cornelk/hashmap"
	"github.com/synapsecns/sanguine/core/metrics"
	"github.com/synapsecns/sanguine/ethergo/signer/signer"
	"github.com/synapsecns/sanguine/ethergo/submitter/db"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
	"math/big"
	"time"
)

const meterName = "github.com/synapsecns/sanguine/ethergo/submitter"

// 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 metrics.go -s otelRecorder -i iOtelRecorder -p submitter -o metrics_generated.go -c "autogenerated file"
type otelRecorder struct {
	metrics metrics.Handler
	// meter is the metrics meter.
	meter metric.Meter
	// confirmedQueueGauge is the gauge for the confirmed queue.
	confirmedQueueGauge metric.Int64ObservableGauge
	// oldestPendingGauge is the gauge for the oldest pending transaction.
	oldestPendingGauge metric.Float64ObservableGauge
	// numPendingGauge is the gauge for the number of pending transactions.
	numPendingGauge metric.Int64ObservableGauge
	// nonceGauge is the gauge for the current nonce.
	nonceGauge metric.Int64ObservableGauge
	// gasBalanceGauge is the gauge for the gas balance.
	gasBalanceGauge metric.Float64ObservableGauge
	// numPendingTxes is used for metrics.
	// note: numPendingTxes will stop counting at DefaultMaxResultsPerChain.
	numPendingTxes *hashmap.Map[uint32, int]
	// currentNonces is used for metrics.
	// chainID -> nonce
	currentNonces *hashmap.Map[uint32, uint64]
	// currentGasBalance is used for metrics.
	// chainID -> balance
	currentGasBalances *hashmap.Map[uint32, *big.Int]
	// oldestPendingPerChain is the oldest pending transaction.
	// chainID -> time
	oldestPendingPerChain *hashmap.Map[uint32, time.Duration]
	// confirmedQueueCount is the count of the confirmed queue.
	confirmedQueueCount *hashmap.Map[uint32, int]
	// signer is the signer for signing transactions.
	signer signer.Signer
}

// nolint: cyclop
func newOtelRecorder(meterHandler metrics.Handler, signerT signer.Signer) (_ iOtelRecorder, err error) {
	or := otelRecorder{
		metrics:               meterHandler,
		meter:                 meterHandler.Meter(meterName),
		numPendingTxes:        hashmap.New[uint32, int](),
		currentNonces:         hashmap.New[uint32, uint64](),
		currentGasBalances:    hashmap.New[uint32, *big.Int](),
		oldestPendingPerChain: hashmap.New[uint32, time.Duration](),
		confirmedQueueCount:   hashmap.New[uint32, int](),
		signer:                signerT,
	}

	or.numPendingGauge, err = or.meter.Int64ObservableGauge("num_pending_txes")
	if err != nil {
		return nil, fmt.Errorf("could not create num pending txes gauge: %w", err)
	}

	_, err = or.meter.RegisterCallback(or.recordNumPending, or.numPendingGauge)
	if err != nil {
		return nil, fmt.Errorf("could not register callback for num pending txes gauge: %w", err)
	}

	or.nonceGauge, err = or.meter.Int64ObservableGauge("current_nonce")
	if err != nil {
		return nil, fmt.Errorf("could not create nonce gauge: %w", err)
	}

	_, err = or.meter.RegisterCallback(or.recordNonces, or.nonceGauge)
	if err != nil {
		return nil, fmt.Errorf("could not register callback for nonce gauge: %w", err)
	}

	or.gasBalanceGauge, err = or.meter.Float64ObservableGauge("gas_balance")
	if err != nil {
		return nil, fmt.Errorf("could not create gas balance gauge: %w", err)
	}

	_, err = or.meter.RegisterCallback(or.recordBalance, or.gasBalanceGauge)
	if err != nil {
		return nil, fmt.Errorf("could not register callback for gas balance gauge: %w", err)
	}

	or.oldestPendingGauge, err = or.meter.Float64ObservableGauge("oldest_pending_tx", metric.WithUnit("s"))
	if err != nil {
		return nil, fmt.Errorf("could not create oldest pending gauge: %w", err)
	}

	_, err = or.meter.RegisterCallback(or.recordOldestPendingTx, or.oldestPendingGauge)
	if err != nil {
		return nil, fmt.Errorf("could not register callback for oldest pending gauge: %w", err)
	}

	or.confirmedQueueGauge, err = or.meter.Int64ObservableGauge("confirmed_queue")
	if err != nil {
		return nil, fmt.Errorf("could not create confirmed queue gauge: %w", err)
	}

	_, err = or.meter.RegisterCallback(or.recordConfirmedQueue, or.confirmedQueueGauge)
	if err != nil {
		return nil, fmt.Errorf("could not register callback for confirmed queue gauge: %w", err)
	}

	return &or, nil
}

func (o *otelRecorder) recordNumPending(_ context.Context, observer metric.Observer) (err error) {
	if o.metrics == nil || o.numPendingGauge == nil || o.numPendingTxes == nil {
		return nil
	}

	o.numPendingTxes.Range(func(chainID uint32, numPending int) bool {
		opts := metric.WithAttributes(
			attribute.Int(metrics.ChainID, int(chainID)),
			attribute.String("wallet", o.signer.Address().Hex()),
		)
		observer.ObserveInt64(o.numPendingGauge, int64(numPending), opts)

		return true
	})

	return nil
}

func (o *otelRecorder) recordNonces(_ context.Context, observer metric.Observer) (err error) {
	if o.metrics == nil || o.nonceGauge == nil || o.currentNonces == nil {
		return nil
	}

	o.currentNonces.Range(func(chainID uint32, nonce uint64) bool {
		opts := metric.WithAttributes(
			attribute.Int(metrics.ChainID, int(chainID)),
			attribute.String("wallet", o.signer.Address().Hex()),
		)
		observer.ObserveInt64(o.nonceGauge, int64(nonce), opts)
		return true
	})

	return nil
}

func (o *otelRecorder) recordConfirmedQueue(_ context.Context, observer metric.Observer) (err error) {
	if o.metrics == nil || o.nonceGauge == nil || o.currentNonces == nil {
		return nil
	}

	o.confirmedQueueCount.Range(func(chainID uint32, queueSize int) bool {
		opts := metric.WithAttributes(
			attribute.Int(metrics.ChainID, int(chainID)),
			attribute.String("wallet", o.signer.Address().Hex()),
		)
		observer.ObserveInt64(o.confirmedQueueGauge, int64(queueSize), opts)
		return true
	})

	return nil
}

func (o *otelRecorder) recordBalance(_ context.Context, observer metric.Observer) (err error) {
	if o.metrics == nil || o.gasBalanceGauge == nil {
		return nil
	}

	o.currentGasBalances.Range(func(chainID uint32, gasPrice *big.Int) bool {
		opts := metric.WithAttributes(
			attribute.Int(metrics.ChainID, int(chainID)),
			attribute.String("wallet", o.signer.Address().Hex()),
		)

		observer.ObserveFloat64(o.gasBalanceGauge, toFloat(gasPrice), opts)
		return true
	})

	return nil
}

func (o *otelRecorder) recordOldestPendingTx(_ context.Context, observer metric.Observer) (err error) {
	if o.metrics == nil || o.oldestPendingGauge == nil {
		return nil
	}

	o.oldestPendingPerChain.Range(func(chainID uint32, oldestPendingTx time.Duration) bool {
		opts := metric.WithAttributes(
			attribute.Int(metrics.ChainID, int(chainID)),
			attribute.String("wallet", o.signer.Address().Hex()),
		)
		observer.ObserveFloat64(o.oldestPendingGauge, oldestPendingTx.Seconds(), opts)

		return true
	})

	return nil
}

// RecordNonceForChain sets the nonce for a chain.
func (o *otelRecorder) RecordNonceForChain(chainID uint32, nonce uint64) {
	o.currentNonces.Set(chainID, nonce)
}

// RecordGasBalanceForChain sets the gas balance for a chain.
func (o *otelRecorder) RecordGasBalanceForChain(chainID uint32, balance *big.Int) {
	o.currentGasBalances.Set(chainID, balance)
}

// RecordOldestPendingTx sets the oldest pending tx.
func (o *otelRecorder) RecordOldestPendingTx(chainID uint32, lastPending time.Duration) {
	o.oldestPendingPerChain.Set(chainID, lastPending)
}

// RecordNumPendingTxes sets the number of pending txes.
func (o *otelRecorder) RecordNumPendingTxes(chainID uint32, numPending int) {
	o.numPendingTxes.Set(chainID, numPending)
}

// RecordConfirmedQueue sets the confirmed queue count.
func (o *otelRecorder) RecordConfirmedQueue(chainID uint32, queueSize int) {
	o.confirmedQueueCount.Set(chainID, queueSize)
}

// HasNonceForChain checks if a nonce exists for a chain.
func (o *otelRecorder) HasNonceForChain(chainID uint32) bool {
	_, ok := o.currentNonces.Get(chainID)
	return ok
}

// HasGasBalanceForChain checks if a gas balance exists for a chain.
func (o *otelRecorder) HasGasBalanceForChain(chainID uint32) bool {
	_, ok := o.currentGasBalances.Get(chainID)
	return ok
}

// fetchOldestPendingTx fetches the oldest pending tx in the queue.
func fetchOldestPendingTx(txes []db.TX, nonce uint64) time.Time {
	oldestPendingTx := time.Now()
	for _, tx := range txes {
		if tx.Nonce() >= nonce {
			continue
		}

		if tx.CreationTime().Before(oldestPendingTx) {
			oldestPendingTx = tx.CreationTime()
		}
	}

	return oldestPendingTx
}

// calculatePendingTxes calculates the number of pending txes in the queue.
func calculatePendingTxes(txes []db.TX, nonce uint64) int {
	realPendingCount := 0
	for _, tx := range txes {
		// current nonce is going to be transaction count (nonces are offset by -1 since first nonce is 0)
		// so we use equal to here
		if tx.Nonce() >= nonce {
			realPendingCount++
		}
	}

	return realPendingCount
}