Added detailed tests for observability layer

Fixing tests and adding missing changeset

.changeset/thick-mails-applaud.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+Simplify how token and gas prices are stored in the database - user upsert instead of insert/delete flow #db_update

core/services/ccip/observability.go

@@ -0,0 +1,115 @@
+package ccip
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+)
+
+var (
+	sqlLatencyBuckets = []float64{
+		float64(10 * time.Millisecond),
+		float64(20 * time.Millisecond),
+		float64(30 * time.Millisecond),
+		float64(40 * time.Millisecond),
+		float64(50 * time.Millisecond),
+		float64(70 * time.Millisecond),
+		float64(90 * time.Millisecond),
+		float64(100 * time.Millisecond),
+		float64(200 * time.Millisecond),
+		float64(300 * time.Millisecond),
+		float64(400 * time.Millisecond),
+		float64(500 * time.Millisecond),
+		float64(750 * time.Millisecond),
+		float64(1 * time.Second),
+	}
+	ccipQueryDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    "ccip_orm_query_duration",
+		Buckets: sqlLatencyBuckets,
+	}, []string{"query", "destChainSelector"})
+	ccipQueryDatasets = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "ccip_orm_dataset_size",
+	}, []string{"query", "destChainSelector"})
+)
+
+type observedORM struct {
+	ORM
+	queryDuration *prometheus.HistogramVec
+	datasetSize   *prometheus.GaugeVec
+}
+
+var _ ORM = (*observedORM)(nil)
+
+func NewObservedORM(ds sqlutil.DataSource, lggr logger.Logger) (*observedORM, error) {
+	delegate, err := NewORM(ds, lggr)
+	if err != nil {
+		return nil, err
+	}
+
+	return &observedORM{
+		ORM:           delegate,
+		queryDuration: ccipQueryDuration,
+		datasetSize:   ccipQueryDatasets,
+	}, nil
+}
+
+func (o *observedORM) GetGasPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]GasPrice, error) {
+	return withObservedQueryAndResults(o, "GetGasPricesByDestChain", destChainSelector, func() ([]GasPrice, error) {
+		return o.ORM.GetGasPricesByDestChain(ctx, destChainSelector)
+	})
+}
+
+func (o *observedORM) GetTokenPricesByDestChain(ctx context.Context, destChainSelector uint64) ([]TokenPrice, error) {
+	return withObservedQueryAndResults(o, "GetTokenPricesByDestChain", destChainSelector, func() ([]TokenPrice, error) {
+		return o.ORM.GetTokenPricesByDestChain(ctx, destChainSelector)
+	})
+}
+
+func (o *observedORM) UpsertGasPricesForDestChain(ctx context.Context, destChainSelector uint64, gasPrices []GasPrice) (int64, error) {
+	return withObservedQueryAndRowsAffected(o, "UpsertGasPricesForDestChain", destChainSelector, func() (int64, error) {
+		return o.ORM.UpsertGasPricesForDestChain(ctx, destChainSelector, gasPrices)
+	})
+}
+
+func (o *observedORM) UpsertTokenPricesForDestChain(ctx context.Context, destChainSelector uint64, tokenPrices []TokenPrice, interval time.Duration) (int64, error) {
+	return withObservedQueryAndRowsAffected(o, "UpsertTokenPricesForDestChain", destChainSelector, func() (int64, error) {
+		return o.ORM.UpsertTokenPricesForDestChain(ctx, destChainSelector, tokenPrices, interval)
+	})
+}
+
+func withObservedQueryAndRowsAffected(o *observedORM, queryName string, chainSelector uint64, query func() (int64, error)) (int64, error) {
+	rowsAffected, err := withObservedQuery(o, queryName, chainSelector, query)
+	if err == nil {
+		o.datasetSize.
+			WithLabelValues(queryName, strconv.FormatUint(chainSelector, 10)).
+			Set(float64(rowsAffected))
+	}
+	return rowsAffected, err
+}
+
+func withObservedQueryAndResults[T any](o *observedORM, queryName string, chainSelector uint64, query func() ([]T, error)) ([]T, error) {
+	results, err := withObservedQuery(o, queryName, chainSelector, query)
+	if err == nil {
+		o.datasetSize.
+			WithLabelValues(queryName, strconv.FormatUint(chainSelector, 10)).
+			Set(float64(len(results)))
+	}
+	return results, err
+}
+
+func withObservedQuery[T any](o *observedORM, queryName string, chainSelector uint64, query func() (T, error)) (T, error) {
+	queryStarted := time.Now()
+	defer func() {
+		o.queryDuration.
+			WithLabelValues(queryName, strconv.FormatUint(chainSelector, 10)).
+			Observe(float64(time.Since(queryStarted)))
+	}()
+	return query()
+}

core/services/ccip/observability_test.go

@@ -0,0 +1,94 @@
+package ccip
+
+import (
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	io_prometheus_client "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/assets"
+	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
+	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest"
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+)
+
+func Test_MetricsAreTrackedForAllMethods(t *testing.T) {
+	ctx := testutils.Context(t)
+	db := pgtest.NewSqlxDB(t)
+	ccipORM, err := NewObservedORM(db, logger.TestLogger(t))
+	require.NoError(t, err)
+
+	tokenPrices := []TokenPrice{
+		{
+			TokenAddr:  "0xA",
+			TokenPrice: assets.NewWei(big.NewInt(1e18)),
+		},
+		{
+			TokenAddr:  "0xB",
+			TokenPrice: assets.NewWei(big.NewInt(1e18)),
+		},
+	}
+	tokensUpserted, err := ccipORM.UpsertTokenPricesForDestChain(ctx, 100, tokenPrices, time.Second)
+	require.NoError(t, err)
+	assert.Equal(t, len(tokenPrices), int(tokensUpserted))
+	assert.Equal(t, len(tokenPrices), counterFromGaugeByLabels(ccipORM.datasetSize, "UpsertTokenPricesForDestChain", "100"))
+	assert.Equal(t, 0, counterFromGaugeByLabels(ccipORM.datasetSize, "UpsertTokenPricesForDestChain", "200"))
+
+	tokens, err := ccipORM.GetTokenPricesByDestChain(ctx, 100)
+	require.NoError(t, err)
+	assert.Equal(t, len(tokenPrices), len(tokens))
+	assert.Equal(t, len(tokenPrices), counterFromGaugeByLabels(ccipORM.datasetSize, "GetTokenPricesByDestChain", "100"))
+	assert.Equal(t, 1, counterFromHistogramByLabels(t, ccipORM.queryDuration, "GetTokenPricesByDestChain", "100"))
+
+	gasPrices := []GasPrice{
+		{
+			SourceChainSelector: 200,
+			GasPrice:            assets.NewWei(big.NewInt(1e18)),
+		},
+		{
+			SourceChainSelector: 201,
+			GasPrice:            assets.NewWei(big.NewInt(1e18)),
+		},
+		{
+			SourceChainSelector: 202,
+			GasPrice:            assets.NewWei(big.NewInt(1e18)),
+		},
+	}
+	gasUpserted, err := ccipORM.UpsertGasPricesForDestChain(ctx, 100, gasPrices)
+	require.NoError(t, err)
+	assert.Equal(t, len(gasPrices), int(gasUpserted))
+	assert.Equal(t, len(gasPrices), counterFromGaugeByLabels(ccipORM.datasetSize, "UpsertGasPricesForDestChain", "100"))
+	assert.Equal(t, 0, counterFromGaugeByLabels(ccipORM.datasetSize, "UpsertGasPricesForDestChain", "200"))
+
+	gas, err := ccipORM.GetGasPricesByDestChain(ctx, 100)
+	require.NoError(t, err)
+	assert.Equal(t, len(gasPrices), len(gas))
+	assert.Equal(t, len(gasPrices), counterFromGaugeByLabels(ccipORM.datasetSize, "GetGasPricesByDestChain", "100"))
+	assert.Equal(t, 1, counterFromHistogramByLabels(t, ccipORM.queryDuration, "GetGasPricesByDestChain", "100"))
+}
+
+func counterFromHistogramByLabels(t *testing.T, histogramVec *prometheus.HistogramVec, labels ...string) int {
+	observer, err := histogramVec.GetMetricWithLabelValues(labels...)
+	require.NoError(t, err)
+
+	metricCh := make(chan prometheus.Metric, 1)
+	observer.(prometheus.Histogram).Collect(metricCh)
+	close(metricCh)
+
+	metric := <-metricCh
+	pb := &io_prometheus_client.Metric{}
+	err = metric.Write(pb)
+	require.NoError(t, err)
+
+	return int(pb.GetHistogram().GetSampleCount())
+}
+
+func counterFromGaugeByLabels(gaugeVec *prometheus.GaugeVec, labels ...string) int {
+	value := testutil.ToFloat64(gaugeVec.WithLabelValues(labels...))
+	return int(value)
+}