forked from JointCloud/JCC-CSScheduler
75 lines
2.0 KiB
Go
75 lines
2.0 KiB
Go
package task
|
||
|
||
import (
|
||
"fmt"
|
||
"time"
|
||
|
||
"gitlink.org.cn/cloudream/common/api/storage"
|
||
"gitlink.org.cn/cloudream/common/models"
|
||
"gitlink.org.cn/cloudream/common/pkgs/logger"
|
||
"gitlink.org.cn/cloudream/common/pkgs/task"
|
||
"gitlink.org.cn/cloudream/scheduler/common/globals"
|
||
exectsk "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/executor/task"
|
||
)
|
||
|
||
type StorageCreatePackage struct {
|
||
userID int64
|
||
storageID int64
|
||
path string
|
||
bucketID int64
|
||
name string
|
||
redundancy models.TypedRedundancyInfo
|
||
}
|
||
|
||
func NewStorageCreatePackage(userID int64, storageID int64, path string, bucketID int64, name string, redundancy models.TypedRedundancyInfo) *StorageCreatePackage {
|
||
return &StorageCreatePackage{
|
||
userID: userID,
|
||
storageID: storageID,
|
||
path: path,
|
||
bucketID: bucketID,
|
||
name: name,
|
||
redundancy: redundancy,
|
||
}
|
||
}
|
||
|
||
func (t *StorageCreatePackage) Execute(task *task.Task[TaskContext], ctx TaskContext, complete CompleteFn) {
|
||
log := logger.WithType[StorageCreatePackage]("Task")
|
||
log.Debugf("begin")
|
||
defer log.Debugf("end")
|
||
|
||
err := t.do(task.ID(), ctx)
|
||
if err != nil {
|
||
//TODO 若任务失败，上报的状态failed字段根据情况修改
|
||
ctx.reporter.Report(task.ID(), exectsk.NewCreatePackageTaskStatus("failed", err.Error(), 0))
|
||
}
|
||
ctx.reporter.ReportNow()
|
||
|
||
complete(err, CompleteOption{
|
||
RemovingDelay: time.Minute,
|
||
})
|
||
}
|
||
|
||
func (t *StorageCreatePackage) do(taskID string, ctx TaskContext) error {
|
||
stgCli, err := globals.CloudreamStoragePool.Acquire()
|
||
if err != nil {
|
||
return fmt.Errorf("new cloudream storage client: %w", err)
|
||
}
|
||
defer stgCli.Close()
|
||
|
||
resp, err := stgCli.StorageCreatePackage(storage.StorageCreatePackageReq{
|
||
UserID: t.userID,
|
||
StorageID: t.storageID,
|
||
Path: t.path,
|
||
BucketID: t.bucketID,
|
||
Name: t.name,
|
||
Redundancy: t.redundancy,
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// TODO 根据接口result返回情况修改
|
||
ctx.reporter.Report(taskID, exectsk.NewCreatePackageTaskStatus("completed", "", resp.PackageID))
|
||
return nil
|
||
}
|