feat: add blocks watcher

Author: rayanebel

.gitignore

@@ -1,3 +1,5 @@
 config.yaml
 .codegpt
-bin/
+bin/
+config/
+*.db*

cmd/watcher/app/config/config.go

@@ -11,11 +11,19 @@ type Config struct {
 	Pools                pools.Pools          `mapstructure:"pools"`
 	HTTP                 HTTPConfig           `mapstructure:"http"`
 	Network              string               `mapstructure:"network"`
+	Cardano              CardanoConfig        `mapstructure:"cardano"`
+	Database             DatabaseConfig       `mapstructure:"database"`
 	Blockfrost           BlockFrostConfig     `mapstructure:"blockfrost"`
+	BlockWatcherConfig   BlockWatcherConfig   `mapstructure:"block-watcher"`
 	PoolWatcherConfig    PoolWatcherConfig    `mapstructure:"pool-watcher"`
 	NetworkWatcherConfig NetworkWatcherConfig `mapstructure:"network-watcher"`
 }
 
+type BlockWatcherConfig struct {
+	Enabled         bool `mapstructure:"enabled"`
+	RefreshInterval int  `mapstructure:"refresh-interval"`
+}
+
 type PoolWatcherConfig struct {
 	Enabled         bool `mapstructure:"enabled"`
 	RefreshInterval int  `mapstructure:"refresh-interval"`
@@ -38,6 +46,16 @@ type BlockFrostConfig struct {
 	Timeout     int    `mapstructure:"timeout"`
 }
 
+type CardanoConfig struct {
+	ConfigDir  string `mapstructure:"config-dir"`
+	Timezone   string `mapstructure:"timezone"`
+	SocketPath string `mapstructure:"socket-path"`
+}
+
+type DatabaseConfig struct {
+	Path string `mapstructure:"path"`
+}
+
 func (c *Config) Validate() error {
 	switch c.Network {
 	case "mainnet", "preprod":

cmd/watcher/app/watcher.go

@@ -11,13 +11,19 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/jmoiron/sqlx"
 	"github.com/kilnfi/cardano-validator-watcher/cmd/watcher/app/config"
 	"github.com/kilnfi/cardano-validator-watcher/internal/blockfrost"
 	"github.com/kilnfi/cardano-validator-watcher/internal/blockfrost/blockfrostapi"
+	"github.com/kilnfi/cardano-validator-watcher/internal/cardano"
+	"github.com/kilnfi/cardano-validator-watcher/internal/cardano/cardanocli"
+	"github.com/kilnfi/cardano-validator-watcher/internal/database"
 	"github.com/kilnfi/cardano-validator-watcher/internal/metrics"
 	"github.com/kilnfi/cardano-validator-watcher/internal/pools"
 	"github.com/kilnfi/cardano-validator-watcher/internal/server/http"
+	"github.com/kilnfi/cardano-validator-watcher/internal/slotleader"
 	"github.com/kilnfi/cardano-validator-watcher/internal/watcher"
+	"github.com/kilnfi/cardano-validator-watcher/migrations"
 	"github.com/prometheus/client_golang/prometheus"
 	"golang.org/x/sync/errgroup"
 
@@ -76,29 +82,40 @@ func NewWatcherCommand() *cobra.Command {
 	cmd.Flags().StringP("http-server-host", "", http.ServerDefaultHost, "host on which HTTP server should listen")
 	cmd.Flags().IntP("http-server-port", "", http.ServerDefaultPort, "port on which HTTP server should listen")
 	cmd.Flags().StringP("network", "", "preprod", "cardano network ID")
+	cmd.Flags().StringP("database-path", "", "watcher.db", "path to the local database mainly used by cardano client")
+	cmd.Flags().StringP("cardano-config-dir", "", "/config", "path to the directory where the cardano config files are stored")
+	cmd.Flags().StringP("cardano-timezone", "", "UTC", "timezone to use with cardano-cli - https://en.wikipedia.org/wiki/List_of_tz_database_time_zones")
+	cmd.Flags().StringP("cardano-socket-path", "", "/var/run/cardano.socket", "socket path to communicate with a cardano node")
 	cmd.Flags().StringP("blockfrost-project-id", "", "", "blockfrost project id")
 	cmd.Flags().StringP("blockfrost-endpoint", "", "", "blockfrost API endpoint")
 	cmd.Flags().IntP("blockfrost-max-routines", "", 10, "number of routines used by blockfrost to perform concurrent actions")
 	cmd.Flags().IntP("blockfrost-timeout", "", 60, "Timeout for requests to the Blockfrost API (in seconds)")
-	cmd.Flags().BoolP("pool-watcher-enabled", "", true, "Enable pool watcher")
-	cmd.Flags().IntP("pool-watcher-refresh-interval", "", 60, "Interval at which the pool watcher collects data about the monitored pools (in seconds)")
 	cmd.Flags().BoolP("network-watcher-enabled", "", true, "Enable network watcher")
 	cmd.Flags().IntP("network-watcher-refresh-interval", "", 60, "Interval at which the network watcher collects data about the network (in seconds)")
+	cmd.Flags().BoolP("pool-watcher-enabled", "", true, "Enable pool watcher")
+	cmd.Flags().IntP("pool-watcher-refresh-interval", "", 60, "Interval at which the pool watcher collects data about the monitored pools (in seconds)")
+	cmd.Flags().BoolP("block-watcher-enabled", "", true, "Enable block watcher")
+	cmd.Flags().IntP("block-watcher-refresh-interval", "", 60, "Interval at which the block watcher collects and process slots (in seconds)")
 
 	// bind flag to viper
 	checkError(viper.BindPFlag("log-level", cmd.Flag("log-level")), "unable to bind log-level flag")
 	checkError(viper.BindPFlag("http.host", cmd.Flag("http-server-host")), "unable to bind http-server-host flag")
 	checkError(viper.BindPFlag("http.port", cmd.Flag("http-server-port")), "unable to bind http-server-port flag")
 	checkError(viper.BindPFlag("network", cmd.Flag("network")), "unable to bind network flag")
+	checkError(viper.BindPFlag("database.path", cmd.Flag("database-path")), "unable to bind database-path flag")
+	checkError(viper.BindPFlag("cardano.config-dir", cmd.Flag("cardano-config-dir")), "unable to bind cardano-config-dir flag")
+	checkError(viper.BindPFlag("cardano.timezone", cmd.Flag("cardano-timezone")), "unable to bind cardano-timezone flag")
+	checkError(viper.BindPFlag("cardano.socket-path", cmd.Flag("cardano-socket-path")), "unable to bind cardano-socket-path flag")
 	checkError(viper.BindPFlag("blockfrost.project-id", cmd.Flag("blockfrost-project-id")), "unable to bind blockfrost-project-id flag")
 	checkError(viper.BindPFlag("blockfrost.endpoint", cmd.Flag("blockfrost-endpoint")), "unable to bind blockfrost-endpoint flag")
 	checkError(viper.BindPFlag("blockfrost.max-routines", cmd.Flag("blockfrost-max-routines")), "unable to bind blockfrost-max-routines flag")
 	checkError(viper.BindPFlag("blockfrost.timeout", cmd.Flag("blockfrost-timeout")), "unable to bind blockfrost-timeout flag")
-	checkError(viper.BindPFlag("pool-watcher.enabled", cmd.Flag("pool-watcher-enabled")), "unable to bind pool-watcher-enabled flag")
-	checkError(viper.BindPFlag("pool-watcher.refresh-interval", cmd.Flag("pool-watcher-refresh-interval")), "unable to bind pool-watcher-refresh-interval flag")
 	checkError(viper.BindPFlag("network-watcher.enabled", cmd.Flag("network-watcher-enabled")), "unable to bind network-watcher-enabled flag")
 	checkError(viper.BindPFlag("network-watcher.refresh-interval", cmd.Flag("network-watcher-refresh-interval")), "unable to bind network-watcher-refresh-interval flag")
-
+	checkError(viper.BindPFlag("pool-watcher.enabled", cmd.Flag("pool-watcher-enabled")), "unable to bind pool-watcher-enabled flag")
+	checkError(viper.BindPFlag("pool-watcher.refresh-interval", cmd.Flag("pool-watcher-refresh-interval")), "unable to bind pool-watcher-refresh-interval flag")
+	checkError(viper.BindPFlag("block-watcher.enabled", cmd.Flag("block-watcher-enabled")), "unable to bind block-watcher-enabled flag")
+	checkError(viper.BindPFlag("block-watcher.refresh-interval", cmd.Flag("block-watcher-refresh-interval")), "unable to bind block-watcher-refresh-interval flag")
 	return cmd
 }
 
@@ -146,29 +163,60 @@ func run(_ *cobra.Command, _ []string) error {
 
 	eg, ctx := errgroup.WithContext(ctx)
 
+	// Connect and Init the DB
+	dbOpts := database.Options{
+		URL:          "?_journal=WAL&_timeout=15000&_fk=true&cache=shared",
+		Path:         cfg.Database.Path,
+		MaxOpenConns: 1,
+	}
+	database := database.NewDatabase(dbOpts)
+	if err := database.Connect(ctx); err != nil {
+		return fmt.Errorf("unable to connect to database: %w", err)
+	}
+	if err := database.MigrateUp(migrations.FS); err != nil {
+		return fmt.Errorf("unable to migrate database: %w", err)
+	}
+
 	// Initialize blockfrost and cardano clients with options
 	blockfrost := createBlockfrostClient()
+	cardano := createCardanoClient(blockfrost)
 
 	// Initialize prometheus metrics
 	registry := prometheus.NewRegistry()
 	metrics := metrics.NewCollection()
 	metrics.MustRegister(registry)
 
+	epoch, err := blockfrost.GetLatestEpoch(ctx)
+	if err != nil {
+		return fmt.Errorf("unable to get latest epoch: %w", err)
+	}
+
+	// Launch slot leader calculation for the current slot
+	slotLeaderService := slotleader.NewSlotLeaderService(database.DB, cardano, blockfrost, cfg.Pools, metrics)
+	if err := slotLeaderService.Refresh(ctx, epoch); err != nil {
+		return fmt.Errorf("unable to refresh slot leaders: %w", err)
+	}
+
 	// Start HTTP server
 	if err := startHTTPServer(eg, registry); err != nil {
 		return fmt.Errorf("unable to start http server: %w", err)
 	}
 
-	// Start Network Watcher
-	if cfg.NetworkWatcherConfig.Enabled {
-		startNetworkWatcher(ctx, eg, blockfrost, metrics)
-	}
-
 	// Start Pool Watcher
 	if cfg.PoolWatcherConfig.Enabled {
 		startPoolWatcher(ctx, eg, blockfrost, metrics, cfg.Pools)
 	}
 
+	// Start Block Watcher
+	if cfg.BlockWatcherConfig.Enabled {
+		startBlockWatcher(ctx, eg, cardano, blockfrost, slotLeaderService, metrics, cfg.Pools, database.DB)
+	}
+
+	// Start Network Watcher
+	if cfg.NetworkWatcherConfig.Enabled {
+		startNetworkWatcher(ctx, eg, blockfrost, metrics)
+	}
+
 	<-ctx.Done()
 	logger.Info("shutting down")
 
@@ -200,6 +248,17 @@ func createBlockfrostClient() blockfrost.Client {
 	return blockfrostapi.NewClient(opts)
 }
 
+func createCardanoClient(blockfrost blockfrost.Client) cardano.CardanoClient {
+	opts := cardanocli.ClientOptions{
+		ConfigDir:  cfg.Cardano.ConfigDir,
+		Network:    cfg.Network,
+		SocketPath: cfg.Cardano.SocketPath,
+		Timezone:   cfg.Cardano.Timezone,
+		DBPath:     cfg.Database.Path,
+	}
+	return cardanocli.NewClient(opts, blockfrost, &cardanocli.RealCommandExecutor{})
+}
+
 func startHTTPServer(eg *errgroup.Group, registry *prometheus.Registry) error {
 	var err error
 
@@ -255,6 +314,7 @@ func startPoolWatcher(
 	})
 }
 
+// startNetworkWatcher starts the network watcher service
 func startNetworkWatcher(
 	ctx context.Context,
 	eg *errgroup.Group,
@@ -279,6 +339,33 @@ func startNetworkWatcher(
 	})
 }
 
+// startBlockWatcher starts the block watcher service
+func startBlockWatcher(
+	ctx context.Context,
+	eg *errgroup.Group,
+	cardano cardano.CardanoClient,
+	blockfrost blockfrost.Client,
+	sl slotleader.SlotLeader,
+	metrics *metrics.Collection,
+	pools pools.Pools,
+	db *sqlx.DB,
+) {
+	eg.Go(func() error {
+		options := watcher.BlockWatcherOptions{
+			RefreshInterval: time.Second * time.Duration(cfg.BlockWatcherConfig.RefreshInterval),
+		}
+		blockWatcher := watcher.NewBlockWatcher(cardano, blockfrost, sl, pools, metrics, db, options)
+		logger.Info(
+			"starting watcher",
+			slog.String("component", "block-watcher"),
+		)
+		if err := blockWatcher.Start(ctx); err != nil {
+			return fmt.Errorf("unable to start block watcher: %w", err)
+		}
+		return nil
+	})
+}
+
 // checkError is a helper function to log an error and exit the program
 // used for the flag parsing
 func checkError(err error, msg string) {

internal/metrics/metrics.go

@@ -20,6 +20,13 @@ type Collection struct {
 	PoolsPledgeMet                    *prometheus.GaugeVec
 	PoolsSaturationLevel              *prometheus.GaugeVec
 	MonitoredValidatorsCount          *prometheus.GaugeVec
+	MissedBlocks                      *prometheus.CounterVec
+	ConsecutiveMissedBlocks           *prometheus.GaugeVec
+	OrphanedBlocks                    *prometheus.CounterVec
+	ValidatedBlocks                   *prometheus.CounterVec
+	ExpectedBlocks                    *prometheus.GaugeVec
+	LatestSlotProcessedByBlockWatcher prometheus.Gauge
+	NextSlotLeader                    *prometheus.GaugeVec
 }
 
 func NewCollection() *Collection {
@@ -127,6 +134,61 @@ func NewCollection() *Collection {
 			},
 			[]string{"status"},
 		),
+		MissedBlocks: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "missed_blocks_total",
+				Help:      "number of missed blocks in the current epoch",
+			},
+			[]string{"pool_name", "pool_id", "pool_instance", "epoch"},
+		),
+		ConsecutiveMissedBlocks: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "consecutive_missed_blocks",
+				Help:      "number of consecutive missed blocks in a row",
+			},
+			[]string{"pool_name", "pool_id", "pool_instance", "epoch"},
+		),
+		ValidatedBlocks: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "validated_blocks_total",
+				Help:      "number of validated blocks in the current epoch",
+			},
+			[]string{"pool_name", "pool_id", "pool_instance", "epoch"},
+		),
+		OrphanedBlocks: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "orphaned_blocks_total",
+				Help:      "number of orphaned blocks in the current epoch",
+			},
+			[]string{"pool_name", "pool_id", "pool_instance", "epoch"},
+		),
+		ExpectedBlocks: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "expected_blocks",
+				Help:      "number of expected blocks in the current epoch",
+			},
+			[]string{"pool_name", "pool_id", "pool_instance", "epoch"},
+		),
+		LatestSlotProcessedByBlockWatcher: prometheus.NewGauge(
+			prometheus.GaugeOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "latest_slot_processed_by_block_watcher",
+				Help:      "latest slot processed by the block watcher",
+			},
+		),
+		NextSlotLeader: prometheus.NewGaugeVec(
+			prometheus.GaugeOpts{
+				Namespace: "cardano_validator_watcher",
+				Name:      "next_slot_leader",
+				Help:      "next slot leader for each monitored pool",
+			},
+			[]string{"pool_name", "pool_id", "pool_instance", "epoch"},
+		),
 	}
 }
 
@@ -147,4 +209,11 @@ func (m *Collection) MustRegister(reg prometheus.Registerer) {
 	reg.MustRegister(m.PoolsPledgeMet)
 	reg.MustRegister(m.PoolsSaturationLevel)
 	reg.MustRegister(m.MonitoredValidatorsCount)
+	reg.MustRegister(m.MissedBlocks)
+	reg.MustRegister(m.ConsecutiveMissedBlocks)
+	reg.MustRegister(m.ValidatedBlocks)
+	reg.MustRegister(m.OrphanedBlocks)
+	reg.MustRegister(m.ExpectedBlocks)
+	reg.MustRegister(m.LatestSlotProcessedByBlockWatcher)
+	reg.MustRegister(m.NextSlotLeader)
 }

internal/metrics/metrics_test.go

@@ -22,12 +22,18 @@ func TestMustRegister(t *testing.T) {
 	metrics.PoolsPledgeMet.WithLabelValues("pool_name", "pool_id", "pool_instance").Set(1)
 	metrics.PoolsSaturationLevel.WithLabelValues("pool_name", "pool_id", "pool_instance").Set(85)
 	metrics.MonitoredValidatorsCount.WithLabelValues("active").Set(10)
+	metrics.MissedBlocks.WithLabelValues("pool_name", "pool_id", "pool_instance", "epoch").Inc()
+	metrics.ConsecutiveMissedBlocks.WithLabelValues("pool_name", "pool_id", "pool_instance", "epoch").Inc()
+	metrics.ValidatedBlocks.WithLabelValues("pool_name", "pool_id", "pool_instance", "epoch").Inc()
+	metrics.OrphanedBlocks.WithLabelValues("pool_name", "pool_id", "pool_instance", "epoch").Inc()
+	metrics.ExpectedBlocks.WithLabelValues("pool_name", "pool_id", "pool_instance", "epoch").Set(2)
+	metrics.NextSlotLeader.WithLabelValues("pool_name", "pool_id", "pool_instance", "epoch").Set(2)
 
 	registry := prometheus.NewRegistry()
 	metrics.MustRegister(registry)
 
 	// The expected number of metrics to be registered, based on the definitions provided in the Collection struct.
-	expectedMetricsCount := 14
+	expectedMetricsCount := 21
 
 	var totalRegisteredMetrics int
 	size, _ := registry.Gather()

internal/watcher/errors.go

@@ -0,0 +1,20 @@
+package watcher
+
+import (
+	"errors"
+	"fmt"
+)
+
+var (
+	ErrBlockFrostAPINotReachable = errors.New("blockfrost API is not reachable")
+	ErrCardanoNodeNotReachable   = errors.New("cardano node is not reachable")
+)
+
+type ErrNoSlotsAssignedToPool struct {
+	PoolID string
+	Epoch  int
+}
+
+func (e *ErrNoSlotsAssignedToPool) Error() string {
+	return fmt.Sprintf("Pool %s has no slots assigned for epoch %d. Consider excluding this pool.", e.PoolID, e.Epoch)
+}