mirror of
https://github.com/yunionio/cloudpods.git
synced 2026-09-24 16:03:43 +08:00
fix: task add progress status (#20275)
Co-authored-by: Qiu Jian <qiujian@yunionyun.com>
This commit is contained in:
@@ -23,6 +23,7 @@ import (
|
||||
type TaskListInput struct {
|
||||
ModelBaseListInput
|
||||
ProjectizedResourceListInput
|
||||
StatusResourceBaseListInput
|
||||
|
||||
Id []string `json:"id" help:"id filter"`
|
||||
ObjId []string `json:"obj_id" help:"object id filter"`
|
||||
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
|
||||
"yunion.io/x/jsonutils"
|
||||
|
||||
"yunion.io/x/onecloud/pkg/cloudcommon/db"
|
||||
"yunion.io/x/onecloud/pkg/cloudcommon/db/quotas"
|
||||
"yunion.io/x/onecloud/pkg/mcclient"
|
||||
)
|
||||
@@ -33,6 +34,8 @@ type ITask interface {
|
||||
GetUserCred() mcclient.TokenCredential
|
||||
GetTaskId() string
|
||||
SetStage(stageName string, data *jsonutils.JSONDict) error
|
||||
GetObject() db.IStandaloneModel
|
||||
GetObjects() []db.IStandaloneModel
|
||||
|
||||
GetTaskRequestHeader() http.Header
|
||||
|
||||
@@ -41,4 +44,7 @@ type ITask interface {
|
||||
|
||||
GetPendingUsage(quota quotas.IQuota, index int) error
|
||||
ClearPendingUsage(index int) error
|
||||
|
||||
SetProgressAndStatus(progress float32, status string) error
|
||||
SetProgress(progress float32) error
|
||||
}
|
||||
|
||||
@@ -75,6 +75,7 @@ const (
|
||||
type STaskManager struct {
|
||||
db.SModelBaseManager
|
||||
db.SProjectizedResourceBaseManager
|
||||
db.SStatusResourceBaseManager
|
||||
}
|
||||
|
||||
var TaskManager *STaskManager
|
||||
@@ -89,6 +90,7 @@ func init() {
|
||||
type STask struct {
|
||||
db.SModelBase
|
||||
db.SProjectizedResourceBase
|
||||
db.SStatusResourceBase
|
||||
|
||||
// 资源创建时间
|
||||
CreatedAt time.Time `nullable:"false" created_at:"true" index:"true" get:"user" list:"user" json:"created_at"`
|
||||
@@ -1030,6 +1032,10 @@ func (manager *STaskManager) ListItemFilter(
|
||||
if err != nil {
|
||||
return q, errors.Wrap(err, "SResourceBaseManager.ListItemFilter")
|
||||
}
|
||||
q, err = manager.SStatusResourceBaseManager.ListItemFilter(ctx, q, userCred, input.StatusResourceBaseListInput)
|
||||
if err != nil {
|
||||
return q, errors.Wrap(err, "SStatusResourceBaseManager.ListItemFilter")
|
||||
}
|
||||
|
||||
if len(input.Id) > 0 {
|
||||
q = q.In("id", input.Id)
|
||||
@@ -1164,3 +1170,26 @@ func (manager *STaskManager) OrderByExtraFields(
|
||||
// }
|
||||
return q, nil
|
||||
}
|
||||
|
||||
func (task *STask) SetProgressAndStatus(progress float32, status string) error {
|
||||
_, err := db.Update(task, func() error {
|
||||
task.SetProgressValue(progress)
|
||||
task.SetStatusValue(status)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "Update")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (task *STask) SetProgress(progress float32) error {
|
||||
_, err := db.Update(task, func() error {
|
||||
task.SetProgressValue(progress)
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "Update")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -87,7 +87,7 @@ func ensureBucket() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func Put(ctx context.Context, filePath, objName string) (string, error) {
|
||||
func PutStream(ctx context.Context, file io.Reader, fSize int64, objName string) (string, error) {
|
||||
if client == nil {
|
||||
return "", ErrClientNotInit
|
||||
}
|
||||
@@ -95,7 +95,16 @@ func Put(ctx context.Context, filePath, objName string) (string, error) {
|
||||
if err != nil {
|
||||
return "", errors.Wrap(err, "client.getBucket")
|
||||
}
|
||||
const blockSizeMB = 100
|
||||
err = cloudprovider.UploadObject(ctx, bucket, objName, blockSizeMB*1000*1000, file, fSize, cloudprovider.ACLPrivate, "", nil, false)
|
||||
if err != nil {
|
||||
return "", errors.Wrap(err, "cloudprovider.UploadObject")
|
||||
}
|
||||
log.Debugf("put object %s size %d", objName, fSize)
|
||||
return client.Location(objName), nil
|
||||
}
|
||||
|
||||
func Put(ctx context.Context, filePath, objName string) (string, error) {
|
||||
finfo, err := os.Stat(filePath)
|
||||
if err != nil {
|
||||
return "", errors.Wrap(err, "os.Stat")
|
||||
@@ -106,13 +115,7 @@ func Put(ctx context.Context, filePath, objName string) (string, error) {
|
||||
return "", errors.Wrap(err, "os.Open")
|
||||
}
|
||||
defer file.Close()
|
||||
const blockSizeMB = 100
|
||||
err = cloudprovider.UploadObject(ctx, bucket, objName, blockSizeMB*1000*1000, file, fSize, cloudprovider.ACLPrivate, "", nil, false)
|
||||
if err != nil {
|
||||
return "", errors.Wrap(err, "cloudprovider.UploadObject")
|
||||
}
|
||||
log.Debugf("put object %s size %d", objName, fSize)
|
||||
return client.Location(objName), nil
|
||||
return PutStream(ctx, file, fSize, objName)
|
||||
}
|
||||
|
||||
func Get(ctx context.Context, fileName string) (int64, io.ReadCloser, error) {
|
||||
|
||||
Reference in New Issue
Block a user