report observations exporter

cmd/monitoring/main.go

@@ -57,27 +57,51 @@ func main() {
 		return
 	}
 
+	// per-feed sources
 	feedBalancesSourceFactory := monitoring.NewFeedBalancesSourceFactory(
 		chainReader,
 		logger.With(log, "component", "source-feed-balances"),
 	)
+	txDetailsSourceFactory := monitoring.NewTxDetailsSourceFactory(
+		chainReader,
+		logger.With(log, "component", "source-tx-details"),
+	)
+	monitor.SourceFactories = append(monitor.SourceFactories,
+		feedBalancesSourceFactory,
+		txDetailsSourceFactory,
+	)
+
+	// network sources
 	nodeBalancesSourceFactory := monitoring.NewNodeBalancesSourceFactory(
 		chainReader,
 		logger.With(log, "component", "source-node-balances"),
 	)
-	monitor.SourceFactories = append(monitor.SourceFactories, feedBalancesSourceFactory)
-	monitor.NetworkSourceFactories = append(monitor.NetworkSourceFactories, nodeBalancesSourceFactory)
+	monitor.NetworkSourceFactories = append(monitor.NetworkSourceFactories,
+		nodeBalancesSourceFactory,
+	)
 
+	// per-feed exporters
 	feedBalancesExporterFactory := exporter.NewFeedBalancesFactory(
 		logger.With(log, "component", "solana-prom-exporter"),
 		metrics.NewFeedBalances(logger.With(log, "component", "solana-metrics")),
 	)
+	reportObservationsFactory := exporter.NewReportObservationsFactory(
+		logger.With(log, "component", "solana-prome-exporter"),
+		metrics.NewReportObservations(logger.With(log, "component", "solana-metrics")),
+	)
+	monitor.ExporterFactories = append(monitor.ExporterFactories,
+		feedBalancesExporterFactory,
+		reportObservationsFactory,
+	)
+
+	// network exporters
 	nodeBalancesExporterFactory := exporter.NewNodeBalancesFactory(
 		logger.With(log, "component", "solana-prom-exporter"),
 		metrics.NewNodeBalances,
 	)
-	monitor.ExporterFactories = append(monitor.ExporterFactories, feedBalancesExporterFactory)
-	monitor.NetworkExporterFactories = append(monitor.NetworkExporterFactories, nodeBalancesExporterFactory)
+	monitor.NetworkExporterFactories = append(monitor.NetworkExporterFactories,
+		nodeBalancesExporterFactory,
+	)
 
 	monitor.Run()
 	log.Infow("monitor stopped")

pkg/monitoring/exporter/reportobservations.go

@@ -0,0 +1,82 @@
+package exporter
+
+import (
+	"context"
+
+	commonMonitoring "github.com/smartcontractkit/chainlink-common/pkg/monitoring"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/metrics"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
+)
+
+func NewReportObservationsFactory(
+	log commonMonitoring.Logger,
+	metrics metrics.ReportObservations,
+) commonMonitoring.ExporterFactory {
+	return &reportObservationsFactory{
+		log,
+		metrics,
+	}
+}
+
+type reportObservationsFactory struct {
+	log     commonMonitoring.Logger
+	metrics metrics.ReportObservations
+}
+
+func (p *reportObservationsFactory) NewExporter(
+	params commonMonitoring.ExporterParams,
+) (commonMonitoring.Exporter, error) {
+	return &reportObservations{
+		metrics.FeedInput{
+			AccountAddress: params.FeedConfig.GetContractAddress(),
+			FeedID:         params.FeedConfig.GetContractAddress(),
+			ChainID:        params.ChainConfig.GetChainID(),
+			ContractStatus: params.FeedConfig.GetContractStatus(),
+			ContractType:   params.FeedConfig.GetContractType(),
+			FeedName:       params.FeedConfig.GetName(),
+			FeedPath:       params.FeedConfig.GetPath(),
+			NetworkID:      params.ChainConfig.GetNetworkID(),
+			NetworkName:    params.ChainConfig.GetNetworkName(),
+		},
+		p.log,
+		p.metrics,
+	}, nil
+}
+
+type reportObservations struct {
+	label   metrics.FeedInput // static for each feed
+	log     commonMonitoring.Logger
+	metrics metrics.ReportObservations
+}
+
+func (p *reportObservations) Export(ctx context.Context, data interface{}) {
+	details, err := types.MakeTxDetails(data)
+	if err != nil {
+		return // skip if input could not be parsed
+	}
+
+	// skip on no updates
+	if len(details) == 0 {
+		return
+	}
+
+	// sanity check: find non-empty detail
+	// assumption: details ordered from latest -> earliest
+	var latest types.TxDetails
+	for _, d := range details {
+		if !d.Empty() {
+			latest = d
+			break
+		}
+	}
+	if latest.Empty() {
+		p.log.Errorw("exporter could not find non-empty TxDetails", "feed", p.label.ToPromLabels())
+		return
+	}
+
+	p.metrics.SetCount(latest.ObservationCount, p.label)
+}
+
+func (p *reportObservations) Cleanup(_ context.Context) {
+	p.metrics.Cleanup(p.label)
+}

pkg/monitoring/exporter/reportobservations_test.go

@@ -0,0 +1,50 @@
+package exporter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zapcore"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	commonMonitoring "github.com/smartcontractkit/chainlink-common/pkg/monitoring"
+	"github.com/smartcontractkit/chainlink-common/pkg/utils"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/metrics/mocks"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/testutils"
+	"github.com/smartcontractkit/chainlink-solana/pkg/monitoring/types"
+)
+
+func TestReportObservations(t *testing.T) {
+	ctx := utils.Context(t)
+	lgr, logs := logger.TestObserved(t, zapcore.ErrorLevel)
+	m := mocks.NewReportObservations(t)
+	m.On("SetCount", mock.Anything, mock.Anything).Once()
+	m.On("Cleanup", mock.Anything).Once()
+
+	factory := NewReportObservationsFactory(lgr, m)
+
+	chainConfig := testutils.GenerateChainConfig()
+	feedConfig := testutils.GenerateFeedConfig()
+	exporter, err := factory.NewExporter(commonMonitoring.ExporterParams{ChainConfig: chainConfig, FeedConfig: feedConfig, Nodes: []commonMonitoring.NodeConfig{}})
+	require.NoError(t, err)
+
+	// happy path
+	exporter.Export(ctx, []types.TxDetails{{ObservationCount: 10}})
+	exporter.Cleanup(ctx)
+
+	// not txdetails type - no calls to mock
+	assert.NotPanics(t, func() { exporter.Export(ctx, 1) })
+
+	// zero txdetails - no calls to mock
+	exporter.Export(ctx, []types.TxDetails{})
+
+	// empty txdetails
+	exporter.Export(ctx, []types.TxDetails{{}})
+	assert.Equal(t, 1, logs.FilterMessage("exporter could not find non-empty TxDetails").Len())
+
+	// multiple TxDetails should only call for the first non-empty one
+	m.On("SetCount", uint8(1), mock.Anything).Once()
+	exporter.Export(ctx, []types.TxDetails{{}, {ObservationCount: 1}, {ObservationCount: 10}})
+}

pkg/monitoring/metrics/reportobservations.go

@@ -9,7 +9,7 @@ import (
 //go:generate mockery --name ReportObservations --output ./mocks/
 
 type ReportObservations interface {
-	SetCount(count uint64, feedInput FeedInput)
+	SetCount(count uint8, feedInput FeedInput)
 	Cleanup(feedInput FeedInput)
 }
 
@@ -23,7 +23,7 @@ func NewReportObservations(log commonMonitoring.Logger) *reportObservations {
 	return &reportObservations{newSimpleGauge(log, types.ReportObservationMetric)}
 }
 
-func (ro *reportObservations) SetCount(count uint64, feedInput FeedInput) {
+func (ro *reportObservations) SetCount(count uint8, feedInput FeedInput) {
 	ro.set(float64(count), feedInput.ToPromLabels())
 }
 

pkg/monitoring/metrics/reportobservations_test.go

@@ -24,7 +24,7 @@ func TestReportObservations(t *testing.T) {
 	inputs := FeedInput{NetworkName: t.Name()}
 
 	// set gauge
-	assert.NotPanics(t, func() { m.SetCount(uint64(v), inputs) })
+	assert.NotPanics(t, func() { m.SetCount(uint8(v), inputs) })
 	promBal := testutil.ToFloat64(g.With(inputs.ToPromLabels()))
 	assert.Equal(t, float64(v), promBal)
 

pkg/monitoring/source_txdetails.go

@@ -81,6 +81,6 @@ func (s *txDetailsSource) Fetch(ctx context.Context) (interface{}, error) {
 		details = append(details, res)
 	}
 
-	// only return successful OCR2 transmit transactions
+	// only return successful OCR2 transmit transactions (slice/array)
 	return details, nil
 }