feat: implement pool owner DRep registration monitoring

.golangci.yaml

@@ -43,7 +43,6 @@ linters:
     - usestdlibvars
     - whitespace
     - wrapcheck
-    - tagliatelle
     - sqlclosecheck
     - promlinter
     - prealloc

internal/blockfrost/blockfrost.go

@@ -23,4 +23,20 @@ type Client interface {
 	GetGenesisInfo(ctx context.Context) (blockfrost.GenesisBlock, error)
 	GetAllPools(ctx context.Context) ([]string, error)
 	GetNetworkInfo(ctx context.Context) (blockfrost.NetworkInfo, error)
+	GetAccountInfo(ctx context.Context, stakeAddress string) (Account, error)
+}
+
+// TODO: remove it when the blockfrost-go library will be updated and return the DrepID
+type Account struct {
+	StakeAddress       string  `json:"stake_address"`
+	Active             bool    `json:"active"`
+	ActiveEpoch        *int64  `json:"active_epoch"`
+	ControlledAmount   string  `json:"controlled_amount"`
+	RewardsSum         string  `json:"rewards_sum"`
+	WithdrawalsSum     string  `json:"withdrawals_sum"`
+	ReservesSum        string  `json:"reserves_sum"`
+	TreasurySum        string  `json:"treasury_sum"`
+	WithdrawableAmount string  `json:"withdrawable_amount"`
+	PoolID             *string `json:"pool_id"`
+	DrepID             *string `json:"drep_id"`
 }

internal/blockfrost/blockfrostapi/blockfrost.go

@@ -2,7 +2,11 @@ package blockfrostapi
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"io"
 	"net/http"
+	"net/url"
 	"time"
 
 	"github.com/blockfrost/blockfrost-go"
@@ -11,6 +15,8 @@ import (
 
 type Client struct {
 	blockfrost blockfrost.APIClient
+	apiURL     string
+	projectID  string
 }
 
 var _ bf.Client = (*Client)(nil)
@@ -34,6 +40,8 @@ func NewClient(opts ClientOptions) *Client {
 				},
 			},
 		),
+		apiURL:    opts.Server,
+		projectID: opts.ProjectID,
 	}
 }
 
@@ -173,3 +181,37 @@ func (c *Client) GetAllPools(ctx context.Context) ([]string, error) {
 func (c *Client) GetNetworkInfo(ctx context.Context) (blockfrost.NetworkInfo, error) {
 	return c.blockfrost.Network(ctx)
 }
+
+func (c *Client) GetAccountInfo(ctx context.Context, address string) (bf.Account, error) {
+	account := bf.Account{}
+	url, err := url.JoinPath(c.apiURL, "accounts", address)
+	if err != nil {
+		return account, fmt.Errorf("failed to join URL path to get account details: %w", err)
+	}
+
+	cctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(cctx, http.MethodGet, url, nil)
+	if err != nil {
+		return account, fmt.Errorf("failed to create request to get account details: %w", err)
+	}
+
+	req.Header.Set("Project_id", c.projectID)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return account, fmt.Errorf("failed to send request to get account details: %w", err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return account, fmt.Errorf("failed to read response body to get account details: %w", err)
+	}
+
+	if err = json.Unmarshal(body, &account); err != nil {
+		return account, fmt.Errorf("failed to unmarshal account info: %w", err)
+	}
+
+	return account, nil
+}

internal/metrics/metrics.go

@@ -19,6 +19,7 @@ type Collection struct {
 	RelaysPerPool                     *prometheus.GaugeVec
 	PoolsPledgeMet                    *prometheus.GaugeVec
 	PoolsSaturationLevel              *prometheus.GaugeVec
+	PoolsDRepRegistered               *prometheus.GaugeVec
 	MonitoredValidatorsCount          *prometheus.GaugeVec
 	MissedBlocks                      *prometheus.CounterVec
 	ConsecutiveMissedBlocks           *prometheus.GaugeVec
@@ -127,6 +128,14 @@ func NewCollection() *Collection {
 			},
 			[]string{"pool_name", "pool_id", "pool_instance"},
 		),
+		PoolsDRepRegistered: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "pool_drep_registered",
+				Help:      "Whether the pool owner is registered to a DRep (0 or 1)",
+			},
+			[]string{"pool_name", "pool_id", "pool_instance"},
+		),
 		MonitoredValidatorsCount: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
 				Namespace: "cardano_validator_watcher",
@@ -216,6 +225,7 @@ func (m *Collection) MustRegister(reg prometheus.Registerer) {
 	reg.MustRegister(m.NextEpochStartTime)
 	reg.MustRegister(m.PoolsPledgeMet)
 	reg.MustRegister(m.PoolsSaturationLevel)
+	reg.MustRegister(m.PoolsDRepRegistered)
 	reg.MustRegister(m.MonitoredValidatorsCount)
 	reg.MustRegister(m.MissedBlocks)
 	reg.MustRegister(m.ConsecutiveMissedBlocks)

internal/watcher/watcher_block_state_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/DATA-DOG/go-sqlmock"
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
 	"github.com/blockfrost/blockfrost-go"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"

internal/watcher/watcher_pool.go

@@ -29,6 +29,7 @@ type PoolWatcher struct {
 	poolstats   pools.PoolStats
 	healthStore *HealthStore
 	cache       *ristretto.Cache[string, interface{}]
+	cacheTTL    time.Duration
 	opts        PoolWatcherOptions
 }
 
@@ -65,6 +66,7 @@ func NewPoolWatcher(
 		poolstats:   pools.GetPoolStats(),
 		healthStore: healthStore,
 		cache:       cache,
+		cacheTTL:    2 * opts.RefreshInterval,
 		opts:        opts,
 	}, nil
 }
@@ -160,6 +162,17 @@ func (w *PoolWatcher) fetch(ctx context.Context) error {
 			return fmt.Errorf("unable to retrieve relays for pool '%s': %w", pool.ID, err)
 		}
 		w.metrics.RelaysPerPool.WithLabelValues(*poolMetadata.Ticker, pool.ID, pool.Instance).Set(float64(len(poolRelays)))
+
+		// check if a pool owner is registered to a DRep
+		poolAccountInfo, err := w.getAccountInfo(ctx, poolInfo.RewardAccount)
+		if err != nil {
+			return fmt.Errorf("unable to retrieve account info for pool '%s': %w", pool.ID, err)
+		}
+		if poolAccountInfo.DrepID != nil {
+			w.metrics.PoolsDRepRegistered.WithLabelValues(pool.Name, pool.ID, pool.Instance).Set(1)
+		} else {
+			w.metrics.PoolsDRepRegistered.WithLabelValues(pool.Name, pool.ID, pool.Instance).Set(0)
+		}
 	}
 
 	return nil
@@ -175,7 +188,7 @@ func (w *PoolWatcher) getPoolMetadata(ctx context.Context, PoolID string) (bfAPI
 		if err != nil {
 			return metadata, fmt.Errorf("unable to retrieve metadata for pool '%s': %w", PoolID, err)
 		}
-		w.cache.SetWithTTL(PoolID+"_metadata", poolMetadata, 1, 2*w.opts.RefreshInterval)
+		w.cache.SetWithTTL(PoolID+"_metadata", poolMetadata, 1, w.cacheTTL)
 		w.cache.Wait()
 	}
 	metadata = poolMetadata.(bfAPI.PoolMetadata)
@@ -193,7 +206,7 @@ func (w *PoolWatcher) getPoolRelays(ctx context.Context, PoolID string) ([]bfAPI
 		if err != nil {
 			return relays, fmt.Errorf("unable to retrieve relays for pool '%s': %w", PoolID, err)
 		}
-		w.cache.SetWithTTL(PoolID+"_relays", poolRelays, 1, 2*w.opts.RefreshInterval)
+		w.cache.SetWithTTL(PoolID+"_relays", poolRelays, 1, w.cacheTTL)
 		w.cache.Wait()
 	}
 	relays = poolRelays.([]bfAPI.PoolRelay)
@@ -211,10 +224,27 @@ func (w *PoolWatcher) getPoolInfo(ctx context.Context, PoolID string) (bfAPI.Poo
 		if err != nil {
 			return pool, fmt.Errorf("unable to retrieve info for pool '%s': %w", PoolID, err)
 		}
-		w.cache.SetWithTTL(PoolID+"_info", poolInfo, 1, 2*w.opts.RefreshInterval)
+		w.cache.SetWithTTL(PoolID+"_info", poolInfo, 1, w.cacheTTL)
 		w.cache.Wait()
 	}
 	pool = poolInfo.(bfAPI.Pool)
 
 	return pool, nil
 }
+
+func (w *PoolWatcher) getAccountInfo(ctx context.Context, address string) (blockfrost.Account, error) {
+	var err error
+	var account blockfrost.Account
+
+	accountInfo, ok := w.cache.Get(address + "_info")
+	if !ok {
+		accountInfo, err = w.blockfrost.GetAccountInfo(ctx, address)
+		if err != nil {
+			return blockfrost.Account{}, fmt.Errorf("unable to retrieve info for address '%s': %w", address, err)
+		}
+		w.cache.SetWithTTL(address+"_info", accountInfo, 1, w.cacheTTL)
+		w.cache.Wait()
+	}
+	account = accountInfo.(blockfrost.Account)
+	return account, nil
+}