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

Merge pull request '增加根据package查询node信息' (#4) from feature_sjc into master

This commit is contained in:
baohan 2023-08-24 12:27:14 +08:00
parent fcc5fd1421 18cd88060b
commit 3184a1fa2d
1 changed files with 93 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
internal/services/package.go
View File

@ -6,6 +6,7 @@ import (
"github.com/jmoiron/sqlx"
"gitlink.org.cn/cloudream/common/consts/errorcode"
"gitlink.org.cn/cloudream/common/models"
"gitlink.org.cn/cloudream/common/pkgs/logger"
"gitlink.org.cn/cloudream/common/pkgs/mq"
coormq "gitlink.org.cn/cloudream/storage-common/pkgs/mq/coordinator"
@ -187,3 +188,95 @@ func (svc *Service) DeletePackage(msg *coormq.DeletePackage) (*coormq.DeletePack
return mq.ReplyOK(coormq.NewDeletePackageResp())
}
func (svc *Service) GetCacheNodesByPackage(msg *coormq.GetCacheNodesByPackage) (*coormq.GetCacheNodesByPackageResp, *mq.CodeMessage) {
isAva, err := svc.db.Package().IsAvailable(svc.db.SQLCtx(), msg.UserID, msg.PackageID)
if err != nil {
logger.WithField("UserID", msg.UserID).
WithField("PackageID", msg.PackageID).
Warnf("check package available failed, err: %s", err.Error())
return mq.ReplyFailed[coormq.GetCacheNodesByPackageResp](errorcode.OperationFailed, "check package available failed")
}
if !isAva {
logger.WithField("UserID", msg.UserID).
WithField("PackageID", msg.PackageID).
Warnf("package is not available to the user")
return mq.ReplyFailed[coormq.GetCacheNodesByPackageResp](errorcode.OperationFailed, "package is not available to the user")
}
pkg, err := svc.db.Package().GetByID(svc.db.SQLCtx(), msg.PackageID)
if err != nil {
logger.WithField("PackageID", msg.PackageID).
Warnf("get package: %s", err.Error())
return nil, mq.Failed(errorcode.OperationFailed, "get package failed")
}
redunancyType := pkg.Redundancy.Type
uniqueNodeIDs := make(map[int64]bool)
var nodeIDs []int64
if redunancyType == models.RedundancyRep {
// 备份方式为rep
objectRepDatas, err := svc.db.ObjectRep().GetWithNodeIDInPackage(svc.db.SQLCtx(), msg.PackageID)
if err != nil {
logger.WithField("PackageID", msg.PackageID).
Warnf("get objectRepDatas by packageID failed, err: %s", err.Error())
return mq.ReplyFailed[coormq.GetCacheNodesByPackageResp](errorcode.OperationFailed, "get objectRepDatas by packageID failed")
}
for _, data := range objectRepDatas {
for _, nodeID := range data.NodeIDs {
if !uniqueNodeIDs[nodeID] {
uniqueNodeIDs[nodeID] = true
nodeIDs = append(nodeIDs, nodeID)
}
}
}
} else if redunancyType == models.RedundancyEC {
// 备份方式为ec
objectECDatas, err := svc.db.ObjectBlock().GetWithNodeIDInPackage(svc.db.SQLCtx(), msg.PackageID)
if err != nil {
logger.WithField("PackageID", msg.PackageID).
Warnf("get objectECDatas by packageID failed, err: %s", err.Error())
return mq.ReplyFailed[coormq.GetCacheNodesByPackageResp](errorcode.OperationFailed, "get objectECDatas by packageID failed")
}
for _, ecData := range objectECDatas {
for _, block := range ecData.Blocks {
for _, nodeID := range block.NodeIDs {
if !uniqueNodeIDs[nodeID] {
uniqueNodeIDs[nodeID] = true
nodeIDs = append(nodeIDs, nodeID)
}
}
}
}
} else {
logger.WithField("PackageID", msg.PackageID).
Warnf("Redundancy type %s is wrong", pkg.Redundancy.Type)
return mq.ReplyFailed[coormq.GetCacheNodesByPackageResp](errorcode.OperationFailed, "redundancy type is wrong")
}
return mq.ReplyOK(coormq.NewGetCacheNodesByPackageResp(nodeIDs, redunancyType))
}
func (svc *Service) GetStorageNodesByPackage(msg *coormq.GetStorageNodesByPackage) (*coormq.GetStorageNodesByPackageResp, *mq.CodeMessage) {
storages, err := svc.db.StoragePackage().FindPackageStorages(svc.db.SQLCtx(), msg.PackageID)
if err != nil {
logger.WithField("PackageID", msg.PackageID).
Warnf("get storages by packageID failed, err: %s", err.Error())
return mq.ReplyFailed[coormq.GetStorageNodesByPackageResp](errorcode.OperationFailed, "get storages by packageID failed")
}
uniqueNodeIDs := make(map[int64]bool)
var nodeIDs []int64
for _, stg := range storages {
if !uniqueNodeIDs[stg.NodeID] {
uniqueNodeIDs[stg.NodeID] = true
nodeIDs = append(nodeIDs, stg.NodeID)
}
}
return mq.ReplyOK(coormq.NewGetStorageNodesByPackageResp(nodeIDs))
}