Merge pull request #5 from kilnfi/feat/add-block-watcher

feat: add a new watcher for monitoring block validation performance

Author: rayanebel

.gitignore

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

.mockery.yaml

@@ -1,6 +1,15 @@
 with-expecter: true
 dir: "{{.InterfaceDir}}/mocks/"
 packages:
+  github.com/kilnfi/cardano-validator-watcher/internal/cardano:
+    interfaces:
+      CardanoClient:
+  github.com/kilnfi/cardano-validator-watcher/internal/cardano/cardanocli:
+    interfaces:
+      CommandExecutor:
   github.com/kilnfi/cardano-validator-watcher/internal/blockfrost:
     interfaces:
-      Client:
+      Client:
+  github.com/kilnfi/cardano-validator-watcher/internal/slotleader:
+    interfaces:
+      SlotLeader:

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) {

go.mod

@@ -5,13 +5,17 @@ go 1.22.5
 replace github.com/mitchellh/mapstructure => github.com/go-viper/mapstructure v1.6.0
 
 require (
+	github.com/DATA-DOG/go-sqlmock v1.5.2
 	github.com/blockfrost/blockfrost-go v0.2.2
 	github.com/dgraph-io/ristretto v0.2.0
+	github.com/jmoiron/sqlx v1.4.0
+	github.com/mattn/go-sqlite3 v1.14.24
+	github.com/pressly/goose/v3 v3.24.0
 	github.com/prometheus/client_golang v1.19.1
 	github.com/spf13/cobra v1.8.0
 	github.com/spf13/viper v1.19.0
-	github.com/stretchr/testify v1.9.0
-	golang.org/x/sync v0.7.0
+	github.com/stretchr/testify v1.10.0
+	golang.org/x/sync v0.10.0
 )
 
 require (
@@ -25,6 +29,7 @@ require (
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
+	github.com/mfridman/interpolate v0.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
@@ -34,6 +39,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
+	github.com/sethvargo/go-retry v0.3.0 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.11.0 // indirect
 	github.com/spf13/cast v1.6.0 // indirect
@@ -42,8 +48,8 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 // indirect
-	golang.org/x/sys v0.20.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect