Merge branch 'master' into leo/db-ops-migrate-en-results-to-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

@@ -1458,11 +1458,25 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			defaultConfig.registerDBPruneThreshold,
 			fmt.Sprintf("specifies the number of blocks below the latest stored block height to keep in register db. default: %d", defaultConfig.registerDBPruneThreshold))
 
-		flags.DurationVar(&builder.rpcConf.WebSocketConfig.InactivityTimeout,
+		// websockets config
+		flags.DurationVar(
+			&builder.rpcConf.WebSocketConfig.InactivityTimeout,
 			"websocket-inactivity-timeout",
 			defaultConfig.rpcConf.WebSocketConfig.InactivityTimeout,
-			"specifies the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
-
+			"the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed",
+		)
+		flags.Uint64Var(
+			&builder.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"websocket-max-subscriptions-per-connection",
+			defaultConfig.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"the maximum number of active WebSocket subscriptions allowed per connection",
+		)
+		flags.Float64Var(
+			&builder.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"websocket-max-responses-per-second",
+			defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			fmt.Sprintf("the maximum number of responses that can be sent to a single client per second. Default: %f. if set to 0, no limit is applied to the number of responses per second.", defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond),
+		)
 		flags.BoolVar(
 			&builder.rpcConf.EnableWebSocketsStreamAPI,
 			"experimental-enable-websockets-stream-api",

cmd/access/node_builder/access_node_builder.go

@@ -239,8 +239,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

@@ -820,11 +820,25 @@ func (builder *ObserverServiceBuilder) extraFlags() {
 			defaultConfig.registerDBPruneThreshold,
 			fmt.Sprintf("specifies the number of blocks below the latest stored block height to keep in register db. default: %d", defaultConfig.registerDBPruneThreshold))
 
-		flags.DurationVar(&builder.rpcConf.WebSocketConfig.InactivityTimeout,
+		// websockets config
+		flags.DurationVar(
+			&builder.rpcConf.WebSocketConfig.InactivityTimeout,
 			"websocket-inactivity-timeout",
 			defaultConfig.rpcConf.WebSocketConfig.InactivityTimeout,
-			"specifies the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
-
+			"the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed",
+		)
+		flags.Uint64Var(
+			&builder.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"websocket-max-subscriptions-per-connection",
+			defaultConfig.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"the maximum number of active WebSocket subscriptions allowed per connection",
+		)
+		flags.Float64Var(
+			&builder.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"websocket-max-responses-per-second",
+			defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			fmt.Sprintf("the maximum number of responses that can be sent to a single client per second. Default: %f. if set to 0, no limit is applied to the number of responses per second.", defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond),
+		)
 		flags.BoolVar(
 			&builder.rpcConf.EnableWebSocketsStreamAPI,
 			"experimental-enable-websockets-stream-api",

cmd/observer/node_builder/observer_builder.go

@@ -171,7 +171,8 @@ 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.pebbleDir, "pebbledir", defaultConfig.pebbleDir, "directory to store the public pebble 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)")
 	fnb.flags.StringVarP(&fnb.BaseConfig.level, "loglevel", "l", defaultConfig.level, "level for logging output")
@@ -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

@@ -17,7 +17,7 @@ func InitPebbleDB(logger zerolog.Logger, dir string) (*pebble.DB, io.Closer, err
 	// since we've set an default directory for the pebble DB, this check
 	// is not necessary, but rather a sanity check
 	if dir == "not set" {
-		return nil, nil, fmt.Errorf("missing required flag '--pebbledir'")
+		return nil, nil, fmt.Errorf("missing required flag '--pebble-dir'")
 	}
 
 	// Pre-create DB path

cmd/scaffold/pebble_db.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)
 	},
 }