Add new TXM and Block History Estimator metrics

integration-tests/relayinterface/buffer_payload_test.go

@@ -66,7 +66,8 @@ func Test_BufferPayload(t *testing.T) {
 	}, nil).Maybe()
 
 	lgr, logs := logger.TestObserved(t, zapcore.DebugLevel)
-	txmgr := txm.NewTxm("localnet", loader, nil, cfg, mkey, lgr)
+	txmgr, err := txm.NewTxm("localnet", loader, nil, cfg, mkey, lgr)
+	require.NoError(t, err)
 	err = txmgr.Start(t.Context())
 	require.NoError(t, err)
 

integration-tests/relayinterface/chain_components_test.go

@@ -1148,7 +1148,8 @@ func (h *helper) Init(t *testing.T) {
 		return sig[:]
 	}, nil)
 
-	txm := txm.NewTxm("localnet", loader, nil, cfg, mkey, lggr)
+	txm, err := txm.NewTxm("localnet", loader, nil, cfg, mkey, lggr)
+	require.NoError(t, err)
 	err = txm.Start(t.Context())
 	require.NoError(t, err)
 

integration-tests/relayinterface/lookups_test.go

@@ -571,7 +571,8 @@ func TestLookupTables(t *testing.T) {
 	mkey := keyMocks.NewSimpleKeystore(t)
 	lggr := logger.Test(t)
 
-	txm := txm.NewTxm("localnet", loader, nil, cfg, mkey, lggr)
+	txm, err := txm.NewTxm("localnet", loader, nil, cfg, mkey, lggr)
+	require.NoError(t, err)
 
 	cw, err := chainwriter.NewSolanaChainWriterService(logger.Test(t), multiClient, txm, nil, chainwriter.ChainWriterConfig{})
 	require.NoError(t, err)

integration-tests/smoke/log_poller_test.go

@@ -349,7 +349,7 @@ func (s *logSender) sendInstruction(
 	}
 
 	sig, err := s.client.SendTransactionWithOpts(
-		context.Background(),
+		ctx,
 		tx,
 		rpc.TransactionOpts{
 			PreflightCommitment: rpc.CommitmentConfirmed,

pkg/monitoring/exporter/feedbalances_test.go

@@ -20,7 +20,7 @@ func TestFeedBalances(t *testing.T) {
 	t.Parallel()
 
 	t.Run("it should export balance updates then clean up", func(t *testing.T) {
-		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+		ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
 		defer cancel()
 		mockMetrics := mocks.NewFeedBalances(t)
 		factory := NewFeedBalancesFactory(testutils.NewNullLogger(), mockMetrics)

pkg/monitoring/source_envelope_test.go

@@ -1,7 +1,6 @@
 package monitoring
 
 import (
-	"context"
 	"encoding/binary"
 	"math/big"
 	"testing"
@@ -162,7 +161,7 @@ func TestEnvelopeSource(t *testing.T) {
 	factory := NewEnvelopeSourceFactory(chainReader, lgr)
 	source, err := factory.NewSource(chainConfig, feedConfig)
 	require.NoError(t, err)
-	rawEnvelope, err := source.Fetch(context.Background())
+	rawEnvelope, err := source.Fetch(t.Context())
 	require.NoError(t, err)
 	envelope, ok := rawEnvelope.(commonMonitoring.Envelope)
 	require.True(t, ok)

pkg/solana/cache_test.go

@@ -2,7 +2,6 @@ package solana
 
 import (
 	"bytes"
-	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -108,7 +107,7 @@ func TestGetState(t *testing.T) {
 	reader := testSetupReader(t, mockServer.URL)
 	getReader := func() (client.AccountReader, error) { return reader, nil }
 	// happy path does not error (actual state decoding handled in types_test)
-	_, _, err := GetState(context.TODO(), getReader, solana.PublicKey{}, "")
+	_, _, err := GetState(t.Context(), getReader, solana.PublicKey{}, "")
 	require.NoError(t, err)
 }
 
@@ -135,17 +134,17 @@ func TestGetLatestTransmission(t *testing.T) {
 
 	reader := testSetupReader(t, mockServer.URL)
 	getReader := func() (client.AccountReader, error) { return reader, nil }
-	a, _, err := GetLatestTransmission(context.TODO(), getReader, solana.PublicKey{}, "")
+	a, _, err := GetLatestTransmission(t.Context(), getReader, solana.PublicKey{}, "")
 	assert.NoError(t, err)
 	assert.Equal(t, expectedTime, a.Timestamp)
 	assert.Equal(t, expectedAns, a.Data.String())
 
 	// fail if returned transmission header is too short
-	_, _, err = GetLatestTransmission(context.TODO(), getReader, solana.PublicKey{}, "")
+	_, _, err = GetLatestTransmission(t.Context(), getReader, solana.PublicKey{}, "")
 	assert.Error(t, err)
 
 	// fail if returned transmission is too short
-	_, _, err = GetLatestTransmission(context.TODO(), getReader, solana.PublicKey{}, "")
+	_, _, err = GetLatestTransmission(t.Context(), getReader, solana.PublicKey{}, "")
 	assert.Error(t, err)
 }
 
@@ -230,14 +229,14 @@ func TestNilPointerHandling(t *testing.T) {
 	getReader := func() (client.AccountReader, error) { return reader, nil }
 
 	// fail on get state query
-	_, _, err := GetState(context.TODO(), getReader, solana.PublicKey{}, "")
+	_, _, err := GetState(t.Context(), getReader, solana.PublicKey{}, "")
 	assert.EqualError(t, err, errString+"GetState.GetAccountInfoWithOpts")
 
 	// fail on transmissions header query
-	_, _, err = GetLatestTransmission(context.TODO(), getReader, solana.PublicKey{}, "")
+	_, _, err = GetLatestTransmission(t.Context(), getReader, solana.PublicKey{}, "")
 	assert.EqualError(t, err, errString+"GetLatestTransmission.GetAccountInfoWithOpts.Header")
 
 	passFirst = true // allow proper response for header query, fail on transmission
-	_, _, err = GetLatestTransmission(context.TODO(), getReader, solana.PublicKey{}, "")
+	_, _, err = GetLatestTransmission(t.Context(), getReader, solana.PublicKey{}, "")
 	assert.EqualError(t, err, errString+"GetLatestTransmission.GetAccountInfoWithOpts.Transmission")
 }

pkg/solana/chain.go

@@ -366,7 +366,11 @@ func newChain(id string, cfg *config.TOMLConfig, ks core.Keystore, lggr logger.L
 	}
 
 	ch.lp = logpoller.New(logger.Sugared(logger.Named(lggr, "LogPoller")), orm, ch.multiClient, cfg)
-	ch.txm = txm.NewTxm(ch.id, tc, sendTx, cfg, ks, lggr)
+	solTxm, err := txm.NewTxm(ch.id, tc, sendTx, cfg, ks, lggr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize solana txm: %w", err)
+	}
+	ch.txm = solTxm
 	ch.balanceMonitor = monitor.NewBalanceMonitor(ch.id, cfg, lggr, ks, bc)
 	return &ch, nil
 }

pkg/solana/config_tracker_test.go

@@ -1,7 +1,6 @@
 package solana
 
 import (
-	"context"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -19,7 +18,7 @@ func TestLatestBlockHeight(t *testing.T) {
 	}))
 	defer mockServer.Close()
 
-	ctx := context.Background()
+	ctx := t.Context()
 	c := &ConfigTracker{
 		getReader: func() (client.Reader, error) { return testSetupReader(t, mockServer.URL), nil },
 	}

pkg/solana/fees/block_history.go

@@ -7,14 +7,26 @@ import (
 	"slices"
 	"sync"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/utils/mathutil"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/client"
 	"github.com/smartcontractkit/chainlink-solana/pkg/solana/config"
 )
 
+var (
+	promBHEComputeUnitPrice = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "solana_bhe_compute_unit_price",
+		Help: "The compute unit price determined by the Solana block history estimator",
+	}, []string{"chainID"})
+)
+
 var _ Estimator = &blockHistoryEstimator{}
 
 var errNoComputeUnitPriceCollected = fmt.Errorf("no compute unit prices collected")
@@ -33,6 +45,10 @@ type blockHistoryEstimator struct {
 
 	cache   blockMedianCache
 	cacheMu sync.RWMutex
+
+	// metrics
+	computeUnitPrice metric.Int64Gauge
+	chainID          string
 }
 
 type blockMedianCache struct {
@@ -43,18 +59,25 @@ type blockMedianCache struct {
 // NewBlockHistoryEstimator creates a new fee estimator that parses historical fees from a fetched block
 // Note: getRecentPrioritizationFees is not used because it provides the lowest prioritization fee for an included tx in the block
 // which is not effective enough for increasing the chances of block inclusion
-func NewBlockHistoryEstimator(c func(context.Context) (client.ReaderWriter, error), cfg config.Config, lgr logger.Logger) (*blockHistoryEstimator, error) {
+func NewBlockHistoryEstimator(c func(context.Context) (client.ReaderWriter, error), cfg config.Config, lgr logger.Logger, chainID string) (*blockHistoryEstimator, error) {
 	if cfg.BlockHistorySize() < 1 {
 		return nil, fmt.Errorf("invalid block history depth: %d", cfg.BlockHistorySize())
 	}
 
+	computeUnitPrice, err := beholder.GetMeter().Int64Gauge("solana_bhe_compute_unit_price")
+	if err != nil {
+		return nil, fmt.Errorf("failed to register solana block history estimator average compute unit price metric: %w", err)
+	}
+
 	return &blockHistoryEstimator{
-		chStop: make(chan struct{}),
-		client: c,
-		cfg:    cfg,
-		lgr:    lgr,
-		price:  cfg.ComputeUnitPriceDefault(), // use default value
-		cache:  blockMedianCache{storedBlockRange: make([]uint64, 0, cfg.BlockHistorySize()), medianMap: make(map[uint64]ComputeUnitPrice, cfg.BlockHistorySize())},
+		chStop:           make(chan struct{}),
+		client:           c,
+		cfg:              cfg,
+		lgr:              lgr,
+		price:            cfg.ComputeUnitPriceDefault(), // use default value
+		cache:            blockMedianCache{storedBlockRange: make([]uint64, 0, cfg.BlockHistorySize()), medianMap: make(map[uint64]ComputeUnitPrice, cfg.BlockHistorySize())},
+		computeUnitPrice: computeUnitPrice,
+		chainID:          chainID,
 	}, nil
 }
 
@@ -120,7 +143,7 @@ func (bhe *blockHistoryEstimator) calculatePrice(ctx context.Context) error {
 		if err := bhe.populateCache(ctx, bhe.cfg.BlockHistoryBatchLoadSize(), bhe.cfg.BlockHistorySize()); err != nil {
 			return fmt.Errorf("failed to populate cache: %w", err)
 		}
-		return bhe.calculatePriceFromMultipleBlocks(bhe.cfg.BlockHistorySize())
+		return bhe.calculatePriceFromMultipleBlocks(ctx, bhe.cfg.BlockHistorySize())
 	default:
 		return bhe.calculatePriceFromLatestBlock(ctx)
 	}
@@ -166,10 +189,13 @@ func (bhe *blockHistoryEstimator) calculatePriceFromLatestBlock(ctx context.Cont
 		"count", len(feeData.Prices),
 	)
 
+	// Record the compute unit price for prometheus and beholder metrics
+	bhe.recordComputeUnitPrice(ctx, v)
+
 	return nil
 }
 
-func (bhe *blockHistoryEstimator) calculatePriceFromMultipleBlocks(desiredBlockCount uint64) error {
+func (bhe *blockHistoryEstimator) calculatePriceFromMultipleBlocks(ctx context.Context, desiredBlockCount uint64) error {
 	bhe.cacheMu.RLock()
 	defer bhe.cacheMu.RUnlock()
 	blockMedians := make([]ComputeUnitPrice, 0, desiredBlockCount)
@@ -199,6 +225,9 @@ func (bhe *blockHistoryEstimator) calculatePriceFromMultipleBlocks(desiredBlockC
 		"numBlocks", len(blockMedians),
 	)
 
+	// Record the compute unit price for prometheus and beholder metrics
+	bhe.recordComputeUnitPrice(ctx, avgOfMedians)
+
 	return nil
 }
 
@@ -310,3 +339,8 @@ func (bhe *blockHistoryEstimator) populateCache(ctx context.Context, loadBatch,
 	bhe.cache.storedBlockRange = bhe.cache.storedBlockRange[excessBlocks:]
 	return nil
 }
+
+func (bhe *blockHistoryEstimator) recordComputeUnitPrice(ctx context.Context, avgOfMedians ComputeUnitPrice) {
+	promBHEComputeUnitPrice.WithLabelValues(bhe.chainID).Set(float64(avgOfMedians))
+	bhe.computeUnitPrice.Record(ctx, int64(avgOfMedians), metric.WithAttributes(attribute.String("chainID", bhe.chainID))) //nolint:gosec // compute unit prices cannot exceed int64 max
+}