jhnine
/
JCC-Storage
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化ClientPool定义，方便测试

This commit is contained in:
Sydonian 2023-09-26 10:00:06 +08:00
parent 325111dc69
commit 7745e89829
19 changed files with 81 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
agent/internal/services/mq/storage.go
View File

@ -27,7 +27,7 @@ func (svc *Service) StartStorageLoadPackage(msg *agtmq.StartStorageLoadPackage)
return nil, mq.Failed(errorcode.OperationFailed, "new coordinator client failed")
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
getStgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(msg.UserID, msg.StorageID))
if err != nil {
@ -163,7 +163,7 @@ func (svc *Service) StartStorageCreatePackage(msg *agtmq.StartStorageCreatePacka
return nil, mq.Failed(errorcode.OperationFailed, "new coordinator client failed")
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
getStgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(msg.UserID, msg.StorageID))
if err != nil {

4
agent/internal/task/cache_move_package.go
View File

@ -61,7 +61,7 @@ func (t *CacheMovePackage) do(ctx TaskContext) error {
if err != nil {
return fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
pkgResp, err := coorCli.GetPackage(coormq.NewGetPackage(t.userID, t.packageID))
if err != nil {
@ -76,7 +76,7 @@ func (t *CacheMovePackage) do(ctx TaskContext) error {
return nil
}
func (t *CacheMovePackage) moveRep(ctx TaskContext, coorCli *coormq.PoolClient, pkg model.Package) error {
func (t *CacheMovePackage) moveRep(ctx TaskContext, coorCli *coormq.Client, pkg model.Package) error {
getRepResp, err := coorCli.GetPackageObjectRepData(coormq.NewGetPackageObjectRepData(pkg.PackageID))
if err != nil {
return fmt.Errorf("getting package object rep data: %w", err)

8
client/internal/services/bucket.go
View File

@ -27,7 +27,7 @@ func (svc *BucketService) GetUserBuckets(userID int64) ([]model.Bucket, error) {
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
resp, err := coorCli.GetUserBuckets(coormq.NewGetUserBuckets(userID))
if err != nil {
@ -42,7 +42,7 @@ func (svc *BucketService) GetBucketPackages(userID int64, bucketID int64) ([]mod
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
resp, err := coorCli.GetBucketPackages(coormq.NewGetBucketPackages(userID, bucketID))
if err != nil {
@ -57,7 +57,7 @@ func (svc *BucketService) CreateBucket(userID int64, bucketName string) (int64,
if err != nil {
return 0, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
// TODO 只有阅读了系统操作的源码，才能知道要加哪些锁，但用户的命令可能会调用不止一个系统操作。
// 因此加锁的操作还是必须在用户命令里完成，但具体加锁的内容，则需要被封装起来与系统操作放到一起，方便管理，避免分散改动。
@ -85,7 +85,7 @@ func (svc *BucketService) DeleteBucket(userID int64, bucketID int64) error {
if err != nil {
return fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
// TODO 检查用户是否有删除这个Bucket的权限。检查的时候可以只上UserBucket的Read锁

4
client/internal/services/cacah.go
View File

@ -23,7 +23,7 @@ func (svc *CacheService) StartCacheMovePackage(userID int64, packageID int64, no
if err != nil {
return "", fmt.Errorf("new agent client: %w", err)
}
defer agentCli.Close()
defer stgglb.AgentMQPool.Release(agentCli)
startResp, err := agentCli.StartCacheMovePackage(agtmq.NewStartCacheMovePackage(userID, packageID))
if err != nil {
@ -38,7 +38,7 @@ func (svc *CacheService) WaitCacheMovePackage(nodeID int64, taskID string, waitT
if err != nil {
return true, nil, fmt.Errorf("new agent client: %w", err)
}
defer agentCli.Close()
defer stgglb.AgentMQPool.Release(agentCli)
waitResp, err := agentCli.WaitCacheMovePackage(agtmq.NewWaitCacheMovePackage(taskID, waitTimeout.Milliseconds()))
if err != nil {

12
client/internal/services/package.go
View File

@ -28,7 +28,7 @@ func (svc *PackageService) DownloadPackage(userID int64, packageID int64) (itera
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
mutex, err := reqbuilder.NewBuilder().
// 用于判断用户是否有对象权限
@ -89,7 +89,7 @@ func (svc *PackageService) DownloadPackage(userID int64, packageID int64) (itera
}
}
func (svc *PackageService) downloadRepPackage(packageID int64, objects []model.Object, coorCli *coormq.PoolClient) (*iterator.RepObjectIterator, error) {
func (svc *PackageService) downloadRepPackage(packageID int64, objects []model.Object, coorCli *coormq.Client) (*iterator.RepObjectIterator, error) {
getObjRepDataResp, err := coorCli.GetPackageObjectRepData(coormq.NewGetPackageObjectRepData(packageID))
if err != nil {
return nil, fmt.Errorf("getting package object rep data: %w", err)
@ -101,7 +101,7 @@ func (svc *PackageService) downloadRepPackage(packageID int64, objects []model.O
return iter, nil
}
func (svc *PackageService) downloadECPackage(pkg model.Package, objects []model.Object, coorCli *coormq.PoolClient) (*iterator.ECObjectIterator, error) {
func (svc *PackageService) downloadECPackage(pkg model.Package, objects []model.Object, coorCli *coormq.Client) (*iterator.ECObjectIterator, error) {
getObjECDataResp, err := coorCli.GetPackageObjectECData(coormq.NewGetPackageObjectECData(pkg.PackageID))
if err != nil {
return nil, fmt.Errorf("getting package object ec data: %w", err)
@ -185,7 +185,7 @@ func (svc *PackageService) DeletePackage(userID int64, packageID int64) error {
if err != nil {
return fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
mutex, err := reqbuilder.NewBuilder().
Metadata().
@ -220,7 +220,7 @@ func (svc *PackageService) GetCachedNodes(userID int64, packageID int64) (stgsdk
if err != nil {
return stgsdk.PackageCachingInfo{}, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
resp, err := coorCli.GetPackageCachedNodes(coormq.NewGetPackageCachedNodes(userID, packageID))
if err != nil {
@ -240,7 +240,7 @@ func (svc *PackageService) GetLoadedNodes(userID int64, packageID int64) ([]int6
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
resp, err := coorCli.GetPackageLoadedNodes(coormq.NewGetPackageLoadedNodes(userID, packageID))
if err != nil {

2
client/internal/services/scanner.go
View File

@ -21,7 +21,7 @@ func (svc *ScannerService) PostEvent(event scevt.Event, isEmergency bool, dontMe
if err != nil {
return fmt.Errorf("new scacnner client: %w", err)
}
defer scCli.Close()
defer stgglb.ScannerMQPool.Release(scCli)
err = scCli.PostEvent(scmq.NewPostEvent(event, isEmergency, dontMerge))
if err != nil {

6
client/internal/services/storage.go
View File

@ -44,7 +44,7 @@ func (svc *StorageService) StartStorageCreatePackage(userID int64, bucketID int6
if err != nil {
return 0, "", fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
stgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(userID, storageID))
if err != nil {
@ -55,7 +55,7 @@ func (svc *StorageService) StartStorageCreatePackage(userID int64, bucketID int6
if err != nil {
return 0, "", fmt.Errorf("new agent client: %w", err)
}
defer agentCli.Close()
defer stgglb.AgentMQPool.Release(agentCli)
startResp, err := agentCli.StartStorageCreatePackage(agtmq.NewStartStorageCreatePackage(userID, bucketID, name, storageID, path, redundancy, nodeAffinity))
if err != nil {
@ -71,7 +71,7 @@ func (svc *StorageService) WaitStorageCreatePackage(nodeID int64, taskID string,
// TODO 失败是否要当做任务已经结束？
return true, 0, fmt.Errorf("new agent client: %w", err)
}
defer agentCli.Close()
defer stgglb.AgentMQPool.Release(agentCli)
waitResp, err := agentCli.WaitStorageCreatePackage(agtmq.NewWaitStorageCreatePackage(taskID, waitTimeout.Milliseconds()))
if err != nil {

10
client/internal/task/storage_load_package.go
View File

@ -62,7 +62,7 @@ func (t *StorageLoadPackage) do(ctx TaskContext) error {
if err != nil {
return fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
getStgResp, err := coorCli.GetStorageInfo(coormq.NewGetStorageInfo(t.userID, t.storageID))
if err != nil {
@ -70,13 +70,13 @@ func (t *StorageLoadPackage) do(ctx TaskContext) error {
}
// 然后向代理端发送移动文件的请求
agentClient, err := stgglb.AgentMQPool.Acquire(getStgResp.NodeID)
agentCli, err := stgglb.AgentMQPool.Acquire(getStgResp.NodeID)
if err != nil {
return fmt.Errorf("create agent client to %d failed, err: %w", getStgResp.NodeID, err)
}
defer agentClient.Close()
defer stgglb.AgentMQPool.Release(agentCli)
agentMoveResp, err := agentClient.StartStorageLoadPackage(
agentMoveResp, err := agentCli.StartStorageLoadPackage(
agtmq.NewStartStorageLoadPackage(
t.userID,
t.packageID,
@ -87,7 +87,7 @@ func (t *StorageLoadPackage) do(ctx TaskContext) error {
}
for {
waitResp, err := agentClient.WaitStorageLoadPackage(agtmq.NewWaitStorageLoadPackage(agentMoveResp.TaskID, int64(time.Second)*5))
waitResp, err := agentCli.WaitStorageLoadPackage(agtmq.NewWaitStorageLoadPackage(agentMoveResp.TaskID, int64(time.Second)*5))
if err != nil {
return fmt.Errorf("wait loading package: %w", err)
}

6
common/globals/pools.go
View File

@ -9,11 +9,11 @@ import (
scmq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/scanner"
)
var AgentMQPool *agtmq.Pool
var AgentMQPool agtmq.Pool
var CoordinatorMQPool *coormq.Pool
var CoordinatorMQPool coormq.Pool
var ScannerMQPool *scmq.Pool
var ScannerMQPool scmq.Pool
func InitMQPool(cfg *stgmq.Config) {
AgentMQPool = agtmq.NewPool(cfg)

2
common/pkgs/cmd/create_rep_package.go
View File

@ -275,7 +275,7 @@ func pinIPFSFile(nodeID int64, fileHash string) error {
if err != nil {
return fmt.Errorf("new agent client: %w", err)
}
defer agtCli.Close()
defer stgglb.AgentMQPool.Release(agtCli)
// 然后让最近节点pin本地上传的文件
pinObjResp, err := agtCli.StartPinningObject(agtmq.NewStartPinningObject(fileHash))

6
common/pkgs/cmd/download_package.go
View File

@ -38,7 +38,7 @@ func (t *DownloadPackage) Execute(ctx *DownloadPackageContext) error {
if err != nil {
return fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
getPkgResp, err := coorCli.GetPackage(coormq.NewGetPackage(t.userID, t.packageID))
if err != nil {
@ -65,7 +65,7 @@ func (t *DownloadPackage) downloadRep(ctx *DownloadPackageContext) (iterator.Dow
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
getObjsResp, err := coorCli.GetPackageObjects(coormq.NewGetPackageObjects(t.userID, t.packageID))
if err != nil {
@ -89,7 +89,7 @@ func (t *DownloadPackage) downloadEC(ctx *DownloadPackageContext, pkg model.Pack
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
getObjsResp, err := coorCli.GetPackageObjects(coormq.NewGetPackageObjects(t.userID, t.packageID))
if err != nil {

4
common/pkgs/iterator/ec_object_iterator.go
View File

@ -48,7 +48,7 @@ func (i *ECObjectIterator) MoveNext() (*IterDownloadingObject, error) {
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
if !i.inited {
i.inited = true
@ -69,7 +69,7 @@ func (i *ECObjectIterator) MoveNext() (*IterDownloadingObject, error) {
return item, err
}
func (iter *ECObjectIterator) doMove(coorCli *coormq.PoolClient) (*IterDownloadingObject, error) {
func (iter *ECObjectIterator) doMove(coorCli *coormq.Client) (*IterDownloadingObject, error) {
obj := iter.objects[iter.currentIndex]
ecData := iter.objectECData[iter.currentIndex]

4
common/pkgs/iterator/rep_object_iterator.go
View File

@ -60,7 +60,7 @@ func (i *RepObjectIterator) MoveNext() (*IterDownloadingObject, error) {
if err != nil {
return nil, fmt.Errorf("new coordinator client: %w", err)
}
defer coorCli.Close()
defer stgglb.CoordinatorMQPool.Release(coorCli)
if !i.inited {
i.inited = true
@ -81,7 +81,7 @@ func (i *RepObjectIterator) MoveNext() (*IterDownloadingObject, error) {
return item, err
}
func (i *RepObjectIterator) doMove(coorCli *coormq.PoolClient) (*IterDownloadingObject, error) {
func (i *RepObjectIterator) doMove(coorCli *coormq.Client) (*IterDownloadingObject, error) {
repData := i.objectRepData[i.currentIndex]
if len(repData.NodeIDs) == 0 {
return nil, fmt.Errorf("no node has this file %s", repData.FileHash)

36
common/pkgs/mq/agent/client.go
View File

@ -6,12 +6,12 @@ import (
)
type Client struct {
rabbitCli *mq.RabbitMQClient
rabbitCli *mq.RabbitMQTransport
id int64
}
func NewClient(id int64, cfg *stgmq.Config) (*Client, error) {
rabbitCli, err := mq.NewRabbitMQClient(cfg.MakeConnectingURL(), stgmq.MakeAgentQueueName(id), "")
rabbitCli, err := mq.NewRabbitMQTransport(cfg.MakeConnectingURL(), stgmq.MakeAgentQueueName(id), "")
if err != nil {
return nil, err
}
@ -26,36 +26,24 @@ func (c *Client) Close() {
c.rabbitCli.Close()
}
type PoolClient struct {
*Client
owner *Pool
type Pool interface {
Acquire(id int64) (*Client, error)
Release(cli *Client)
}
func (c *PoolClient) Close() {
c.owner.Release(c)
}
type Pool struct {
type pool struct {
mqcfg *stgmq.Config
}
func NewPool(mqcfg *stgmq.Config) *Pool {
return &Pool{
func NewPool(mqcfg *stgmq.Config) Pool {
return &pool{
mqcfg: mqcfg,
}
}
func (p *Pool) Acquire(id int64) (*PoolClient, error) {
cli, err := NewClient(id, p.mqcfg)
if err != nil {
return nil, err
}
return &PoolClient{
Client: cli,
owner: p,
}, nil
func (p *pool) Acquire(id int64) (*Client, error) {
return NewClient(id, p.mqcfg)
}
func (p *Pool) Release(cli *PoolClient) {
cli.Client.Close()
func (p *pool) Release(cli *Client) {
cli.Close()
}

36
common/pkgs/mq/coordinator/client.go
View File

@ -6,11 +6,11 @@ import (
)
type Client struct {
rabbitCli *mq.RabbitMQClient
rabbitCli *mq.RabbitMQTransport
}
func NewClient(cfg *stgmq.Config) (*Client, error) {
rabbitCli, err := mq.NewRabbitMQClient(cfg.MakeConnectingURL(), stgmq.COORDINATOR_QUEUE_NAME, "")
rabbitCli, err := mq.NewRabbitMQTransport(cfg.MakeConnectingURL(), stgmq.COORDINATOR_QUEUE_NAME, "")
if err != nil {
return nil, err
}
@ -24,36 +24,24 @@ func (c *Client) Close() {
c.rabbitCli.Close()
}
type PoolClient struct {
*Client
owner *Pool
type Pool interface {
Acquire() (*Client, error)
Release(cli *Client)
}
func (c *PoolClient) Close() {
c.owner.Release(c)
}
type Pool struct {
type pool struct {
mqcfg *stgmq.Config
}
func NewPool(mqcfg *stgmq.Config) *Pool {
return &Pool{
func NewPool(mqcfg *stgmq.Config) Pool {
return &pool{
mqcfg: mqcfg,
}
}
func (p *Pool) Acquire() (*PoolClient, error) {
cli, err := NewClient(p.mqcfg)
if err != nil {
return nil, err
}
return &PoolClient{
Client: cli,
owner: p,
}, nil
func (p *pool) Acquire() (*Client, error) {
return NewClient(p.mqcfg)
}
func (p *Pool) Release(cli *PoolClient) {
cli.Client.Close()
func (p *pool) Release(cli *Client) {
cli.Close()
}

36
common/pkgs/mq/scanner/client.go
View File

@ -6,11 +6,11 @@ import (
)
type Client struct {
rabbitCli *mq.RabbitMQClient
rabbitCli *mq.RabbitMQTransport
}
func NewClient(cfg *stgmq.Config) (*Client, error) {
rabbitCli, err := mq.NewRabbitMQClient(cfg.MakeConnectingURL(), stgmq.SCANNER_QUEUE_NAME, "")
rabbitCli, err := mq.NewRabbitMQTransport(cfg.MakeConnectingURL(), stgmq.SCANNER_QUEUE_NAME, "")
if err != nil {
return nil, err
}
@ -24,36 +24,24 @@ func (c *Client) Close() {
c.rabbitCli.Close()
}
type PoolClient struct {
*Client
owner *Pool
type Pool interface {
Acquire() (*Client, error)
Release(cli *Client)
}
func (c *PoolClient) Close() {
c.owner.Release(c)
}
type Pool struct {
type pool struct {
mqcfg *stgmq.Config
}
func NewPool(mqcfg *stgmq.Config) *Pool {
return &Pool{
func NewPool(mqcfg *stgmq.Config) Pool {
return &pool{
mqcfg: mqcfg,
}
}
func (p *Pool) Acquire() (*PoolClient, error) {
cli, err := NewClient(p.mqcfg)
if err != nil {
return nil, err
}
return &PoolClient{
Client: cli,
owner: p,
}, nil
func (p *pool) Acquire() (*Client, error) {
return NewClient(p.mqcfg)
}
func (p *Pool) Release(cli *PoolClient) {
cli.Client.Close()
func (p *pool) Release(cli *Client) {
cli.Close()
}

6
scanner/internal/event/agent_check_cache.go
View File

@ -126,14 +126,14 @@ func (t *AgentCheckCache) startCheck(execCtx ExecuteContext, isComplete bool, ca
log := logger.WithType[AgentCheckCache]("Event")
// 然后向代理端发送移动文件的请求
agentClient, err := stgglb.AgentMQPool.Acquire(t.NodeID)
agtCli, err := stgglb.AgentMQPool.Acquire(t.NodeID)
if err != nil {
log.WithField("NodeID", t.NodeID).Warnf("create agent client failed, err: %s", err.Error())
return
}
defer agentClient.Close()
defer stgglb.AgentMQPool.Release(agtCli)
checkResp, err := agentClient.CheckCache(agtmq.NewCheckCache(isComplete, caches), mq.RequestOption{Timeout: time.Minute})
checkResp, err := agtCli.CheckCache(agtmq.NewCheckCache(isComplete, caches), mq.RequestOption{Timeout: time.Minute})
if err != nil {
log.WithField("NodeID", t.NodeID).Warnf("checking ipfs: %s", err.Error())
return

6
scanner/internal/event/agent_check_state.go
View File

@ -61,14 +61,14 @@ func (t *AgentCheckState) Execute(execCtx ExecuteContext) {
return
}
agentClient, err := stgglb.AgentMQPool.Acquire(t.NodeID)
agtCli, err := stgglb.AgentMQPool.Acquire(t.NodeID)
if err != nil {
log.WithField("NodeID", t.NodeID).Warnf("create agent client failed, err: %s", err.Error())
return
}
defer agentClient.Close()
defer stgglb.AgentMQPool.Release(agtCli)
getResp, err := agentClient.GetState(agtmq.NewGetState(), mq.RequestOption{Timeout: time.Second * 30})
getResp, err := agtCli.GetState(agtmq.NewGetState(), mq.RequestOption{Timeout: time.Second * 30})
if err != nil {
log.WithField("NodeID", t.NodeID).Warnf("getting state: %s", err.Error())

6
scanner/internal/event/agent_check_storage.go
View File

@ -143,14 +143,14 @@ func (t *AgentCheckStorage) startCheck(execCtx ExecuteContext, stg model.Storage
log := logger.WithType[AgentCheckStorage]("Event")
// 投递任务
agentClient, err := stgglb.AgentMQPool.Acquire(stg.NodeID)
agtCli, err := stgglb.AgentMQPool.Acquire(stg.NodeID)
if err != nil {
log.WithField("NodeID", stg.NodeID).Warnf("create agent client failed, err: %s", err.Error())
return
}
defer agentClient.Close()
defer stgglb.AgentMQPool.Release(agtCli)
checkResp, err := agentClient.StorageCheck(agtmq.NewStorageCheck(stg.StorageID, stg.Directory, isComplete, packages), mq.RequestOption{Timeout: time.Minute})
checkResp, err := agtCli.StorageCheck(agtmq.NewStorageCheck(stg.StorageID, stg.Directory, isComplete, packages), mq.RequestOption{Timeout: time.Minute})
if err != nil {
log.WithField("NodeID", stg.NodeID).Warnf("checking storage: %s", err.Error())
return