Merge branch 'master' into fxamacker/close-pebbledb-batch

Author: fxamacker

cmd/access/node_builder/access_node_builder.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/execution_builder.go

@@ -54,6 +54,7 @@ import (
 	"github.com/onflow/flow-go/engine/execution/ingestion/fetcher"
 	"github.com/onflow/flow-go/engine/execution/ingestion/stop"
 	"github.com/onflow/flow-go/engine/execution/ingestion/uploader"
+	"github.com/onflow/flow-go/engine/execution/migration"
 	exeprovider "github.com/onflow/flow-go/engine/execution/provider"
 	exepruner "github.com/onflow/flow-go/engine/execution/pruner"
 	"github.com/onflow/flow-go/engine/execution/rpc"
@@ -92,10 +93,13 @@ import (
 	"github.com/onflow/flow-go/state/protocol/blocktimer"
 	storageerr "github.com/onflow/flow-go/storage"
 	storage "github.com/onflow/flow-go/storage/badger"
+	"github.com/onflow/flow-go/storage/dbops"
 	"github.com/onflow/flow-go/storage/operation"
+	"github.com/onflow/flow-go/storage/operation/badgerimpl"
 	"github.com/onflow/flow-go/storage/operation/pebbleimpl"
 	storagepebble "github.com/onflow/flow-go/storage/pebble"
 	"github.com/onflow/flow-go/storage/store"
+	"github.com/onflow/flow-go/storage/store/chained"
 )
 
 const (
@@ -136,13 +140,17 @@ type ExecutionNode struct {
 	registerStore  *storehouse.RegisterStore
 
 	// storage
-	events        storageerr.Events
-	serviceEvents storageerr.ServiceEvents
-	txResults     storageerr.TransactionResults
-	results       storageerr.ExecutionResults
-	receipts      storageerr.ExecutionReceipts
-	myReceipts    storageerr.MyExecutionReceipts
-	commits       storageerr.Commits
+	events          storageerr.Events
+	eventsReader    storageerr.EventsReader
+	serviceEvents   storageerr.ServiceEvents
+	txResults       storageerr.TransactionResults
+	txResultsReader storageerr.TransactionResultsReader
+	results         storageerr.ExecutionResults
+	resultsReader   storageerr.ExecutionResultsReader
+	receipts        storageerr.ExecutionReceipts
+	myReceipts      storageerr.MyExecutionReceipts
+	commits         storageerr.Commits
+	commitsReader   storageerr.CommitsReader
 
 	chunkDataPackDB        *pebble.DB
 	chunkDataPacks         storageerr.ChunkDataPacks
@@ -216,11 +224,11 @@ func (builder *ExecutionNodeBuilder) LoadComponentsAndModules() {
 		Module("blobservice peer manager dependencies", exeNode.LoadBlobservicePeerManagerDependencies).
 		Module("bootstrap", exeNode.LoadBootstrapper).
 		Module("register store", exeNode.LoadRegisterStore).
+		Module("migrate last executed block", exeNode.MigrateLastSealedExecutedResultToPebble).
 		AdminCommand("get-transactions", func(conf *NodeConfig) commands.AdminCommand {
 			return storageCommands.NewGetTransactionsCommand(conf.State, conf.Storage.Payloads, exeNode.collections)
 		}).
 		Component("execution state ledger", exeNode.LoadExecutionStateLedger).
-
 		// TODO: Modules should be able to depends on components
 		// Because all modules are always bootstrapped first, before components,
 		// its not possible to have a module depending on a Component.
@@ -331,15 +339,32 @@ func (exeNode *ExecutionNode) LoadExecutionStorage(
 	node *NodeConfig,
 ) error {
 	db := node.ProtocolDB
+
+	exeNode.events = store.NewEvents(node.Metrics.Cache, db)
+	exeNode.serviceEvents = store.NewServiceEvents(node.Metrics.Cache, db)
 	exeNode.commits = store.NewCommits(node.Metrics.Cache, db)
 	exeNode.results = store.NewExecutionResults(node.Metrics.Cache, db)
 	exeNode.receipts = store.NewExecutionReceipts(node.Metrics.Cache, db, exeNode.results, storage.DefaultCacheSize)
 	exeNode.myReceipts = store.NewMyExecutionReceipts(node.Metrics.Cache, db, exeNode.receipts)
-
-	// Needed for gRPC server, make sure to assign to main scoped vars
-	exeNode.events = store.NewEvents(node.Metrics.Cache, db)
-	exeNode.serviceEvents = store.NewServiceEvents(node.Metrics.Cache, db)
 	exeNode.txResults = store.NewTransactionResults(node.Metrics.Cache, db, exeNode.exeConf.transactionResultsCacheSize)
+
+	if dbops.IsBadgerBased(node.dbops) {
+		// if data are stored in badger, we can use the same storage for all data
+		exeNode.eventsReader = exeNode.events
+		exeNode.commitsReader = exeNode.commits
+		exeNode.resultsReader = exeNode.results
+		exeNode.txResultsReader = exeNode.txResults
+	} else if dbops.IsPebbleBatch(node.dbops) {
+		// when data are stored in pebble, we need to use chained storage to query data from
+		// both pebble and badger
+		// note the pebble storage is the first argument, and badger storage is the second, so
+		// the data will be queried from pebble first, then badger
+		badgerDB := badgerimpl.ToDB(node.DB)
+		exeNode.eventsReader = chained.NewEvents(exeNode.events, store.NewEvents(node.Metrics.Cache, badgerDB))
+		exeNode.commitsReader = chained.NewCommits(exeNode.commits, store.NewCommits(node.Metrics.Cache, badgerDB))
+		exeNode.resultsReader = chained.NewExecutionResults(exeNode.results, store.NewExecutionResults(node.Metrics.Cache, badgerDB))
+		exeNode.txResultsReader = chained.NewTransactionResults(exeNode.txResults, store.NewTransactionResults(node.Metrics.Cache, badgerDB, exeNode.exeConf.transactionResultsCacheSize))
+	}
 	return nil
 }
 
@@ -470,6 +495,7 @@ func (exeNode *ExecutionNode) LoadGCPBlockDataUploader(
 	)
 
 	// Setting up RetryableUploader for GCP uploader
+	// deprecated
 	retryableUploader := uploader.NewBadgerRetryableUploaderWrapper(
 		asyncUploader,
 		node.Storage.Blocks,
@@ -730,6 +756,16 @@ func (exeNode *ExecutionNode) LoadExecutionDataGetter(node *NodeConfig) error {
 	return nil
 }
 
+func (exeNode *ExecutionNode) MigrateLastSealedExecutedResultToPebble(node *NodeConfig) error {
+	// Migrate the last sealed executed
+	err := migration.MigrateLastSealedExecutedResultToPebble(node.Logger, node.DB, node.PebbleDB, node.State, node.RootSeal)
+	if err != nil {
+		return fmt.Errorf("could not migrate last sealed executed result to pebble: %w", err)
+	}
+
+	return nil
+}
+
 func (exeNode *ExecutionNode) LoadExecutionState(
 	node *NodeConfig,
 ) (
@@ -765,6 +801,8 @@ func (exeNode *ExecutionNode) LoadExecutionState(
 	exeNode.chunkDataPackDB = chunkDataPackDB
 	exeNode.chunkDataPacks = chunkDataPacks
 
+	// migrate execution data for last sealed and executed block
+
 	exeNode.executionState = state.NewExecutionState(
 		exeNode.ledgerStorage,
 		exeNode.commits,
@@ -1353,10 +1391,10 @@ func (exeNode *ExecutionNode) LoadGrpcServer(
 		exeNode.scriptsEng,
 		node.Storage.Headers,
 		node.State,
-		exeNode.events,
-		exeNode.results,
-		exeNode.txResults,
-		exeNode.commits,
+		exeNode.eventsReader,
+		exeNode.resultsReader,
+		exeNode.txResultsReader,
+		exeNode.commitsReader,
 		exeNode.metricsProvider,
 		node.RootChainID,
 		signature.NewBlockSignerDecoder(exeNode.committee),
@@ -1370,7 +1408,10 @@ func (exeNode *ExecutionNode) LoadBootstrapper(node *NodeConfig) error {
 	// check if the execution database already exists
 	bootstrapper := bootstrap.NewBootstrapper(node.Logger)
 
-	commit, bootstrapped, err := bootstrapper.IsBootstrapped(node.ProtocolDB)
+	// in order to support switching from badger to pebble in the middle of the spork,
+	// we will check if the execution database has been bootstrapped by reading the state from badger db.
+	// and if not, bootstrap both badger and pebble db.
+	commit, bootstrapped, err := bootstrapper.IsBootstrapped(badgerimpl.ToDB(node.DB))
 	if err != nil {
 		return fmt.Errorf("could not query database to know whether database has been bootstrapped: %w", err)
 	}
@@ -1395,7 +1436,12 @@ func (exeNode *ExecutionNode) LoadBootstrapper(node *NodeConfig) error {
 			return fmt.Errorf("could not load bootstrap state from checkpoint file: %w", err)
 		}
 
-		err = bootstrapper.BootstrapExecutionDatabase(node.ProtocolDB, node.RootSeal)
+		err = bootstrapper.BootstrapExecutionDatabase(badgerimpl.ToDB(node.DB), node.RootSeal)
+		if err != nil {
+			return fmt.Errorf("could not bootstrap execution database: %w", err)
+		}
+
+		err = bootstrapper.BootstrapExecutionDatabase(pebbleimpl.ToDB(node.PebbleDB), node.RootSeal)
 		if err != nil {
 			return fmt.Errorf("could not bootstrap execution database: %w", err)
 		}

cmd/observer/node_builder/observer_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/util/cmd/read-badger/cmd/stats.go

@@ -0,0 +1,58 @@
+package cmd
+
+import (
+	"fmt"
+	"runtime"
+
+	"github.com/rs/zerolog/log"
+	"github.com/spf13/cobra"
+
+	"github.com/onflow/flow-go/cmd/util/cmd/common"
+	"github.com/onflow/flow-go/storage"
+	"github.com/onflow/flow-go/storage/operation"
+	"github.com/onflow/flow-go/storage/operation/badgerimpl"
+	"github.com/onflow/flow-go/storage/operation/pebbleimpl"
+	"github.com/onflow/flow-go/storage/pebble"
+)
+
+var flagDBType string
+
+func init() {
+	rootCmd.AddCommand(statsCmd)
+	statsCmd.Flags().StringVar(&flagDBType, "dbtype", "badger", "database type to use (badger or pebble)")
+}
+
+var statsCmd = &cobra.Command{
+	Use:   "stats",
+	Short: "get stats for the database, such as key count, total value size, min/max value size etc",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		var sdb storage.DB
+		if flagDBType == "badger" {
+			db := common.InitStorage(flagDatadir)
+			defer db.Close()
+			sdb = badgerimpl.ToDB(db)
+		} else if flagDBType == "pebble" {
+			pdb, err := pebble.MustOpenDefaultPebbleDB(log.Logger, flagPebbleDir)
+			if err != nil {
+				return fmt.Errorf("failed to open pebble db: %w", err)
+			}
+			defer pdb.Close()
+			sdb = pebbleimpl.ToDB(pdb)
+		} else {
+			return fmt.Errorf("invalid db type")
+		}
+
+		numWorkers := runtime.NumCPU()
+		if numWorkers > 256 {
+			numWorkers = 256
+		}
+		log.Info().Msgf("getting stats for %s db at %s with %v workers0", flagDBType, flagDatadir, numWorkers)
+		stats, err := operation.SummarizeKeysByFirstByteConcurrent(log.Logger, sdb.Reader(), numWorkers)
+		if err != nil {
+			return fmt.Errorf("failed to get stats: %w", err)
+		}
+
+		operation.PrintStats(log.Logger, stats)
+		return nil
+	},
+}

engine/access/rest/http/request/address.go engine/access/rest/common/parser/address.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/address_test.go engine/access/rest/common/parser/address_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"strings"

engine/access/rest/http/request/arguments.go engine/access/rest/common/parser/arguments.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"
@@ -7,6 +7,7 @@ import (
 )
 
 const maxArgumentsLength = 100
+const MaxAllowedScriptArguments = 100
 
 type Arguments [][]byte
 
@@ -25,7 +26,7 @@ func (a *Arguments) Parse(raw []string) error {
 	}
 
 	if len(args) > maxArgumentsLength {
-		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments)
+		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments)
 	}
 
 	*a = args

engine/access/rest/http/request/arguments_test.go engine/access/rest/common/parser/arguments_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"
@@ -21,13 +21,13 @@ func TestArguments_InvalidParse(t *testing.T) {
 		assert.EqualError(t, err, "invalid argument encoding: illegal base64 data at input byte 0", a)
 	}
 
-	tooLong := make([]string, maxAllowedScriptArguments+1)
+	tooLong := make([]string, MaxAllowedScriptArguments+1)
 	for i := range tooLong {
 		tooLong[i] = "dGVzdA=="
 	}
 
 	err := arguments.Parse(tooLong)
-	assert.EqualError(t, err, fmt.Sprintf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments))
+	assert.EqualError(t, err, fmt.Sprintf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments))
 }
 
 func TestArguments_ValidParse(t *testing.T) {

engine/access/rest/http/request/proposal_key.go engine/access/rest/common/parser/proposal_key.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/signature_test.go engine/access/rest/common/parser/signature_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"encoding/hex"

engine/access/rest/http/request/signatures.go engine/access/rest/common/parser/signatures.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"