Merge pull request #7092 from onflow/leo/util-read-results-receipts

[Util] update util to read results from pebble

Author: zhangchiqing

admin/README.md

@@ -125,3 +125,9 @@ curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"
 ```
 curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"commandName": "create-chunk-data-packs-checkpoint" }'
 ```
+
+### Trigger pebble protocol database checkpoints
+Useful for reading protocol state data from the checkpoints using the read-badger util without stopping the node process.
+```
+curl localhost:9002/admin/run_command -H 'Content-Type: application/json' -d '{"commandName": "create-pebble-checkpoint" }'
+```

admin/commands/storage/chunk_data_packs_db_checkpoint.go

@@ -0,0 +1,59 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"path"
+	"time"
+
+	"github.com/cockroachdb/pebble"
+	"github.com/rs/zerolog/log"
+
+	"github.com/onflow/flow-go/admin"
+	"github.com/onflow/flow-go/admin/commands"
+)
+
+var _ commands.AdminCommand = (*PebbleDBCheckpointCommand)(nil)
+
+// PebbleDBCheckpointCommand creates a checkpoint for pebble database for querying the data
+// while keeping the node alive.
+type PebbleDBCheckpointCommand struct {
+	checkpointDir string
+	dbname        string // dbname is for logging purposes only
+	pebbleDB      *pebble.DB
+}
+
+func NewPebbleDBCheckpointCommand(checkpointDir string, dbname string, pebbleDB *pebble.DB) *PebbleDBCheckpointCommand {
+	return &PebbleDBCheckpointCommand{
+		checkpointDir: checkpointDir,
+		dbname:        dbname,
+		pebbleDB:      pebbleDB,
+	}
+}
+
+func (c *PebbleDBCheckpointCommand) Handler(ctx context.Context, req *admin.CommandRequest) (interface{}, error) {
+	log.Info().Msgf("admintool: creating %v database checkpoint", c.dbname)
+
+	targetDir := nextTmpFolder(c.checkpointDir)
+
+	log.Info().Msgf("admintool: creating %v database checkpoint at: %v", c.dbname, targetDir)
+
+	err := c.pebbleDB.Checkpoint(targetDir)
+	if err != nil {
+		return nil, admin.NewInvalidAdminReqErrorf("failed to create %v pebbledb checkpoint at %v: %w", c.dbname, targetDir, err)
+	}
+
+	log.Info().Msgf("admintool: successfully created %v database checkpoint at: %v", c.dbname, targetDir)
+
+	return fmt.Sprintf("successfully created %v db checkpoint at %v", c.dbname, targetDir), nil
+}
+
+func (c *PebbleDBCheckpointCommand) Validator(req *admin.CommandRequest) error {
+	return nil
+}
+
+func nextTmpFolder(dir string) string {
+	// use timestamp as folder name
+	folderName := time.Now().Format("2006-01-02_15-04-05")
+	return path.Join(dir, folderName)
+}

admin/commands/storage/pebble_checkpoint.go

@@ -231,8 +231,8 @@ func (builder *ExecutionNodeBuilder) LoadComponentsAndModules() {
 		// Load the admin tool when chunk data packs db are initialized in execution state
 		AdminCommand("create-chunk-data-packs-checkpoint", func(config *NodeConfig) commands.AdminCommand {
 			// by default checkpoints will be created under "/data/chunk_data_packs_checkpoints_dir"
-			return storageCommands.NewChunksCheckpointCommand(exeNode.exeConf.chunkDataPackCheckpointsDir,
-				exeNode.chunkDataPackDB)
+			return storageCommands.NewPebbleDBCheckpointCommand(exeNode.exeConf.chunkDataPackCheckpointsDir,
+				"chunk_data_pack", exeNode.chunkDataPackDB)
 		}).
 		Component("stop control", exeNode.LoadStopControl).
 		Component("execution state ledger WAL compactor", exeNode.LoadExecutionStateLedgerWALCompactor).

cmd/execution_builder.go

@@ -155,6 +155,7 @@ type BaseConfig struct {
 	DynamicStartupSleepInterval time.Duration
 	datadir                     string
 	pebbleDir                   string
+	pebbleCheckpointsDir        string
 	dbops                       string
 	badgerDB                    *badger.DB
 	pebbleDB                    *pebble.DB

cmd/node_builder.go

@@ -171,6 +171,7 @@ func (fnb *FlowNodeBuilder) BaseFlags() {
 	fnb.flags.StringVar(&fnb.BaseConfig.BindAddr, "bind", defaultConfig.BindAddr, "address to bind on")
 	fnb.flags.StringVarP(&fnb.BaseConfig.BootstrapDir, "bootstrapdir", "b", defaultConfig.BootstrapDir, "path to the bootstrap directory")
 	fnb.flags.StringVarP(&fnb.BaseConfig.datadir, "datadir", "d", defaultConfig.datadir, "directory to store the public database (protocol state)")
+	fnb.flags.StringVar(&fnb.BaseConfig.pebbleCheckpointsDir, "pebble-checkpoints-dir", defaultConfig.pebbleCheckpointsDir, "directory to store the checkpoints for the public pebble database (protocol state)")
 	fnb.flags.StringVar(&fnb.BaseConfig.pebbleDir, "pebble-dir", defaultConfig.pebbleDir, "directory to store the public pebble database (protocol state)")
 	fnb.flags.StringVar(&fnb.BaseConfig.secretsdir, "secretsdir", defaultConfig.secretsdir, "directory to store private database (secrets)")
 	fnb.flags.StringVar(&fnb.BaseConfig.dbops, "dbops", defaultConfig.dbops, "database operations to use (badger-transaction, batch-update, pebble-update)")
@@ -2102,6 +2103,9 @@ func (fnb *FlowNodeBuilder) RegisterDefaultAdminCommands() {
 		return storageCommands.NewReadSealsCommand(config.State, config.Storage.Seals, config.Storage.Index)
 	}).AdminCommand("get-latest-identity", func(config *NodeConfig) commands.AdminCommand {
 		return common.NewGetIdentityCommand(config.IdentityProvider)
+	}).AdminCommand("create-pebble-checkpoint", func(config *NodeConfig) commands.AdminCommand {
+		// by default checkpoints will be created under "/data/protocol_pebble_checkpoints"
+		return storageCommands.NewPebbleDBCheckpointCommand(config.pebbleCheckpointsDir, "protocol", config.PebbleDB)
 	})
 }
 

cmd/scaffold.go

@@ -1,12 +1,16 @@
 package cmd
 
 import (
+	"fmt"
+
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
 
-	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	findBlockByCommits "github.com/onflow/flow-go/cmd/util/cmd/read-badger/cmd/find-block-by-commits"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/metrics"
+	"github.com/onflow/flow-go/storage"
+	"github.com/onflow/flow-go/storage/store"
 )
 
 func init() {
@@ -22,25 +26,28 @@ var commitsCmd = &cobra.Command{
 	Use:   "commits",
 	Short: "get commit by block ID",
 	Run: func(cmd *cobra.Command, args []string) {
-		_, db := InitStorages()
-		defer db.Close()
+		err := WithStorage(func(db storage.DB) error {
 
-		commits := common.InitExecutionStorages(db).Commits
+			commits := store.NewCommits(metrics.NewNoopCollector(), db)
 
-		log.Info().Msgf("got flag block id: %s", flagBlockID)
-		blockID, err := flow.HexStringToIdentifier(flagBlockID)
-		if err != nil {
-			log.Error().Err(err).Msg("malformed block id")
-			return
-		}
+			log.Info().Msgf("got flag block id: %s", flagBlockID)
+			blockID, err := flow.HexStringToIdentifier(flagBlockID)
+			if err != nil {
+				return fmt.Errorf("malformed block id: %w", err)
+			}
+
+			log.Info().Msgf("getting commit by block id: %v", blockID)
+			commit, err := commits.ByBlockID(blockID)
+			if err != nil {
+				return fmt.Errorf("could not get commit for block id: %v: %w", blockID, err)
+			}
+
+			log.Info().Msgf("commit: %x", commit)
+			return nil
+		})
 
-		log.Info().Msgf("getting commit by block id: %v", blockID)
-		commit, err := commits.ByBlockID(blockID)
 		if err != nil {
-			log.Error().Err(err).Msgf("could not get commit for block id: %v", blockID)
-			return
+			log.Error().Err(err).Msg("could not get events")
 		}
-
-		log.Info().Msgf("commit: %x", commit)
 	},
 }

cmd/util/cmd/read-badger/cmd/commits.go

@@ -1,11 +1,16 @@
 package cmd
 
 import (
+	"fmt"
+
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
 
 	"github.com/onflow/flow-go/cmd/util/cmd/common"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/metrics"
+	"github.com/onflow/flow-go/storage"
+	"github.com/onflow/flow-go/storage/store"
 )
 
 var flagEventType string
@@ -23,81 +28,79 @@ func init() {
 
 var eventsCmd = &cobra.Command{
 	Use:   "events",
-	Short: "Read events from badger",
+	Short: "Read events",
 	Run: func(cmd *cobra.Command, args []string) {
-		_, db := InitStorages()
-		defer db.Close()
-
-		events := common.InitExecutionStorages(db).Events
-
-		if flagEventType != "" && flagTransactionID != "" {
-			log.Error().Msg("provide only one of --transaction-id or --event-type")
-			return
-		}
-
-		log.Info().Msgf("got flag block id: %s", flagBlockID)
-		blockID, err := flow.HexStringToIdentifier(flagBlockID)
-		if err != nil {
-			log.Error().Err(err).Msg("malformed block id")
-			return
-		}
+		err := WithStorage(func(db storage.DB) error {
+			events := store.NewEvents(metrics.NewNoopCollector(), db)
 
-		if flagTransactionID != "" {
-			log.Info().Msgf("got flag transaction id: %s", flagTransactionID)
-			transactionID, err := flow.HexStringToIdentifier(flagTransactionID)
-			if err != nil {
-				log.Error().Err(err).Msg("malformed transaction id")
-				return
+			if flagEventType != "" && flagTransactionID != "" {
+				return fmt.Errorf("provide only one of --transaction-id or --event-type")
 			}
 
-			log.Info().Msgf("getting events for block id: %v, transaction id: %v", blockID, transactionID)
-			events, err := events.ByBlockIDTransactionID(blockID, transactionID)
+			log.Info().Msgf("got flag block id: %s", flagBlockID)
+			blockID, err := flow.HexStringToIdentifier(flagBlockID)
 			if err != nil {
-				log.Error().Err(err).Msgf("could not get events for block id: %v, transaction id: %v", blockID, transactionID)
-				return
+				return fmt.Errorf("malformed block id: %w", err)
 			}
 
-			for _, event := range events {
-				common.PrettyPrint(event)
-			}
-			return
-		}
+			if flagTransactionID != "" {
+				log.Info().Msgf("got flag transaction id: %s", flagTransactionID)
+				transactionID, err := flow.HexStringToIdentifier(flagTransactionID)
+				if err != nil {
+					return fmt.Errorf("malformed traansaction id: %w", err)
+				}
 
-		if flagEventType != "" {
-			validEvents := map[string]bool{
-				"flow.AccountCreated": true,
-				"flow.AccountUpdated": true,
-				"flow.EpochCommit":    true,
-				"flow.EpochSetup":     true,
-			}
-			if _, ok := validEvents[flagEventType]; ok {
-				log.Info().Msgf("getting events for block id: %v, event type: %s", blockID, flagEventType)
-				events, err := events.ByBlockIDEventType(blockID, flow.EventType(flagEventType))
+				log.Info().Msgf("getting events for block id: %v, transaction id: %v", blockID, transactionID)
+				events, err := events.ByBlockIDTransactionID(blockID, transactionID)
 				if err != nil {
-					log.Error().Err(err).Msgf("could not get events for block id: %v, event type: %s", blockID, flagEventType)
-					return
+					return fmt.Errorf("could not get events for block id: %v, transaction id: %v: %w", blockID, transactionID, err)
 				}
 
 				for _, event := range events {
 					common.PrettyPrint(event)
 				}
-				return
+				return nil
 			}
 
-			log.Fatal().Msgf("not a valid event type: %s", flagEventType)
-			return
-		}
+			if flagEventType != "" {
+				validEvents := map[string]bool{
+					"flow.AccountCreated": true,
+					"flow.AccountUpdated": true,
+					"flow.EpochCommit":    true,
+					"flow.EpochSetup":     true,
+				}
+				if _, ok := validEvents[flagEventType]; ok {
+					log.Info().Msgf("getting events for block id: %v, event type: %s", blockID, flagEventType)
+					events, err := events.ByBlockIDEventType(blockID, flow.EventType(flagEventType))
+					if err != nil {
+						return fmt.Errorf("could not get events for block id: %v, event type: %s, %w", blockID, flagEventType, err)
+					}
+
+					for _, event := range events {
+						common.PrettyPrint(event)
+					}
+					return nil
+				}
 
-		// just fetch events for block
-		log.Info().Msgf("getting events for block id: %v", blockID)
-		evts, err := events.ByBlockID(blockID)
-		if err != nil {
-			log.Error().Err(err).Msgf("could not get events for block id: %v", blockID)
-			return
-		}
+				return fmt.Errorf("not a valid event type: %s", flagEventType)
+			}
+
+			// just fetch events for block
+			log.Info().Msgf("getting events for block id: %v", blockID)
+			evts, err := events.ByBlockID(blockID)
+			if err != nil {
+				return fmt.Errorf("could not get events for block id: %v: %w", blockID, err)
+			}
 
-		for _, event := range evts {
-			common.PrettyPrint(event)
+			for _, event := range evts {
+				common.PrettyPrint(event)
+			}
+
+			return nil
+		})
+
+		if err != nil {
+			log.Error().Err(err).Msg("could not get events")
 		}
 	},
 }