Merge pull request #716 from GeorgeTsagk/tapd-metrics-updated

Add enhanced prometheus metrics for tapd

monitoring/asset_balances_collector.go

@@ -0,0 +1,114 @@
+package monitoring
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// assetBalancesCollector is a Prometheus collector that exports the balances
+// of all taproot assets.
+type assetBalancesCollector struct {
+	collectMx sync.Mutex
+
+	cfg      *PrometheusConfig
+	registry *prometheus.Registry
+
+	balancesVec *prometheus.GaugeVec
+
+	utxosVec *prometheus.GaugeVec
+}
+
+func newAssetBalancesCollector(cfg *PrometheusConfig,
+	registry *prometheus.Registry) (*assetBalancesCollector, error) {
+
+	if cfg == nil {
+		return nil, errors.New("asset collector prometheus cfg is nil")
+	}
+
+	if cfg.AssetStore == nil {
+		return nil, errors.New("asset collector asset store is nil")
+	}
+
+	return &assetBalancesCollector{
+		cfg:      cfg,
+		registry: registry,
+		balancesVec: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "asset_balances",
+				Help: "Balances of all taproot assets",
+			},
+			[]string{"asset_name"},
+		),
+		utxosVec: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "utxos_assets_held",
+				Help: "Number of UTXOs used for taproot assets",
+			},
+			[]string{"outpoint"},
+		),
+	}, nil
+}
+
+// Describe sends the super-set of all possible descriptors of metrics
+// collected by this Collector to the provided channel and returns once the
+// last descriptor has been sent.
+//
+// NOTE: Part of the prometheus.Collector interface.
+func (a *assetBalancesCollector) Describe(ch chan<- *prometheus.Desc) {
+	a.collectMx.Lock()
+	defer a.collectMx.Unlock()
+
+	a.balancesVec.Describe(ch)
+	a.utxosVec.Describe(ch)
+}
+
+// Collect is called by the Prometheus registry when collecting metrics.
+//
+// NOTE: Part of the prometheus.Collector interface.
+func (a *assetBalancesCollector) Collect(ch chan<- prometheus.Metric) {
+	a.collectMx.Lock()
+	defer a.collectMx.Unlock()
+
+	ctxdb, cancel := context.WithTimeout(context.Background(), dbTimeout)
+	defer cancel()
+
+	assets, err := a.cfg.AssetStore.FetchAllAssets(ctxdb, false, false, nil)
+	if err != nil {
+		log.Errorf("unable to fetch assets: %v", err)
+		return
+	}
+
+	utxos, err := a.cfg.AssetStore.FetchManagedUTXOs(ctxdb)
+	if err != nil {
+		log.Errorf("unable to fetch utxos: %v", err)
+		return
+	}
+
+	a.utxosVec.Reset()
+	a.balancesVec.Reset()
+
+	utxoMap := make(map[string]prometheus.Gauge)
+
+	for _, utxo := range utxos {
+		utxoOutpoint := utxo.OutPoint.String()
+		utxoMap[utxoOutpoint] = a.utxosVec.WithLabelValues(utxoOutpoint)
+	}
+
+	for _, asset := range assets {
+		a.balancesVec.WithLabelValues(asset.Tag).
+			Set(float64(asset.Amount))
+
+		utxoGauge, ok := utxoMap[asset.AnchorOutpoint.String()]
+		if !ok {
+			continue
+		}
+
+		utxoGauge.Inc()
+	}
+
+	a.balancesVec.Collect(ch)
+	a.utxosVec.Collect(ch)
+}

monitoring/asset_collector.go

@@ -0,0 +1,128 @@
+package monitoring
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	numAssetsMintedMetric = "num_assets_minted"
+
+	numTotalGroupsMetric = "num_total_groups"
+
+	numTotalSyncsMetric = "num_total_syncs"
+
+	numTotalProofsMetric = "num_total_proofs"
+)
+
+// universeStatsCollector is a Prometheus collector that exports the stats of
+// the universe.
+type universeStatsCollector struct {
+	collectMx sync.Mutex
+
+	cfg      *PrometheusConfig
+	registry *prometheus.Registry
+
+	gauges map[string]prometheus.Gauge
+}
+
+func newUniverseStatsCollector(cfg *PrometheusConfig,
+	registry *prometheus.Registry) (*universeStatsCollector, error) {
+
+	if cfg == nil {
+		return nil, errors.New("universe stats collector prometheus " +
+			"cfg is nil")
+	}
+
+	if cfg.UniverseStats == nil {
+		return nil, errors.New("universe stats collector universe " +
+			"stats is nil")
+	}
+
+	gaugesMap := map[string]prometheus.Gauge{
+		numAssetsMintedMetric: prometheus.NewGauge(
+			prometheus.GaugeOpts{
+				Name: numAssetsMintedMetric,
+				Help: "Total number of assets minted",
+			},
+		),
+		numTotalGroupsMetric: prometheus.NewGauge(
+			prometheus.GaugeOpts{
+				Name: numTotalGroupsMetric,
+				Help: "Total number of groups",
+			},
+		),
+		numTotalSyncsMetric: prometheus.NewGauge(
+			prometheus.GaugeOpts{
+				Name: numTotalSyncsMetric,
+				Help: "Total number of syncs",
+			},
+		),
+		numTotalProofsMetric: prometheus.NewGauge(
+			prometheus.GaugeOpts{
+				Name: numTotalProofsMetric,
+				Help: "Total number of proofs",
+			},
+		),
+	}
+
+	return &universeStatsCollector{
+		cfg:      cfg,
+		registry: registry,
+		gauges:   gaugesMap,
+	}, nil
+}
+
+// Describe sends the super-set of all possible descriptors of metrics
+// collected by this Collector to the provided channel and returns once the
+// last descriptor has been sent.
+//
+// NOTE: Part of the prometheus.Collector interface.
+func (a *universeStatsCollector) Describe(ch chan<- *prometheus.Desc) {
+	a.collectMx.Lock()
+	defer a.collectMx.Unlock()
+
+	for _, gauge := range a.gauges {
+		gauge.Describe(ch)
+	}
+}
+
+// Collect is called by the Prometheus registry when collecting metrics.
+//
+// NOTE: Part of the prometheus.Collector interface.
+func (a *universeStatsCollector) Collect(ch chan<- prometheus.Metric) {
+	a.collectMx.Lock()
+	defer a.collectMx.Unlock()
+
+	ctx, cancel := context.WithTimeout(context.Background(), dbTimeout)
+	defer cancel()
+
+	universeStats, err := a.cfg.UniverseStats.AggregateSyncStats(ctx)
+	if err != nil {
+		log.Errorf("unable to get aggregate universe stats: %v", err)
+		return
+	}
+
+	a.gauges[numAssetsMintedMetric].Set(
+		float64(universeStats.NumTotalAssets),
+	)
+
+	a.gauges[numTotalGroupsMetric].Set(
+		float64(universeStats.NumTotalGroups),
+	)
+
+	a.gauges[numTotalSyncsMetric].Set(
+		float64(universeStats.NumTotalSyncs),
+	)
+
+	a.gauges[numTotalProofsMetric].Set(
+		float64(universeStats.NumTotalProofs),
+	)
+
+	for _, gauge := range a.gauges {
+		gauge.Collect(ch)
+	}
+}

monitoring/config.go

@@ -1,6 +1,11 @@
 package monitoring
 
-import "google.golang.org/grpc"
+import (
+	"github.com/lightninglabs/taproot-assets/tapdb"
+	"github.com/lightninglabs/taproot-assets/tapgarden"
+	"github.com/lightninglabs/taproot-assets/universe"
+	"google.golang.org/grpc"
+)
 
 // PrometheusConfig is the set of configuration data that specifies if
 // Prometheus metric exporting is activated, and if so the listening address of
@@ -17,6 +22,18 @@ type PrometheusConfig struct {
 	// generic RPC metrics to monitor the health of the service.
 	RPCServer *grpc.Server
 
+	// UniverseStats is used to collect any stats that are relevant to the
+	// universe.
+	UniverseStats universe.Telemetry
+
+	// AssetStore is used to collect any stats that are relevant to the
+	// asset store.
+	AssetStore *tapdb.AssetStore
+
+	// AssetMinter is used to collect any stats that are relevant to the
+	// asset minter.
+	AssetMinter tapgarden.Planter
+
 	// PerfHistograms indicates if the additional histogram information for
 	// latency, and handling time of gRPC calls should be enabled. This
 	// generates additional data, and consume more memory for the

monitoring/garden_collector.go

@@ -0,0 +1,120 @@
+package monitoring
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/lightninglabs/taproot-assets/tapgarden"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// assetBalancesCollector is a Prometheus collector that exports the balances
+// of all taproot assets.
+type gardenCollector struct {
+	collectMx sync.Mutex
+
+	cfg      *PrometheusConfig
+	registry *prometheus.Registry
+
+	pendingBatches   *prometheus.GaugeVec
+	completedBatches prometheus.Gauge
+}
+
+func newGardenCollector(cfg *PrometheusConfig,
+	registry *prometheus.Registry) (*gardenCollector, error) {
+
+	if cfg == nil {
+		return nil, errors.New("garden collector prometheus cfg is nil")
+	}
+
+	if cfg.AssetStore == nil {
+		return nil, errors.New("garden collector asset store is nil")
+	}
+
+	return &gardenCollector{
+		cfg:      cfg,
+		registry: registry,
+		pendingBatches: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Name: "mint_batches",
+				Help: "Batched mint transactions",
+			},
+			[]string{"batch_pubkey"},
+		),
+		completedBatches: prometheus.NewGauge(
+			prometheus.GaugeOpts{
+				Name: "completed_batches",
+				Help: "Total number of completed mint batches",
+			},
+		),
+	}, nil
+}
+
+// Describe sends the super-set of all possible descriptors of metrics
+// collected by this Collector to the provided channel and returns once the
+// last descriptor has been sent.
+//
+// NOTE: Part of the prometheus.Collector interface.
+func (a *gardenCollector) Describe(ch chan<- *prometheus.Desc) {
+	a.collectMx.Lock()
+	defer a.collectMx.Unlock()
+
+	a.pendingBatches.Describe(ch)
+	a.completedBatches.Describe(ch)
+}
+
+// Collect is called by the Prometheus registry when collecting metrics.
+//
+// NOTE: Part of the prometheus.Collector interface.
+func (a *gardenCollector) Collect(ch chan<- prometheus.Metric) {
+	a.collectMx.Lock()
+	defer a.collectMx.Unlock()
+
+	a.completedBatches.Set(0)
+
+	// Get the number of pending batches.
+	batches, err := a.cfg.AssetMinter.ListBatches(nil)
+	if err != nil {
+		log.Errorf("unable to list batches: %v", err)
+		return
+	}
+
+	completed := 0
+
+	for _, batch := range batches {
+		state := batch.State()
+
+		switch {
+		case state == tapgarden.BatchStatePending ||
+			state == tapgarden.BatchStateFrozen ||
+			state == tapgarden.BatchStateCommitted ||
+			state == tapgarden.BatchStateBroadcast ||
+			state == tapgarden.BatchStateConfirmed:
+
+			if state == tapgarden.BatchStatePending {
+				a.pendingBatches.WithLabelValues(
+					batch.BatchKey.PubKey.X().String(),
+				).Set(
+					float64(len(batch.Seedlings)),
+				)
+			}
+
+		case state == tapgarden.BatchStateFinalized ||
+			state == tapgarden.BatchStateSeedlingCancelled ||
+			state == tapgarden.BatchStateSproutCancelled:
+
+			a.pendingBatches.DeleteLabelValues(
+				batch.BatchKey.PubKey.X().String(),
+			)
+
+			if state == tapgarden.BatchStateFinalized {
+				completed += 1
+			}
+		}
+	}
+
+	a.completedBatches.Set(float64(completed))
+
+	a.pendingBatches.Collect(ch)
+	a.completedBatches.Collect(ch)
+}