feat:add upload download add bandwidth limit

config/config.go

@@ -15,6 +15,7 @@ import (
 	"github.com/bnb-chain/greenfield-storage-provider/pkg/p2p"
 	"github.com/bnb-chain/greenfield-storage-provider/pkg/pprof"
 	"github.com/bnb-chain/greenfield-storage-provider/service/blocksyncer"
+	"github.com/bnb-chain/greenfield-storage-provider/service/metadata"
 	"github.com/bnb-chain/greenfield-storage-provider/service/signer"
 	"github.com/bnb-chain/greenfield-storage-provider/service/stopserving"
 	"github.com/bnb-chain/greenfield-storage-provider/store/config"
@@ -25,22 +26,25 @@ import (
 
 // StorageProviderConfig defines the configuration of storage provider
 type StorageProviderConfig struct {
-	Service           []string
-	SpOperatorAddress string
-	Endpoint          map[string]string
-	ListenAddress     map[string]string
-	SpDBConfig        *config.SQLDBConfig
-	PieceStoreConfig  *storage.PieceStoreConfig
-	ChainConfig       *gnfd.GreenfieldChainConfig
-	SignerCfg         *signer.SignerConfig
-	BlockSyncerCfg    *blocksyncer.Config
-	P2PCfg            *p2p.NodeConfig
-	LogCfg            *LogConfig
-	MetricsCfg        *metrics.MetricsConfig
-	PProfCfg          *pprof.PProfConfig
-	RateLimiter       *localhttp.RateLimiterConfig
-	BandwidthLimiter  *localhttp.BandWidthLimiterConfig
-	DiscontinueCfg    *stopserving.DiscontinueConfig
+	Service            []string
+	SpOperatorAddress  string
+	Endpoint           map[string]string
+	ListenAddress      map[string]string
+	SpDBConfig         *config.SQLDBConfig
+	BsDBConfig         *config.SQLDBConfig
+	BsDBSwitchedConfig *config.SQLDBConfig
+	PieceStoreConfig   *storage.PieceStoreConfig
+	ChainConfig        *gnfd.GreenfieldChainConfig
+	SignerCfg          *signer.SignerConfig
+	BlockSyncerCfg     *blocksyncer.Config
+	P2PCfg             *p2p.NodeConfig
+	LogCfg             *LogConfig
+	MetricsCfg         *metrics.MetricsConfig
+	PProfCfg           *pprof.PProfConfig
+	RateLimiter        *localhttp.RateLimiterConfig
+	DiscontinueCfg     *stopserving.DiscontinueConfig
+	MetadataCfg        *metadata.MetadataConfig
+	BandwidthLimiter   *localhttp.BandWidthLimiterConfig
 }
 
 // JSONMarshal marshal the StorageProviderConfig to json format
@@ -93,19 +97,22 @@ var DefaultStorageProviderConfig = &StorageProviderConfig{
 		model.P2PService:        model.P2PGRPCAddress,
 		model.AuthService:       model.AuthGRPCAddress,
 	},
-	SpOperatorAddress: hex.EncodeToString([]byte(model.SpOperatorAddress)),
-	SpDBConfig:        DefaultSQLDBConfig,
-	PieceStoreConfig:  DefaultPieceStoreConfig,
-	ChainConfig:       DefaultGreenfieldChainConfig,
-	SignerCfg:         signer.DefaultSignerChainConfig,
-	BlockSyncerCfg:    DefaultBlockSyncerConfig,
-	P2PCfg:            DefaultP2PConfig,
-	LogCfg:            DefaultLogConfig,
-	MetricsCfg:        DefaultMetricsConfig,
-	PProfCfg:          DefaultPProfConfig,
-	RateLimiter:       DefaultRateLimiterConfig,
-	BandwidthLimiter:  DefaultBandwidthLimiterConfig,
-	DiscontinueCfg:    stopserving.DefaultDiscontinueConfig,
+	SpOperatorAddress:  hex.EncodeToString([]byte(model.SpOperatorAddress)),
+	SpDBConfig:         DefaultSQLDBConfig,
+	BsDBConfig:         DefaultBsDBConfig,
+	BsDBSwitchedConfig: DefaultBsDBSwitchedConfig,
+	PieceStoreConfig:   DefaultPieceStoreConfig,
+	ChainConfig:        DefaultGreenfieldChainConfig,
+	SignerCfg:          signer.DefaultSignerChainConfig,
+	BlockSyncerCfg:     DefaultBlockSyncerConfig,
+	P2PCfg:             DefaultP2PConfig,
+	LogCfg:             DefaultLogConfig,
+	MetricsCfg:         DefaultMetricsConfig,
+	PProfCfg:           DefaultPProfConfig,
+	RateLimiter:        DefaultRateLimiterConfig,
+	DiscontinueCfg:     stopserving.DefaultDiscontinueConfig,
+	MetadataCfg:        DefaultMetadataConfig,
+	BandwidthLimiter:   DefaultBandwidthLimiterConfig,
 }
 
 // DefaultSQLDBConfig defines the default configuration of SQL DB
@@ -116,6 +123,22 @@ var DefaultSQLDBConfig = &storeconfig.SQLDBConfig{
 	Database: "storage_provider_db",
 }
 
+// DefaultBsDBConfig defines the default configuration of Bs DB
+var DefaultBsDBConfig = &storeconfig.SQLDBConfig{
+	User:     "root",
+	Passwd:   "test_pwd",
+	Address:  "localhost:3306",
+	Database: "block_syncer",
+}
+
+// DefaultBsDBSwitchedConfig defines the default configuration for the switched Bs DB.
+var DefaultBsDBSwitchedConfig = &storeconfig.SQLDBConfig{
+	User:     "root",
+	Passwd:   "test_pwd",
+	Address:  "localhost:3306",
+	Database: "block_syncer_backup",
+}
+
 // DefaultPieceStoreConfig defines the default configuration of piece store
 var DefaultPieceStoreConfig = &storage.PieceStoreConfig{
 	Shards: 0,
@@ -148,6 +171,12 @@ var DefaultMetricsConfig = &metrics.MetricsConfig{
 	HTTPAddress: model.MetricsHTTPAddress,
 }
 
+// DefaultMetadataConfig defines the default configuration of Metadata service
+var DefaultMetadataConfig = &metadata.MetadataConfig{
+	IsMasterDB:                 true,
+	BsDBSwitchCheckIntervalSec: 3600,
+}
+
 type LogConfig struct {
 	Level string
 	Path  string

config/config_template.toml

@@ -31,6 +31,18 @@ Passwd = "test_pwd"
 Address = "localhost:3306"
 Database = "storage_provider_db"
 
+[BsDBConfig]
+User = "root"
+Passwd = "root"
+Address = "localhost:3306"
+Database = "sp_0"
+
+[BsDBSwitchedConfig]
+User = "root"
+Passwd = "root"
+Address = "localhost:3306"
+Database = "sp_0"
+
 [PieceStoreConfig]
 Shards = 0
 
@@ -94,3 +106,7 @@ RatePeriod = "S"
 
 [StopServingCfg]
 BucketKeepAliveDays = 7
+
+[MetadataCfg]
+IsMasterDB = true
+BsDBSwitchCheckIntervalSec = 3600

config/subconfig.go

@@ -107,9 +107,9 @@ func (cfg *StorageProviderConfig) MakeGatewayConfig() (*gateway.GatewayConfig, e
 			HTTPLimitCfg: cfg.RateLimiter.HTTPLimitCfg,
 		}
 		gCfg.BandWidthLimitCfg = &localhttp.BandWidthLimiterConfig{
-			Enable: true,
-			R:      0.1,
-			B:      1,
+			Enable: false,
+			R:      100,
+			B:      1000,
 		}
 	}
 	return gCfg, nil
@@ -227,7 +227,10 @@ func (cfg *StorageProviderConfig) MakeTaskNodeConfig() (*tasknode.TaskNodeConfig
 // MakeMetadataServiceConfig make meta data service config from StorageProviderConfig
 func (cfg *StorageProviderConfig) MakeMetadataServiceConfig() (*metadata.MetadataConfig, error) {
 	mCfg := &metadata.MetadataConfig{
-		SpDBConfig: cfg.SpDBConfig,
+		BsDBConfig:                 cfg.BsDBConfig,
+		BsDBSwitchedConfig:         cfg.BsDBSwitchedConfig,
+		BsDBSwitchCheckIntervalSec: cfg.MetadataCfg.BsDBSwitchCheckIntervalSec,
+		IsMasterDB:                 cfg.MetadataCfg.IsMasterDB,
 	}
 	if _, ok := cfg.ListenAddress[model.MetadataService]; ok {
 		mCfg.GRPCAddress = cfg.ListenAddress[model.MetadataService]
@@ -243,6 +246,12 @@ func (cfg *StorageProviderConfig) MakeManagerServiceConfig() (*manager.ManagerCo
 		SpOperatorAddress: cfg.SpOperatorAddress,
 		ChainConfig:       cfg.ChainConfig,
 		SpDBConfig:        cfg.SpDBConfig,
+		PieceStoreConfig:  cfg.PieceStoreConfig,
+	}
+	if _, ok := cfg.Endpoint[model.MetadataService]; ok {
+		managerConfig.MetadataGrpcAddress = cfg.Endpoint[model.MetadataService]
+	} else {
+		return nil, fmt.Errorf("missing metadata server gRPC address configuration for manager service")
 	}
 	return managerConfig, nil
 }

model/const.go

@@ -113,6 +113,14 @@ const (
 	BsDBAddress = "BS_DB_ADDRESS"
 	// BsDBDataBase defines env variable name for block syncer db database
 	BsDBDataBase = "BS_DB_DATABASE"
+	// BsDBSwitchedUser defines env variable name for switched block syncer db user name
+	BsDBSwitchedUser = "BS_DB_SWITCHED_USER"
+	// BsDBSwitchedPasswd defines env variable name for switched block syncer db user passwd
+	BsDBSwitchedPasswd = "BS_DB_SWITCHED_PASSWORD"
+	// BsDBSwitchedAddress defines env variable name for switched block syncer db address
+	BsDBSwitchedAddress = "BS_DB_SWITCHED_ADDRESS"
+	// BsDBSwitchedDataBase defines env variable name for switched block syncer db database
+	BsDBSwitchedDataBase = "BS_DB_SWITCHED_DATABASE"
 
 	// SpOperatorAddress defines env variable name for sp operator address
 	SpOperatorAddress = "greenfield-storage-provider"

model/piecestore/piece_key.go

@@ -95,3 +95,35 @@ func ComputeSegmentCount(size uint64, spiltSize uint64) uint32 {
 	}
 	return segmentCount
 }
+
+// GenerateObjectSegmentKeyList generate object's segment piece key list.
+func GenerateObjectSegmentKeyList(objectID, objectSize, segmentSize uint64) []string {
+	var (
+		segmentCount uint32
+		segmentIndex uint32
+		pieceKeyList = make([]string, 0)
+	)
+
+	segmentCount = ComputeSegmentCount(objectSize, segmentSize)
+	for segmentIndex < segmentCount {
+		pieceKeyList = append(pieceKeyList, EncodeSegmentPieceKey(objectID, segmentIndex))
+		segmentIndex++
+	}
+	return pieceKeyList
+}
+
+// GenerateObjectECKeyList generate object's ec piece key list.
+func GenerateObjectECKeyList(objectID, objectSize, segmentSize, redundancyIndex uint64) []string {
+	var (
+		segmentCount uint32
+		segmentIndex uint32
+		pieceKeyList = make([]string, 0)
+	)
+
+	segmentCount = ComputeSegmentCount(objectSize, segmentSize)
+	for segmentIndex < segmentCount {
+		pieceKeyList = append(pieceKeyList, EncodeECPieceKey(objectID, segmentIndex, uint32(redundancyIndex)))
+		segmentIndex++
+	}
+	return pieceKeyList
+}

pkg/middleware/http/tokenBucket.go

@@ -1,9 +1,10 @@
 package http
 
 import (
+	"sync"
+
 	"github.com/bnb-chain/greenfield-storage-provider/pkg/log"
 	"golang.org/x/time/rate"
-	"sync"
 )
 
 type BandWidthLimiterConfig struct {

pkg/p2p/node.go

@@ -37,12 +37,12 @@ type Node struct {
 	stopCh            chan struct{}
 }
 
-// NewNode return an instance of Node
+// NewNode return an instance of Node.
 func NewNode(config *NodeConfig, SPAddr string, signer *signerclient.SignerClient) (*Node, error) {
 	if config.PingPeriod < PingPeriodMin {
 		config.PingPeriod = PingPeriodMin
 	}
-	privKey, localAddr, bootstrapIDs, bootstrapAddrs, err := config.ParseConfing()
+	privKey, localAddr, bootstrapIDs, bootstrapAddrs, err := config.ParseConfig()
 	if err != nil {
 		return nil, err
 	}
@@ -197,7 +197,7 @@ func (n *Node) eventLoop() {
 // broadcast sends request to all p2p nodes
 func (n *Node) broadcast(pc protocol.ID, data proto.Message) {
 	for _, peerID := range n.node.Peerstore().PeersWithAddrs() {
-		log.Debugw("broadcast msg to peer", "peer_id", peerID, "protocol", pc)
+		// log.Debugw("broadcast msg to peer", "peer_id", peerID, "protocol", pc)
 		if strings.Compare(n.node.ID().String(), peerID.String()) == 0 {
 			continue
 		}
@@ -222,7 +222,7 @@ func (n *Node) sendToPeer(peerID peer.ID, pc protocol.ID, data proto.Message) er
 		n.peers.DeletePeer(peerID)
 		return err
 	}
-	log.Debugw("success to send msg", "peer_id", peerID, "protocol", pc, "msg", data.String())
+	// log.Debugw("succeed to send msg", "peer_id", peerID, "protocol", pc, "msg", data.String())
 	s.Close()
 	return err
 }

pkg/p2p/node_config.go

@@ -49,8 +49,8 @@ func (cfg *NodeConfig) overrideConfigFromEnv() {
 	}
 }
 
-// ParseConfing parsers the configuration into a format that go-libp2p can use
-func (cfg *NodeConfig) ParseConfing() (privKey crypto.PrivKey, hostAddr ma.Multiaddr, bootstrapIDs []peer.ID, bootstrapAddrs []ma.Multiaddr, err error) {
+// ParseConfig parsers the configuration into a format that go-libp2p can use
+func (cfg *NodeConfig) ParseConfig() (privKey crypto.PrivKey, hostAddr ma.Multiaddr, bootstrapIDs []peer.ID, bootstrapAddrs []ma.Multiaddr, err error) {
 	cfg.overrideConfigFromEnv()
 	if len(cfg.P2PPrivateKey) > 0 {
 		priKeyBytes, err := hex.DecodeString(cfg.P2PPrivateKey)

pkg/p2p/ping.go

@@ -31,7 +31,7 @@ func (n *Node) onPing(s network.Stream) {
 			log.Warnw("failed to response ping", "peer_id", peerID, "error", err)
 			return
 		}
-		log.Debugw("success to response ping", "peer_id", peerID)
+		// log.Debugw("succeed to response ping", "peer_id", peerID)
 	}()
 	ping := &types.Ping{}
 	buf, err := io.ReadAll(s)
@@ -46,7 +46,7 @@ func (n *Node) onPing(s network.Stream) {
 		log.Errorw("failed to unmarshal ping msg", "error", err)
 		return
 	}
-	log.Debugf("%s received ping request from %s. Message: %s", s.Conn().LocalPeer(), s.Conn().RemotePeer(), ping.String())
+	// log.Debugf("%s received ping request from %s. Message: %s", s.Conn().LocalPeer(), s.Conn().RemotePeer(), ping.String())
 
 	err = types.VerifySignature(ping.GetSpOperatorAddress(), ping.GetSignBytes(), ping.GetSignature())
 	if err != nil {
@@ -66,7 +66,7 @@ func (n *Node) onPing(s network.Stream) {
 			nodeInfo.MultiAddr = append(nodeInfo.MultiAddr, addr.String())
 		}
 		pong.Nodes = append(pong.Nodes, nodeInfo)
-		log.Debugw("send node to remote", "node_id", pID.String(), "remote_node", s.Conn().RemotePeer())
+		// log.Debugw("send node to remote", "node_id", pID.String(), "remote_node", s.Conn().RemotePeer())
 	}
 	pong.SpOperatorAddress = n.SpOperatorAddress
 	pong, err = n.signer.SignPongMsg(context.Background(), pong)
@@ -89,7 +89,7 @@ func (n *Node) onPong(s network.Stream) {
 			log.Warnw("failed to receive pong", "peer_id", peerID, "error", err)
 			return
 		}
-		log.Debugw("success to receive pong", "peer_id", peerID)
+		// log.Debugw("succeed to receive pong", "peer_id", peerID)
 	}()
 	pong := &types.Pong{}
 	buf, err := io.ReadAll(s)
@@ -104,7 +104,7 @@ func (n *Node) onPong(s network.Stream) {
 		log.Errorw("failed to unmarshal ping msg", "error", err)
 		return
 	}
-	log.Debugf("%s received pong request from %s.", s.Conn().LocalPeer(), s.Conn().RemotePeer())
+	// log.Debugf("%s received pong request from %s.", s.Conn().LocalPeer(), s.Conn().RemotePeer())
 
 	err = types.VerifySignature(pong.GetSpOperatorAddress(), pong.GetSignBytes(), pong.GetSignature())
 	if err != nil {
@@ -127,6 +127,6 @@ func (n *Node) onPong(s network.Stream) {
 			addrs = append(addrs, addr)
 		}
 		n.node.Peerstore().AddAddrs(pID, addrs, peerstore.PermanentAddrTTL)
-		log.Debugw("receive node from remote and permanent", "remote_node", s.Conn().RemotePeer(), "node_id", pID)
+		// log.Debugw("receive node from remote and permanent", "remote_node", s.Conn().RemotePeer(), "node_id", pID)
 	}
 }