Meta Client: Add prometheus and beholder metrics (#263)

* add prometheus and beholder metrics

* remove prom metrics and simplify otel metrics

* simplify beholder event data

* remove beholder event

* update metrics to add some labels

* add bids histogram and error type counting metrics

* move startTime for more accurate endpoint latency measurement

* remove nil check for bids received record

Author: eduard-cl

pkg/txm/clientwrappers/dualbroadcast/meta_client.go

@@ -132,16 +132,23 @@ type MetaClient struct {
 	ks        MetaClientKeystore
 	customURL *url.URL
 	chainID   *big.Int
+	metrics   *MetaMetrics
 }
 
-func NewMetaClient(lggr logger.Logger, c MetaClientRPC, ks MetaClientKeystore, customURL *url.URL, chainID *big.Int) *MetaClient {
+func NewMetaClient(lggr logger.Logger, c MetaClientRPC, ks MetaClientKeystore, customURL *url.URL, chainID *big.Int) (*MetaClient, error) {
+	metrics, err := NewMetaMetrics(chainID.String())
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Meta metrics: %w", err)
+	}
+
 	return &MetaClient{
 		lggr:      logger.Sugared(logger.Named(lggr, "Txm.Txm.MetaClient")),
 		c:         c,
 		ks:        ks,
 		customURL: customURL,
 		chainID:   chainID,
-	}
+		metrics:   metrics,
+	}, nil
 }
 
 func (a *MetaClient) NonceAt(ctx context.Context, address common.Address, blockNumber *big.Int) (uint64, error) {
@@ -161,10 +168,12 @@ func (a *MetaClient) SendTransaction(ctx context.Context, tx *types.Transaction,
 	if meta != nil && meta.DualBroadcast != nil && *meta.DualBroadcast && !tx.IsPurgeable && meta.DualBroadcastParams != nil && meta.FwdrDestAddress != nil {
 		meta, err := a.SendRequest(ctx, tx, attempt, *meta.DualBroadcastParams, tx.ToAddress)
 		if err != nil {
+			a.metrics.RecordSendRequestError(ctx)
 			return fmt.Errorf("error sending request for transactionID(%d): %w", tx.ID, err)
 		}
 		if meta != nil {
 			if err := a.SendOperation(ctx, tx, attempt, *meta); err != nil {
+				a.metrics.RecordSendOperationError(ctx)
 				return fmt.Errorf("failed to send operation for transactionID(%d): %w", tx.ID, err)
 			}
 			return nil
@@ -315,12 +324,21 @@ func (a *MetaClient) SendRequest(parentCtx context.Context, tx *types.Transactio
 	}
 	req.Header.Add("Content-Type", "application/json")
 
+	// Start timing for endpoint latency measurement
+	startTime := time.Now()
 	resp, err := http.DefaultClient.Do(req)
+	latency := time.Since(startTime)
+
+	// Record latency
+	a.metrics.RecordLatency(ctx, latency)
 	if err != nil {
 		return nil, fmt.Errorf("failed to send POST request: %w", err)
 	}
 	defer resp.Body.Close()
 
+	// Record status code
+	a.metrics.RecordStatusCode(ctx, resp.StatusCode)
+
 	if resp.StatusCode != http.StatusOK {
 		return nil, fmt.Errorf("request %v failed with status: %d", req, resp.StatusCode)
 	}
@@ -338,6 +356,7 @@ func (a *MetaClient) SendRequest(parentCtx context.Context, tx *types.Transactio
 
 	if response.Error.ErrorMessage != "" {
 		if strings.Contains(response.Error.ErrorMessage, "no solver operations received") {
+			a.metrics.RecordBidsReceived(ctx, 0)
 			return nil, nil
 		}
 		return nil, errors.New(response.Error.ErrorMessage)
@@ -347,6 +366,9 @@ func (a *MetaClient) SendRequest(parentCtx context.Context, tx *types.Transactio
 		return nil, nil
 	}
 
+	// Record bid count (number of solver operations received)
+	a.metrics.RecordBidsReceived(ctx, len(response.Result.SOS))
+
 	if r, err := json.MarshalIndent(response.Result, "", "  "); err == nil {
 		a.lggr.Info("Response: ", string(r))
 	}

pkg/txm/clientwrappers/dualbroadcast/meta_metrics.go

@@ -0,0 +1,100 @@
+package dualbroadcast
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+)
+
+// MetaMetrics handles all Meta-related metrics via OTEL
+type MetaMetrics struct {
+	chainID           string
+	statusCodeCounter metric.Int64Counter
+	latencyHistogram  metric.Int64Histogram
+	bidHistogram      metric.Int64Histogram
+	errorCounter      metric.Int64Counter
+}
+
+// NewMetaMetrics creates a new MetaMetrics instance
+func NewMetaMetrics(chainID string) (*MetaMetrics, error) {
+	statusCodeCounter, err := beholder.GetMeter().Int64Counter("meta_endpoint_status_codes")
+	if err != nil {
+		return nil, err
+	}
+
+	latencyHistogram, err := beholder.GetMeter().Int64Histogram("meta_endpoint_latency")
+	if err != nil {
+		return nil, err
+	}
+
+	bidHistogram, err := beholder.GetMeter().Int64Histogram("meta_bids_per_transaction")
+	if err != nil {
+		return nil, err
+	}
+
+	errorCounter, err := beholder.GetMeter().Int64Counter("meta_errors")
+	if err != nil {
+		return nil, err
+	}
+
+	return &MetaMetrics{
+		chainID:           chainID,
+		statusCodeCounter: statusCodeCounter,
+		latencyHistogram:  latencyHistogram,
+		bidHistogram:      bidHistogram,
+		errorCounter:      errorCounter,
+	}, nil
+}
+
+// RecordStatusCode records the HTTP status code from Meta endpoint
+func (m *MetaMetrics) RecordStatusCode(ctx context.Context, statusCode int) {
+	m.statusCodeCounter.Add(ctx, 1,
+		metric.WithAttributes(
+			attribute.String("chainID", m.chainID),
+			attribute.String("statusCode", strconv.Itoa(statusCode)),
+		),
+	)
+}
+
+// RecordLatency records the latency of Meta endpoint requests
+func (m *MetaMetrics) RecordLatency(ctx context.Context, duration time.Duration) {
+	m.latencyHistogram.Record(ctx, duration.Milliseconds(),
+		metric.WithAttributes(
+			attribute.String("chainID", m.chainID),
+		),
+	)
+}
+
+// RecordBidsReceived records the distribution of bids per transaction
+func (m *MetaMetrics) RecordBidsReceived(ctx context.Context, bidCount int) {
+	m.bidHistogram.Record(ctx, int64(bidCount),
+		metric.WithAttributes(
+			attribute.String("chainID", m.chainID),
+		),
+	)
+}
+
+// RecordSendRequestError records errors from SendRequest method
+func (m *MetaMetrics) RecordSendRequestError(ctx context.Context) {
+	m.errorCounter.Add(ctx, 1,
+		metric.WithAttributes(
+			attribute.String("chainID", m.chainID),
+			attribute.String("errorType", "send_request"),
+		),
+	)
+}
+
+// RecordSendOperationError records errors from SendOperation method
+func (m *MetaMetrics) RecordSendOperationError(ctx context.Context) {
+	m.errorCounter.Add(ctx, 1,
+		metric.WithAttributes(
+			attribute.String("chainID", m.chainID),
+			attribute.String("errorType", "send_operation"),
+		),
+	)
+}

pkg/txm/clientwrappers/dualbroadcast/meta_metrics_test.go

@@ -0,0 +1,34 @@
+package dualbroadcast
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMetaMetrics(t *testing.T) {
+	chainID := "1"
+
+	t.Run("NewMetaMetrics", func(t *testing.T) {
+		metrics, err := NewMetaMetrics(chainID)
+		require.NoError(t, err)
+		assert.NotNil(t, metrics)
+		assert.Equal(t, chainID, metrics.chainID)
+	})
+
+	t.Run("RecordBasicMetrics", func(t *testing.T) {
+		metrics, err := NewMetaMetrics(chainID)
+		require.NoError(t, err)
+
+		ctx := t.Context()
+
+		// Test that these don't panic - all metrics methods
+		metrics.RecordStatusCode(ctx, 200)
+		metrics.RecordLatency(ctx, time.Millisecond*100)
+		metrics.RecordBidsReceived(ctx, 5)
+		metrics.RecordSendRequestError(ctx)
+		metrics.RecordSendOperationError(ctx)
+	})
+}

pkg/txm/clientwrappers/dualbroadcast/selector.go

@@ -12,11 +12,11 @@ import (
 	"github.com/smartcontractkit/chainlink-evm/pkg/txm"
 )
 
-func SelectClient(lggr logger.Logger, client client.Client, keyStore keys.ChainStore, url *url.URL, chainID *big.Int) txm.Client {
+func SelectClient(lggr logger.Logger, client client.Client, keyStore keys.ChainStore, url *url.URL, chainID *big.Int) (txm.Client, error) {
 	urlString := url.String()
 	switch {
 	case strings.Contains(urlString, "flashbots"):
-		return NewFlashbotsClient(client, keyStore, url)
+		return NewFlashbotsClient(client, keyStore, url), nil
 	default:
 		return NewMetaClient(lggr, client, keyStore, url, chainID)
 	}

pkg/txmgr/builder.go

@@ -148,7 +148,11 @@ func NewTxmV2(
 	}
 	var c txm.Client
 	if txmV2Config.DualBroadcast() != nil && *txmV2Config.DualBroadcast() && txmV2Config.CustomURL() != nil {
-		c = dualbroadcast.SelectClient(lggr, client, keyStore, txmV2Config.CustomURL(), chainID)
+		var err error
+		c, err = dualbroadcast.SelectClient(lggr, client, keyStore, txmV2Config.CustomURL(), chainID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create dual broadcast client: %w", err)
+		}
 	} else {
 		c = clientwrappers.NewChainClient(client)
 	}