jhnine
/
JCC-CSScheduler
forked from JointCloud/JCC-CSScheduler
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

分布定义存储系统的ID类型

This commit is contained in:
Sydonian 2023-11-30 14:41:28 +08:00
parent 0f5973ee52
commit fe9a12f92b
17 changed files with 81 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
advisor/internal/scheduler/scheduler.go
View File

@ -8,6 +8,7 @@ import (
"github.com/samber/lo"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
uopsdk "gitlink.org.cn/cloudream/common/sdks/unifyops"
"gitlink.org.cn/cloudream/common/utils/math"
@ -362,7 +363,7 @@ func (s *DefaultScheduler) calcResourceLevel(avai float64, need float64) int {
// 计算节点得分情况
func (s *DefaultScheduler) calcFileScore(files jobmod.JobFiles, allCCs map[schsdk.CCID]*candidate) error {
// 只计算运控返回的计算中心上的存储服务的数据权重
cdsNodeToCC := make(map[int64]*candidate)
cdsNodeToCC := make(map[cdssdk.NodeID]*candidate)
for _, cc := range allCCs {
cdsNodeToCC[cc.CC.CDSNodeID] = cc
}
@ -407,7 +408,7 @@ func (s *DefaultScheduler) calcFileScore(files jobmod.JobFiles, allCCs map[schsd
}
// 计算package在各节点的得分情况
func (s *DefaultScheduler) calcPackageFileScore(packageID int64, cdsNodeToCC map[int64]*candidate) (map[schsdk.CCID]*fileDetail, error) {
func (s *DefaultScheduler) calcPackageFileScore(packageID cdssdk.PackageID, cdsNodeToCC map[cdssdk.NodeID]*candidate) (map[schsdk.CCID]*fileDetail, error) {
colCli, err := schglb.CollectorMQPool.Acquire()
if err != nil {
return nil, fmt.Errorf("new collector client: %w", err)
@ -458,7 +459,7 @@ func (s *DefaultScheduler) calcPackageFileScore(packageID int64, cdsNodeToCC map
}
// 计算package在各节点的得分情况
func (s *DefaultScheduler) calcImageFileScore(imageID schsdk.ImageID, allCCs map[schsdk.CCID]*candidate, cdsNodeToCC map[int64]*candidate) (map[schsdk.CCID]*fileDetail, error) {
func (s *DefaultScheduler) calcImageFileScore(imageID schsdk.ImageID, allCCs map[schsdk.CCID]*candidate, cdsNodeToCC map[cdssdk.NodeID]*candidate) (map[schsdk.CCID]*fileDetail, error) {
colCli, err := schglb.CollectorMQPool.Acquire()
if err != nil {
return nil, fmt.Errorf("new collector client: %w", err)

9
client/internal/http/jobset.go
View File

@ -8,6 +8,7 @@ import (
"gitlink.org.cn/cloudream/common/consts/errorcode"
"gitlink.org.cn/cloudream/common/pkgs/logger"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
"gitlink.org.cn/cloudream/common/utils/serder"
)
@ -57,10 +58,10 @@ func (s *JobSetService) Submit(ctx *gin.Context) {
}
type JobSetLocalFileUploadedReq struct {
JobSetID schsdk.JobSetID `json:"jobSetID" binding:"required"`
LocalPath string `json:"localPath" binding:"required"`
Error string `json:"error"`
PackageID int64 `json:"packageID"`
JobSetID schsdk.JobSetID `json:"jobSetID" binding:"required"`
LocalPath string `json:"localPath" binding:"required"`
Error string `json:"error"`
PackageID cdssdk.PackageID `json:"packageID"`
}
func (s *JobSetService) LocalFileUploaded(ctx *gin.Context) {

7
client/internal/prescheduler/default_prescheduler.go
View File

@ -8,6 +8,7 @@ import (
"github.com/samber/lo"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
uopsdk "gitlink.org.cn/cloudream/common/sdks/unifyops"
"gitlink.org.cn/cloudream/common/utils/math"
@ -534,7 +535,7 @@ func (s *DefaultPreScheduler) calcResourceLevel(avai float64, need float64) int
// 计算节点得分情况
func (s *DefaultPreScheduler) calcFileScore(files schsdk.JobFilesInfo, allCCs map[schsdk.CCID]*candidate) error {
// 只计算运控返回的可用计算中心上的存储服务的数据权重
cdsNodeToCC := make(map[int64]*candidate)
cdsNodeToCC := make(map[cdssdk.NodeID]*candidate)
for _, cc := range allCCs {
cdsNodeToCC[cc.CC.CDSNodeID] = cc
}
@ -586,7 +587,7 @@ func (s *DefaultPreScheduler) calcFileScore(files schsdk.JobFilesInfo, allCCs ma
}
// 计算package在各节点的得分情况
func (s *DefaultPreScheduler) calcPackageFileScore(packageID int64, cdsNodeToCC map[int64]*candidate) (map[schsdk.CCID]*fileDetail, error) {
func (s *DefaultPreScheduler) calcPackageFileScore(packageID cdssdk.PackageID, cdsNodeToCC map[cdssdk.NodeID]*candidate) (map[schsdk.CCID]*fileDetail, error) {
colCli, err := schglb.CollectorMQPool.Acquire()
if err != nil {
return nil, fmt.Errorf("new collector client: %w", err)
@ -639,7 +640,7 @@ func (s *DefaultPreScheduler) calcPackageFileScore(packageID int64, cdsNodeToCC
}
// 计算package在各节点的得分情况
func (s *DefaultPreScheduler) calcImageFileScore(imageID schsdk.ImageID, allCCs map[schsdk.CCID]*candidate, cdsNodeToCC map[int64]*candidate) (map[schsdk.CCID]*fileDetail, error) {
func (s *DefaultPreScheduler) calcImageFileScore(imageID schsdk.ImageID, allCCs map[schsdk.CCID]*candidate, cdsNodeToCC map[cdssdk.NodeID]*candidate) (map[schsdk.CCID]*fileDetail, error) {
colCli, err := schglb.CollectorMQPool.Acquire()
if err != nil {
return nil, fmt.Errorf("new collector client: %w", err)

3
client/internal/services/jobset.go
View File

@ -4,6 +4,7 @@ import (
"fmt"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
schglb "gitlink.org.cn/cloudream/scheduler/common/globals"
mgrmq "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/manager"
)
@ -38,7 +39,7 @@ func (svc *JobSetService) Submit(info schsdk.JobSetInfo) (schsdk.JobSetID, *schs
}
// 任务集中某个文件上传完成
func (svc *JobSetService) LocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, errMsg string, packageID int64) error {
func (svc *JobSetService) LocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, errMsg string, packageID cdssdk.PackageID) error {
mgrCli, err := schglb.ManagerMQPool.Acquire()
if err != nil {
return fmt.Errorf("new manager client: %w", err)

9
common/models/job/normal_job.go
View File

@ -2,6 +2,7 @@ package jobmod
import (
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
)
type NormalJob struct {
@ -35,11 +36,11 @@ type JobFiles struct {
}
type PackageJobFile struct {
PackageID int64 `json:"packageID"`
FullPath string `json:"fullPath"` // Load之后的完整文件路径
PackageID cdssdk.PackageID `json:"packageID"`
FullPath string `json:"fullPath"` // Load之后的完整文件路径
}
type ImageJobFile struct {
PackageID *int64 `json:"packageID"`
ImageID schsdk.ImageID `json:"imageID"`
PackageID *cdssdk.PackageID `json:"packageID"`
ImageID schsdk.ImageID `json:"imageID"`
}

7
common/models/job/resource_job.go
View File

@ -1,11 +1,14 @@
package jobmod
import schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
import (
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
)
type ResourceJob struct {
JobBase
Info schsdk.ResourceJobInfo `json:"info"`
ResourcePackageID int64 `json:"resourcePackageID"` // 回源之后得到的PackageID
ResourcePackageID cdssdk.PackageID `json:"resourcePackageID"` // 回源之后得到的PackageID
}
func NewResourceJob(jobSetID schsdk.JobSetID, jobID schsdk.JobID, info schsdk.ResourceJobInfo) *ResourceJob {

7
common/models/models.go
View File

@ -6,6 +6,7 @@ import (
pcmsdk "gitlink.org.cn/cloudream/common/sdks/pcm"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
uopsdk "gitlink.org.cn/cloudream/common/sdks/unifyops"
myreflect "gitlink.org.cn/cloudream/common/utils/reflect"
"gitlink.org.cn/cloudream/common/utils/serder"
@ -23,9 +24,9 @@ type ComputingCenter struct {
// 计算中心在PCM系统的ID
PCMParticipantID pcmsdk.ParticipantID `json:"pcmParticipantID" db:"PCMParticipantID"`
// 计算中心在存储系统的ID
CDSNodeID int64 `json:"cdsNodeID" db:"CDSNodeID"`
CDSNodeID cdssdk.NodeID `json:"cdsNodeID" db:"CDSNodeID"`
// 此算力中心的存储服务对应在存储系统中的ID
CDSStorageID int64 `json:"cdsStorageID" db:"CDSStorageID"`
CDSStorageID cdssdk.StorageID `json:"cdsStorageID" db:"CDSStorageID"`
// 计算中心名称
Name string `json:"name" db:"Name"`
}
@ -34,7 +35,7 @@ type Image struct {
// 调度系统内的镜像ID
ImageID schsdk.ImageID `json:"imageID" db:"ImageID"`
// 镜像文件对应的存储系统PackageID，可以为空，为空则代表此镜像不可被自动导入到算力中心，比如是预制镜像
CDSPackageID *int64 `json:"cdsPackageID" db:"CDSPackageID"`
CDSPackageID *cdssdk.PackageID `json:"cdsPackageID" db:"CDSPackageID"`
// 镜像名称，在调度系统上设置的
Name string `json:"name" db:"Name"`
// 镜像创建时间

3
common/pkgs/db/image.go
View File

@ -5,6 +5,7 @@ import (
"github.com/jmoiron/sqlx"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
schmod "gitlink.org.cn/cloudream/scheduler/common/models"
)
@ -22,7 +23,7 @@ func (*ImageDB) GetByID(ctx SQLContext, id schsdk.ImageID) (schmod.Image, error)
return ret, err
}
func (*ImageDB) Create(ctx SQLContext, cdsPackageID *int64, name string, createTime time.Time) (schsdk.ImageID, error) {
func (*ImageDB) Create(ctx SQLContext, cdsPackageID *cdssdk.PackageID, name string, createTime time.Time) (schsdk.ImageID, error) {
ret, err := ctx.Exec("insert into Image(CDSPackageID, Name, CreateTime) values(?, ?, ?)", cdsPackageID, name, createTime)
if err != nil {
return 0, err

16
common/pkgs/mq/collector/storage.go
View File

@ -16,15 +16,15 @@ var _ = Register(Service.PackageGetCachedStgNodes)
type PackageGetCachedStgNodes struct {
mq.MessageBodyBase
UserID int64 `json:"userID"`
PackageID int64 `json:"packageID"`
UserID cdssdk.UserID `json:"userID"`
PackageID cdssdk.PackageID `json:"packageID"`
}
type PackageGetCachedStgNodesResp struct {
mq.MessageBodyBase
cdssdk.PackageCachingInfo
}
func NewPackageGetCachedStgNodes(userID int64, packageID int64) *PackageGetCachedStgNodes {
func NewPackageGetCachedStgNodes(userID cdssdk.UserID, packageID cdssdk.PackageID) *PackageGetCachedStgNodes {
return &PackageGetCachedStgNodes{
UserID: userID,
PackageID: packageID,
@ -48,21 +48,21 @@ var _ = Register(Service.PackageGetLoadedStgNodes)
type PackageGetLoadedStgNodes struct {
mq.MessageBodyBase
UserID int64 `json:"userID"`
PackageID int64 `json:"packageID"`
UserID cdssdk.UserID `json:"userID"`
PackageID cdssdk.PackageID `json:"packageID"`
}
type PackageGetLoadedStgNodesResp struct {
mq.MessageBodyBase
StgNodeIDs []int64 `json:"stgNodeIDs"`
StgNodeIDs []cdssdk.NodeID `json:"stgNodeIDs"`
}
func NewPackageGetLoadedStgNodes(userID int64, packageID int64) *PackageGetLoadedStgNodes {
func NewPackageGetLoadedStgNodes(userID cdssdk.UserID, packageID cdssdk.PackageID) *PackageGetLoadedStgNodes {
return &PackageGetLoadedStgNodes{
UserID: userID,
PackageID: packageID,
}
}
func NewPackageGetLoadedStgNodesResp(nodeIDs []int64) *PackageGetLoadedStgNodesResp {
func NewPackageGetLoadedStgNodesResp(nodeIDs []cdssdk.NodeID) *PackageGetLoadedStgNodesResp {
return &PackageGetLoadedStgNodesResp{
StgNodeIDs: nodeIDs,
}

8
common/pkgs/mq/executor/task/cache_move_package.go
View File

@ -4,9 +4,9 @@ import cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
type CacheMovePackage struct {
TaskInfoBase
UserID int64 `json:"userID"`
PackageID int64 `json:"packageID"`
StgNodeID int64 `json:"stgNodeID"`
UserID cdssdk.UserID `json:"userID"`
PackageID cdssdk.PackageID `json:"packageID"`
StgNodeID cdssdk.NodeID `json:"stgNodeID"`
}
type CacheMovePackageStatus struct {
TaskStatusBase
@ -14,7 +14,7 @@ type CacheMovePackageStatus struct {
CacheInfos []cdssdk.ObjectCacheInfo `json:"cacheInfos"`
}
func NewCacheMovePackage(userID int64, packageID int64, stgNodeID int64) *CacheMovePackage {
func NewCacheMovePackage(userID cdssdk.UserID, packageID cdssdk.PackageID, stgNodeID cdssdk.NodeID) *CacheMovePackage {
return &CacheMovePackage{
UserID: userID,
PackageID: packageID,

32
common/pkgs/mq/executor/task/storage_create_package.go
View File

@ -4,31 +4,29 @@ import cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
type StorageCreatePackage struct {
TaskInfoBase
UserID int64 `json:"userID"`
StorageID int64 `json:"storageID"`
Path string `json:"path"`
BucketID int64 `json:"bucketID"`
Name string `json:"name"`
Redundancy cdssdk.TypedRedundancyInfo `json:"redundancy"`
UserID cdssdk.UserID `json:"userID"`
StorageID cdssdk.StorageID `json:"storageID"`
Path string `json:"path"`
BucketID cdssdk.BucketID `json:"bucketID"`
Name string `json:"name"`
}
type StorageCreatePackageStatus struct {
TaskStatusBase
Status string `json:"status"`
Error string `json:"error"`
PackageID int64 `json:"packageID"`
Status string `json:"status"`
Error string `json:"error"`
PackageID cdssdk.PackageID `json:"packageID"`
}
func NewStorageCreatePackage(userID int64, storageID int64, filePath string, bucketID int64, name string, redundancy cdssdk.TypedRedundancyInfo) *StorageCreatePackage {
func NewStorageCreatePackage(userID cdssdk.UserID, storageID cdssdk.StorageID, filePath string, bucketID cdssdk.BucketID, name string) *StorageCreatePackage {
return &StorageCreatePackage{
UserID: userID,
StorageID: storageID,
Path: filePath,
BucketID: bucketID,
Name: name,
Redundancy: redundancy,
UserID: userID,
StorageID: storageID,
Path: filePath,
BucketID: bucketID,
Name: name,
}
}
func NewStorageCreatePackageStatus(status string, err string, packageID int64) *StorageCreatePackageStatus {
func NewStorageCreatePackageStatus(status string, err string, packageID cdssdk.PackageID) *StorageCreatePackageStatus {
return &StorageCreatePackageStatus{
Status: status,
Error: err,

10
common/pkgs/mq/executor/task/storage_load_package.go
View File

@ -1,10 +1,12 @@
package task
import cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
type StorageLoadPackage struct {
TaskInfoBase
UserID int64 `json:"userID"`
PackageID int64 `json:"packageID"`
StorageID int64 `json:"storageID"`
UserID cdssdk.UserID `json:"userID"`
PackageID cdssdk.PackageID `json:"packageID"`
StorageID cdssdk.StorageID `json:"storageID"`
}
type StorageLoadPackageStatus struct {
TaskStatusBase
@ -12,7 +14,7 @@ type StorageLoadPackageStatus struct {
FullPath string `json:"fullPath"`
}
func NewStorageLoadPackage(userID int64, packageID int64, storageID int64) *StorageLoadPackage {
func NewStorageLoadPackage(userID cdssdk.UserID, packageID cdssdk.PackageID, storageID cdssdk.StorageID) *StorageLoadPackage {
return &StorageLoadPackage{
UserID: userID,
PackageID: packageID,

11
common/pkgs/mq/manager/job.go
View File

@ -3,6 +3,7 @@ package manager
import (
"gitlink.org.cn/cloudream/common/pkgs/mq"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
jobmod "gitlink.org.cn/cloudream/scheduler/common/models/job"
)
@ -49,16 +50,16 @@ var _ = Register(Service.JobSetLocalFileUploaded)
type JobSetLocalFileUploaded struct {
mq.MessageBodyBase
JobSetID schsdk.JobSetID `json:"jobSetID"`
LocalPath string `json:"localPath"`
Error string `json:"error"` // 如果上传文件失败，那么这个字段说明了失败原因
PackageID int64 `json:"packageID"` // 如果上传文件成功，那么这个字段是上传之后得到的PackageID
JobSetID schsdk.JobSetID `json:"jobSetID"`
LocalPath string `json:"localPath"`
Error string `json:"error"` // 如果上传文件失败，那么这个字段说明了失败原因
PackageID cdssdk.PackageID `json:"packageID"` // 如果上传文件成功，那么这个字段是上传之后得到的PackageID
}
type JobSetLocalFileUploadedResp struct {
mq.MessageBodyBase
}
func NewJobSetLocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, err string, packageID int64) *JobSetLocalFileUploaded {
func NewJobSetLocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, err string, packageID cdssdk.PackageID) *JobSetLocalFileUploaded {
return &JobSetLocalFileUploaded{
JobSetID: jobSetID,
LocalPath: localPath,

5
common/utils/utils.go
View File

@ -7,10 +7,11 @@ import (
"time"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
)
func MakeJobOutputFullPath(stgDir string, userID int64, jobID schsdk.JobID) string {
return filepath.Join(stgDir, strconv.FormatInt(userID, 10), "jobs", string(jobID), "output")
func MakeJobOutputFullPath(stgDir string, userID cdssdk.UserID, jobID schsdk.JobID) string {
return filepath.Join(stgDir, strconv.FormatInt(int64(userID), 10), "jobs", string(jobID), "output")
}
func MakeResourcePackageName(jobID schsdk.JobID) string {

9
manager/internal/jobmgr/event/local_file_uploaded.go
View File

@ -1,16 +1,19 @@
package event
import schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
import (
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
)
// 本地文件上传结束
type LocalFileUploaded struct {
JobSetID schsdk.JobSetID
LocalPath string
Error string
PackageID int64
PackageID cdssdk.PackageID
}
func NewLocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, err string, packageID int64) *LocalFileUploaded {
func NewLocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, err string, packageID cdssdk.PackageID) *LocalFileUploaded {
return &LocalFileUploaded{
JobSetID: jobSetID,
LocalPath: localPath,

1
manager/internal/jobmgr/executing_handler.go
View File

@ -186,7 +186,6 @@ func (h *ExecutingHandler) onResourceJobEvent(evt event.Event, job *executingJob
tarNorJob.OutputFullPath,
resJob.Info.BucketID,
utils.MakeResourcePackageName(resJob.JobID),
resJob.Info.Redundancy,
))
if err != nil {
h.changeJobState(job.job, jobmod.NewStateFailed(err.Error(), job.state))

3
manager/internal/jobmgr/jobmgr.go
View File

@ -9,6 +9,7 @@ import (
"gitlink.org.cn/cloudream/common/pkgs/logger"
schsdk "gitlink.org.cn/cloudream/common/sdks/scheduler"
cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
myreflect "gitlink.org.cn/cloudream/common/utils/reflect"
jobmod "gitlink.org.cn/cloudream/scheduler/common/models/job"
"gitlink.org.cn/cloudream/scheduler/common/pkgs/db"
@ -170,7 +171,7 @@ func (m *Manager) SubmitJobSet(jobSetInfo schsdk.JobSetInfo, preScheduleScheme j
return jobSet, nil
}
func (m *Manager) LocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, err string, packageID int64) error {
func (m *Manager) LocalFileUploaded(jobSetID schsdk.JobSetID, localPath string, err string, packageID cdssdk.PackageID) error {
m.pubLock.Lock()
defer m.pubLock.Unlock()