diff --git a/pkg/apis/tasks.go b/pkg/apis/tasks.go index 62265367e0..8abc374d24 100644 --- a/pkg/apis/tasks.go +++ b/pkg/apis/tasks.go @@ -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"` diff --git a/pkg/cloudcommon/db/taskman/interface.go b/pkg/cloudcommon/db/taskman/interface.go index 9d2974b035..333a3a199a 100644 --- a/pkg/cloudcommon/db/taskman/interface.go +++ b/pkg/cloudcommon/db/taskman/interface.go @@ -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 } diff --git a/pkg/cloudcommon/db/taskman/tasks.go b/pkg/cloudcommon/db/taskman/tasks.go index cd929d3a66..0819c9eb5b 100644 --- a/pkg/cloudcommon/db/taskman/tasks.go +++ b/pkg/cloudcommon/db/taskman/tasks.go @@ -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 +} diff --git a/pkg/image/drivers/s3/minio.go b/pkg/image/drivers/s3/minio.go index 8b5289c71c..b7821835a5 100644 --- a/pkg/image/drivers/s3/minio.go +++ b/pkg/image/drivers/s3/minio.go @@ -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) {