add tx success metric

Author: Richard Patel

balance.go

@@ -11,6 +11,7 @@ import (
 	"go.uber.org/zap"
 )
 
+// balanceScraper retrieves the Pyth publisher balances on request.
 type balanceScraper struct {
 	*prometheus.GaugeVec
 
@@ -28,6 +29,7 @@ func newBalanceScraper(publishers []solana.PublicKey, rpcURL string, log *zap.Lo
 	}
 }
 
+// Collect gets invoked by the Prometheus exporter when a new scrape is requested.
 func (b *balanceScraper) Collect(metrics chan<- prometheus.Metric) {
 	const collectTimeout = 5 * time.Second
 	ctx, cancel := context.WithTimeout(context.Background(), collectTimeout)

logger.go

@@ -18,5 +18,6 @@ func getLogger() *zap.Logger {
 	if err != nil {
 		panic("no logger: " + err.Error())
 	}
+	logger.Info("Log level", zap.Stringer("level", flagLogLevel))
 	return logger
 }

main.go

@@ -134,6 +134,14 @@ func main() {
 	balances := newBalanceScraper(publishKeys.pubkeys, *rpcURL, log.Named("balances"))
 	metrics.Registry.MustRegister(balances)
 
+	// Create tx tailer.
+	txs := newTxScraper(*rpcURL, log.Named("txs"), publishKeys.pubkeys)
+	group.Go(func() error {
+		const scrapeInterval = 5 * time.Second
+		txs.run(ctx, scrapeInterval)
+		return nil
+	})
+
 	if err := group.Wait(); err != nil {
 		log.Fatal("App crashed", zap.Error(err))
 	}

metrics/metrics.go

@@ -25,18 +25,17 @@ var factory = promauto.With(Registry)
 
 // On-chain transaction execution status.
 const (
-	txStatusSuccess = "success"
-	txStatusFailed  = "failed"
+	TxStatusSuccess = "success"
+	TxStatusFailed  = "failed"
 )
 
 var (
-	// RPC request stats
-	rpcRequestsTotal = factory.NewCounterVec(prometheus.CounterOpts{
+	RpcRequestsTotal = factory.NewCounter(prometheus.CounterOpts{
 		Namespace: Namespace,
 		Subsystem: SubsystemExporter,
 		Name:      "rpc_requests_total",
 		Help:      "Number of outgoing RPC requests from pyth_exporter to RPC nodes",
-	}, []string{})
+	})
 	WsActiveConns = factory.NewGauge(prometheus.GaugeOpts{
 		Namespace: Namespace,
 		Subsystem: SubsystemExporter,
@@ -88,13 +87,7 @@ var (
 		Help:      "Last observed slot for Pyth publisher",
 	}, []string{labelProduct, labelPublisher})
 
-	// Publisher Observers
-	metricTxFeesTotal = factory.NewCounterVec(prometheus.CounterOpts{
-		Namespace: Namespace,
-		Name:      "tx_fees_total",
-		Help:      "Approximate amount of SOL in lamports spent on Pyth publishing",
-	}, []string{labelPublisher})
-	metricTxCount = factory.NewCounterVec(prometheus.CounterOpts{
+	TxCount = factory.NewCounterVec(prometheus.CounterOpts{
 		Namespace: Namespace,
 		Name:      "txs_total",
 		Help:      "Approximate number of Pyth transactions sent",

price.go

@@ -8,6 +8,7 @@ import (
 	"go.blockdaemon.com/pyth_exporter/pyth"
 )
 
+// priceScraper scrapes prices out of the on-chain Pyth price accounts.
 type priceScraper struct {
 	productKeys []solana.PublicKey
 	publishKeys []solana.PublicKey

txs.go

@@ -0,0 +1,166 @@
+package main
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/gagliardetto/solana-go"
+	"github.com/gagliardetto/solana-go/rpc"
+	"go.blockdaemon.com/pyth_exporter/metrics"
+	"go.uber.org/zap"
+)
+
+// txScraper polls publisher transactions.
+type txScraper struct {
+	tailers []*txTailer
+	log     *zap.Logger
+	rpc     *rpc.Client
+}
+
+func newTxScraper(rpcURL string, log *zap.Logger, publishers []solana.PublicKey) *txScraper {
+	scraper := &txScraper{
+		rpc:     rpc.New(rpcURL),
+		log:     log,
+		tailers: make([]*txTailer, len(publishers)),
+	}
+
+	for i, pubkey := range publishers {
+		scraper.tailers[i] = newTxTailer(scraper, pubkey)
+	}
+
+	return scraper
+}
+
+func (s *txScraper) run(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+	wait:
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			break wait
+		}
+
+		s.poll(ctx, interval)
+	}
+}
+
+func (s *txScraper) poll(ctx context.Context, interval time.Duration) {
+	ctx, cancel := context.WithTimeout(ctx, interval)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(len(s.tailers))
+	for _, tailer := range s.tailers {
+		go s.pollOne(ctx, &wg, tailer)
+	}
+	wg.Wait()
+}
+
+func (s *txScraper) pollOne(ctx context.Context, wg *sync.WaitGroup, tailer *txTailer) {
+	defer wg.Done()
+	for {
+		isEnd, err := tailer.poll(ctx)
+		if isEnd {
+			break
+		}
+		if err != nil {
+			s.log.Warn("Failed to poll account txs", zap.Error(err))
+		}
+	}
+}
+
+// txTailer "tails" the transaction log of an account.
+type txTailer struct {
+	*txScraper
+
+	pubkey    solana.PublicKey
+	pubkeyStr string
+	lastSlot  uint64
+	lastSig   solana.Signature
+}
+
+func newTxTailer(scraper *txScraper, pubkey solana.PublicKey) *txTailer {
+	return &txTailer{
+		txScraper: scraper,
+		pubkey:    pubkey,
+		pubkeyStr: pubkey.String(),
+	}
+}
+
+func (t *txTailer) refreshLastSig(ctx context.Context) error {
+	oneInt := 1
+	sigs, err := t.rpc.GetSignaturesForAddressWithOpts(ctx, t.pubkey, &rpc.GetSignaturesForAddressOpts{
+		Limit: &oneInt,
+	})
+	if err != nil {
+		return err
+	}
+	metrics.RpcRequestsTotal.Inc()
+
+	if len(sigs) == 0 {
+		return nil // empty account
+	}
+	t.lastSlot = sigs[0].Slot
+	t.lastSig = sigs[0].Signature
+	return nil
+}
+
+// poll retrieves the latest transactions.
+func (t *txTailer) poll(ctx context.Context) (end bool, err error) {
+	t.log.Debug("Polling new txs",
+		zap.String("publisher", t.pubkeyStr),
+		zap.Stringer("last_sig", t.lastSig))
+
+	// Get starting sig if none.
+	if t.lastSig.IsZero() {
+		return true, t.refreshLastSig(ctx)
+	}
+
+	// Get sigs since last check.
+	var tailLimit = 100
+	sigs, err := t.rpc.GetSignaturesForAddressWithOpts(ctx, t.pubkey, &rpc.GetSignaturesForAddressOpts{
+		Limit: &tailLimit,
+		Until: t.lastSig,
+	})
+	if err != nil {
+		return true, err
+	}
+	metrics.RpcRequestsTotal.Inc()
+
+	if len(sigs) == 0 {
+		return true, nil
+	}
+
+	// Iteration is newest to latest.
+	// So write down the first sig as the newest sig, so we can later continue.
+	stopSlot := t.lastSlot
+	if sigs[0].Slot > t.lastSlot {
+		t.lastSlot = sigs[0].Slot
+		t.lastSig = sigs[0].Signature
+	}
+
+	// If the number of returned sigs matches exactly our requested limit, there is probably more.
+	end = len(sigs) != tailLimit
+
+	// Scroll through page.
+	for len(sigs) > 0 && sigs[0].Slot > stopSlot {
+		sig := sigs[0]
+
+		var status string
+		if sig.Err != nil {
+			status = metrics.TxStatusFailed
+		} else {
+			status = metrics.TxStatusSuccess
+		}
+		metrics.TxCount.WithLabelValues(t.pubkeyStr, status).Inc()
+
+		sigs = sigs[1:]
+	}
+
+	return end, nil
+}