From 8abda214331fc82cd73b5522503a4eeae96b781a Mon Sep 17 00:00:00 2001 From: Jian Qiu Date: Wed, 29 May 2024 14:41:21 +0800 Subject: [PATCH] fix: task improvements (#20379) Co-authored-by: Qiu Jian --- pkg/appsrv/workers.go | 24 ++++ pkg/cloudcommon/consts/db.go | 19 +++ pkg/cloudcommon/db/taskman/localtaskworker.go | 22 ++- pkg/cloudcommon/db/taskman/subtasks.go | 36 ++++- pkg/cloudcommon/db/taskman/taskobjs.go | 25 ++++ pkg/cloudcommon/db/taskman/tasks.go | 135 ++++++++++++++---- pkg/cloudcommon/db/taskman/worker.go | 29 ++-- pkg/cloudcommon/options/changes.go | 6 + pkg/cloudcommon/options/options.go | 7 +- pkg/compute/service/service.go | 11 +- .../tasks/guest_save_instance_image_task.go | 9 +- .../tasks/instance_backup_create_task.go | 4 +- .../tasks/instance_backup_syncstatus.go | 2 +- pkg/image/drivers/s3/minio.go | 12 +- pkg/image/models/image_subs.go | 2 +- pkg/image/models/images.go | 4 +- pkg/image/models/storage.go | 26 ++-- pkg/mcclient/modulebase/modules.go | 135 +++++++++++++----- 18 files changed, 394 insertions(+), 114 deletions(-) diff --git a/pkg/appsrv/workers.go b/pkg/appsrv/workers.go index f0802389e4..871c7c9573 100644 --- a/pkg/appsrv/workers.go +++ b/pkg/appsrv/workers.go @@ -181,6 +181,9 @@ type SWorkerManager struct { ignoreOverflow bool cancelPrevIdent bool + + queueInitHook func() error + queueEmptyHook func() error } func NewWorkerManager(name string, workerCount int, backlog int, dbWorker bool) *SWorkerManager { @@ -222,6 +225,14 @@ type sWorkerTask struct { start time.Time } +func (wm *SWorkerManager) SetQueueInitHook(f func() error) { + wm.queueInitHook = f +} + +func (wm *SWorkerManager) SetQueueEmptyHook(f func() error) { + wm.queueEmptyHook = f +} + func (wm *SWorkerManager) EnableCancelPreviousIdenticalTask() { wm.cancelPrevIdent = true } @@ -260,6 +271,12 @@ func (wm *SWorkerManager) removeWorker(worker *SWorker) { } else { wm.detachedWorker.removeWithLock(worker) } + if wm.activeWorker.size()+wm.detachedWorker.size() == 0 && wm.queueEmptyHook != nil { + err := wm.queueEmptyHook() + if err != nil { + log.Errorf("queueEmptyHook fail %s", err) + } + } } func execCallback(task *sWorkerTask) { @@ -283,6 +300,13 @@ func (wm *SWorkerManager) schedule() { } func (wm *SWorkerManager) scheduleWithLock() { + if wm.activeWorker.size()+wm.detachedWorker.size() == 0 && wm.queueInitHook != nil { + err := wm.queueInitHook() + if err != nil { + log.Errorf("queueInitHook fail %s", err) + return + } + } queueSize := wm.queue.Size() if wm.activeWorker.size() < wm.workerCount && queueSize > 0 { wm.workerId += 1 diff --git a/pkg/cloudcommon/consts/db.go b/pkg/cloudcommon/consts/db.go index 57aa780b36..fb0cb15b57 100644 --- a/pkg/cloudcommon/consts/db.go +++ b/pkg/cloudcommon/consts/db.go @@ -26,6 +26,9 @@ var ( defaultDBConnectionString string defaultDBChecksumHashAlgorithm string + + taskWorkerCount int + localTaskWorkerCount int ) func SetDefaultDB(dialect, connStr string) { @@ -52,3 +55,19 @@ func DefaultDBChecksumHashAlgorithm() string { } return "sha256" } + +func SetTaskWorkerCount(cnt int) { + taskWorkerCount = cnt +} + +func SetLocalTaskWorkerCount(cnt int) { + localTaskWorkerCount = cnt +} + +func TaskWorkerCount() int { + return taskWorkerCount +} + +func LocalTaskWorkerCount() int { + return localTaskWorkerCount +} diff --git a/pkg/cloudcommon/db/taskman/localtaskworker.go b/pkg/cloudcommon/db/taskman/localtaskworker.go index 48af21be29..b57abf402d 100644 --- a/pkg/cloudcommon/db/taskman/localtaskworker.go +++ b/pkg/cloudcommon/db/taskman/localtaskworker.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "runtime/debug" + "sync" "yunion.io/x/jsonutils" "yunion.io/x/log" @@ -25,13 +26,15 @@ import ( "yunion.io/x/pkg/util/version" "yunion.io/x/onecloud/pkg/appsrv" + "yunion.io/x/onecloud/pkg/cloudcommon/consts" "yunion.io/x/onecloud/pkg/mcclient/modules/yunionconf" ) var localTaskWorkerMan *appsrv.SWorkerManager +var localTaskWorkerManLock *sync.Mutex func init() { - localTaskWorkerMan = appsrv.NewWorkerManager("LocalTaskWorkerManager", 4, 1024, false) + localTaskWorkerManLock = &sync.Mutex{} } func Error2TaskData(err error) jsonutils.JSONObject { @@ -47,9 +50,6 @@ type localTask struct { } func (t *localTask) Run() { - log.Debugf("XXXXXXXXXXXXXXXXXXLOCAL TASK RUN STARTXXXXXXXXXXXXXXXXX") - defer log.Debugf("XXXXXXXXXXXXXXXXXXLOCAL TASK RUN END XXXXXXXXXXXXXXXXX") - defer func() { if r := recover(); r != nil { yunionconf.BugReport.SendBugReport(context.Background(), version.GetShortString(), string(debug.Stack()), errors.Errorf("%s", r)) @@ -78,6 +78,18 @@ func LocalTaskRunWithWorkers(task ITask, proc func() (jsonutils.JSONObject, erro wm.Run(&t, nil, nil) } +func getLocalTaskWorkerMan() *appsrv.SWorkerManager { + localTaskWorkerManLock.Lock() + defer localTaskWorkerManLock.Unlock() + + if localTaskWorkerMan != nil { + return localTaskWorkerMan + } + log.Infof("LocalTaskWorkerManager %d", consts.LocalTaskWorkerCount()) + localTaskWorkerMan = appsrv.NewWorkerManager("LocalTaskWorkerManager", consts.LocalTaskWorkerCount(), 1024, false) + return localTaskWorkerMan +} + func LocalTaskRun(task ITask, proc func() (jsonutils.JSONObject, error)) { - LocalTaskRunWithWorkers(task, proc, localTaskWorkerMan) + LocalTaskRunWithWorkers(task, proc, getLocalTaskWorkerMan()) } diff --git a/pkg/cloudcommon/db/taskman/subtasks.go b/pkg/cloudcommon/db/taskman/subtasks.go index 37b091530f..586e70784f 100644 --- a/pkg/cloudcommon/db/taskman/subtasks.go +++ b/pkg/cloudcommon/db/taskman/subtasks.go @@ -19,6 +19,7 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/log" + "yunion.io/x/sqlchemy" "yunion.io/x/onecloud/pkg/cloudcommon/db" ) @@ -36,7 +37,12 @@ type SSubTaskmanager struct { var SubTaskManager *SSubTaskmanager func init() { - SubTaskManager = &SSubTaskmanager{SModelBaseManager: db.NewModelBaseManager(SSubTask{}, "subtasks_tbl", "subtask", "subtasks")} + SubTaskManager = &SSubTaskmanager{SModelBaseManager: db.NewModelBaseManager( + SSubTask{}, + "subtasks_tbl", + "subtask", + "subtasks", + )} } type SSubTask struct { @@ -62,12 +68,17 @@ func (manager *SSubTaskmanager) GetSubTask(ptaskId string, subtaskId string) *SS return &subtask } -func (manager *SSubTaskmanager) GetTotalSubtasks(taskId string, stage string, status string) []SSubTask { - subtasks := make([]SSubTask, 0) +func (manager *SSubTaskmanager) getTotalSubtasksQuery(taskId string, stage string, status string) *sqlchemy.SQuery { q := manager.Query().Equals("task_id", taskId).Equals("stage", stage) if len(status) > 0 { q = q.Equals("status", status) } + return q +} + +func (manager *SSubTaskmanager) GetSubtasks(taskId string, stage string, status string) []SSubTask { + subtasks := make([]SSubTask, 0) + q := manager.getTotalSubtasksQuery(taskId, stage, status) err := db.FetchModelObjects(manager, q, &subtasks) if err != nil { log.Errorf("GetInitSubtasks fail %s", err) @@ -76,8 +87,25 @@ func (manager *SSubTaskmanager) GetTotalSubtasks(taskId string, stage string, st return subtasks } +func (manager *SSubTaskmanager) GetSubtasksCount(taskId string, stage string, status string) (int, error) { + q := manager.getTotalSubtasksQuery(taskId, stage, status) + return q.CountWithError() +} + +func (manager *SSubTaskmanager) GetTotalSubtasks(taskId string, stage string) []SSubTask { + return manager.GetSubtasks(taskId, stage, "") +} + +func (manager *SSubTaskmanager) GetTotalSubtasksCount(taskId string, stage string) (int, error) { + return manager.GetSubtasksCount(taskId, stage, "") +} + func (manager *SSubTaskmanager) GetInitSubtasks(taskId string, stage string) []SSubTask { - return manager.GetTotalSubtasks(taskId, stage, SUBTASK_INIT) + return manager.GetSubtasks(taskId, stage, SUBTASK_INIT) +} + +func (manager *SSubTaskmanager) GetInitSubtasksCount(taskId string, stage string) (int, error) { + return manager.GetSubtasksCount(taskId, stage, SUBTASK_INIT) } func (self *SSubTask) SaveResults(failed bool, result jsonutils.JSONObject) error { diff --git a/pkg/cloudcommon/db/taskman/taskobjs.go b/pkg/cloudcommon/db/taskman/taskobjs.go index 9f9e320706..254f280547 100644 --- a/pkg/cloudcommon/db/taskman/taskobjs.go +++ b/pkg/cloudcommon/db/taskman/taskobjs.go @@ -47,6 +47,7 @@ type STaskObject struct { TaskId string `width:"36" charset:"ascii" nullable:"false" primary:"true" index:"true"` // Column(VARCHAR(36, charset='ascii'), nullable=False, primary_key=True, index=True) ObjId string `width:"36" charset:"ascii" nullable:"false" primary:"true"` // Column(VARCHAR(36, charset='ascii'), nullable=False, primary_key=True) + Object string `json:"object" width:"128" charset:"utf8" nullable:"false" list:"user"` } func (manager *STaskObjectManager) GetObjectIds(task *STask) []string { @@ -73,6 +74,30 @@ func (manager *STaskObjectManager) GetObjectIds(task *STask) []string { return ret } +func (manager *STaskObjectManager) GetObjectNames(task *STask) []string { + ret := make([]string, 0) + taskobjs := manager.Query().SubQuery() + q := taskobjs.Query(taskobjs.Field("object")).Equals("task_id", task.Id) + rows, err := q.Rows() + if err != nil { + if err != sql.ErrNoRows { + log.Errorf("TaskObjectManager GetObjectIds fail %s", err) + } + return nil + } + defer rows.Close() + for rows.Next() { + var objId string + err = rows.Scan(&objId) + if err != nil { + log.Errorf("TaskObjectManager GetObjects fetch row fail %s", err) + return nil + } + ret = append(ret, objId) + } + return ret +} + func (manager *STaskObjectManager) FetchOwnerId(ctx context.Context, data jsonutils.JSONObject) (mcclient.IIdentityProvider, error) { return manager.SProjectizedResourceBaseManager.FetchOwnerId(ctx, data) } diff --git a/pkg/cloudcommon/db/taskman/tasks.go b/pkg/cloudcommon/db/taskman/tasks.go index 0819c9eb5b..15e1b177ce 100644 --- a/pkg/cloudcommon/db/taskman/tasks.go +++ b/pkg/cloudcommon/db/taskman/tasks.go @@ -70,6 +70,9 @@ const ( CONVERT_TASK = "convert_task" LANG = "lang" + + taskStatusDone = "done" + TASK_STATUS_QUEUE = "queue" ) type STaskManager struct { @@ -106,7 +109,8 @@ type STask struct { Id string `width:"36" charset:"ascii" primary:"true" list:"user"` // Column(VARCHAR(36, charset='ascii'), primary_key=True, default=get_uuid) - ObjName string `width:"128" charset:"utf8" nullable:"false" list:"user"` // Column(VARCHAR(128, charset='utf8'), nullable=False) + ObjType string `old_name:"obj_name" json:"obj_type" width:"128" charset:"utf8" nullable:"false" list:"user"` + Object string `json:"object" width:"128" charset:"utf8" nullable:"false" list:"user"` // Column(VARCHAR(128, charset='utf8'), nullable=False) ObjId string `width:"128" charset:"ascii" nullable:"false" list:"user" index:"true"` // Column(VARCHAR(ID_LENGTH, charset='ascii'), nullable=False) TaskName string `width:"64" charset:"ascii" nullable:"false" list:"user"` // Column(VARCHAR(64, charset='ascii'), nullable=False) @@ -121,6 +125,10 @@ type STask struct { taskObject db.IStandaloneModel `ignore:"true"` taskObjects []db.IStandaloneModel `ignore:"true"` + + SubTaskCount int `ignore:"true"` + FailSubTaskCnt int `ignore:"true"` + SUccSubTaskCnt int `ignore:"true"` } func (manager *STaskManager) CreateByInsertOrUpdate() bool { @@ -303,8 +311,9 @@ func (manager *STaskManager) NewTask( data := fetchTaskParams(ctx, taskName, taskData, parentTaskId, parentTaskNotifyUrl, pendingUsage) task := &STask{ - ObjName: obj.Keyword(), + ObjType: obj.Keyword(), ObjId: obj.GetId(), + Object: obj.GetName(), TaskName: taskName, UserCred: userCred, Params: data, @@ -324,6 +333,8 @@ func (manager *STaskManager) NewTask( log.Errorf("Task insert error %s", err) return nil, err } + task.SetProgressAndStatus(0, TASK_STATUS_QUEUE) + parentTask := task.GetParentTask() if parentTask != nil { st := &SSubTask{TaskId: parentTask.Id, Stage: parentTask.Stage, SubtaskId: task.Id} @@ -359,7 +370,8 @@ func (manager *STaskManager) NewParallelTask( data := fetchTaskParams(ctx, taskName, taskData, parentTaskId, parentTaskNotifyUrl, pendingUsage) task := &STask{ - ObjName: objs[0].Keyword(), + ObjType: objs[0].Keyword(), + Object: MULTI_OBJECTS_ID, ObjId: MULTI_OBJECTS_ID, TaskName: taskName, UserCred: userCred, @@ -373,19 +385,25 @@ func (manager *STaskManager) NewParallelTask( log.Errorf("Task insert error %s", err) return nil, err } + task.SetProgressAndStatus(0, TASK_STATUS_QUEUE) + domainIds := stringutils2.NewSortedStrings(nil) + tenantIds := stringutils2.NewSortedStrings(nil) for i := range objs { obj := objs[i] to := STaskObject{ TaskId: task.Id, ObjId: obj.GetId(), + Object: obj.GetName(), } ownerId := obj.GetOwnerId() if ownerId != nil { to.DomainId = ownerId.GetProjectDomainId() to.ProjectId = ownerId.GetProjectId() + domainIds = domainIds.Append(to.DomainId) + tenantIds = tenantIds.Append(to.ProjectId) } - to.SetModelManager(TaskObjectManager, &to) + // to.SetModelManager(TaskObjectManager, &to) to.SetModelManager(TaskObjectManager, &to) err := TaskObjectManager.TableSpec().Insert(ctx, &to) @@ -395,6 +413,12 @@ func (manager *STaskManager) NewParallelTask( } } + db.Update(task, func() error { + task.DomainId = strings.Join(domainIds, ",") + task.ProjectId = strings.Join(tenantIds, ",") + return nil + }) + parentTask := task.GetParentTask() if parentTask != nil { st := SSubTask{TaskId: parentTask.Id, Stage: parentTask.Stage, SubtaskId: task.Id} @@ -524,9 +548,9 @@ func execITask(taskValue reflect.Value, task *STask, odata jsonutils.JSONObject, return } - objManager := db.GetModelManager(task.ObjName) + objManager := db.GetModelManager(task.ObjType) if objManager == nil { - msg := fmt.Sprintf("model %s not found??? ...", task.ObjName) + msg := fmt.Sprintf("model %s %s(%s) not found??? ...", task.ObjType, task.Object, task.ObjId) log.Errorf(msg) task.SetStageFailed(ctx, jsonutils.NewString(msg)) task.SaveRequestContext(&ctxData) @@ -535,7 +559,7 @@ func execITask(taskValue reflect.Value, task *STask, odata jsonutils.JSONObject, // log.Debugf("objManager: %s", objManager) objResManager, ok := objManager.(db.IStandaloneModelManager) if !ok { - msg := fmt.Sprintf("model %s is not a resource??? ...", task.ObjName) + msg := fmt.Sprintf("model %s %s(%s) is not a resource??? ...", task.ObjType, task.Object, task.ObjId) log.Errorf(msg) task.SetStageFailed(ctx, jsonutils.NewString(msg)) task.SaveRequestContext(&ctxData) @@ -551,7 +575,7 @@ func execITask(taskValue reflect.Value, task *STask, odata jsonutils.JSONObject, for i, objId := range objIds { obj, err := objResManager.FetchById(objId) if err != nil { - msg := fmt.Sprintf("fail to find %s object %s", task.ObjName, objId) + msg := fmt.Sprintf("fail to find %s object %s", task.ObjType, objId) log.Errorf(msg) task.SetStageFailed(ctx, jsonutils.NewString(msg)) task.SaveRequestContext(&ctxData) @@ -573,7 +597,7 @@ func execITask(taskValue reflect.Value, task *STask, odata jsonutils.JSONObject, } else { obj, err := objResManager.FetchById(task.ObjId) if err != nil { - msg := fmt.Sprintf("fail to find %s object %s", task.ObjName, task.ObjId) + msg := fmt.Sprintf("fail to find %s object %s", task.ObjType, task.ObjId) log.Errorf(msg) task.SetStageFailed(ctx, jsonutils.NewString(msg)) task.SaveRequestContext(&ctxData) @@ -730,25 +754,35 @@ func (self *STask) SetStage(stageName string, data *jsonutils.JSONDict) error { return err } -func (self *STask) GetObjectIdStr() string { - if self.ObjId == MULTI_OBJECTS_ID { - return strings.Join(TaskObjectManager.GetObjectIds(self), ",") +func (task *STask) GetObjectIdStr() string { + if task.ObjId == MULTI_OBJECTS_ID { + return strings.Join(TaskObjectManager.GetObjectIds(task), ",") } else { - return self.ObjId + return task.ObjId } } -func (self *STask) SetStageComplete(ctx context.Context, data *jsonutils.JSONDict) { - log.Infof("XXX TASK %s complete", self.TaskName) - self.SetStage(TASK_STAGE_COMPLETE, data) +func (task *STask) GetObjectStr() string { + if task.ObjId == MULTI_OBJECTS_ID { + return strings.Join(TaskObjectManager.GetObjectNames(task), ",") + } else { + return task.Object + } +} + +func (task *STask) SetStageComplete(ctx context.Context, data *jsonutils.JSONDict) { + log.Infof("XXX TASK %s complete", task.TaskName) + task.SetStage(TASK_STAGE_COMPLETE, data) + task.SetProgressAndStatus(100, taskStatusDone) if data == nil { data = jsonutils.NewDict() } if data.Size() == 0 { - data.Add(jsonutils.NewString(self.GetObjectIdStr()), "id") - data.Add(jsonutils.NewString(self.ObjName), "name") + data.Add(jsonutils.NewString(task.GetObjectIdStr()), "id") + data.Add(jsonutils.NewString(task.GetObjectStr()), "name") + data.Add(jsonutils.NewString(task.ObjType), "type") } - self.NotifyParentTaskComplete(ctx, data, false) + task.NotifyParentTaskComplete(ctx, data, false) } func (self *STask) SetStageFailed(ctx context.Context, reason jsonutils.JSONObject) { @@ -776,6 +810,7 @@ func (self *STask) SetStageFailed(ctx context.Context, reason jsonutils.JSONObje data := jsonutils.NewDict() data.Add(reason, "__failed_reason") self.SetStage(TASK_STAGE_FAILED, data) + self.SetProgressAndStatus(100, taskStatusDone) self.NotifyParentTaskFailure(ctx, reason) } @@ -835,10 +870,11 @@ func (self *STask) NotifyParentTaskFailure(ctx context.Context, reason jsonutils } func (self *STask) IsCurrentStageComplete() bool { - totalSubtasks := SubTaskManager.GetTotalSubtasks(self.Id, self.Stage, "") - initSubtasks := SubTaskManager.GetInitSubtasks(self.Id, self.Stage) - log.Debugf("Task %s IsCurrentStageComplete totalSubtasks %d initSubtasks %d ", self.String(), len(totalSubtasks), len(initSubtasks)) - if len(totalSubtasks) > 0 && len(initSubtasks) == 0 { + totalSubtasksCnt, _ := SubTaskManager.GetTotalSubtasksCount(self.Id, self.Stage) + initSubtasksCnt, _ := SubTaskManager.GetInitSubtasksCount(self.Id, self.Stage) + log.Debugf("Task %s IsCurrentStageComplete totalSubtasks %d initSubtasks %d ", self.String(), totalSubtasksCnt, initSubtasksCnt) + self.SetProgress(float32(totalSubtasksCnt-initSubtasksCnt) / float32(totalSubtasksCnt)) + if totalSubtasksCnt > 0 && initSubtasksCnt == 0 { return true } else { return false @@ -1098,7 +1134,42 @@ func (manager *STaskManager) ListItemFilter( } } - q.DebugQuery2("taskQuery") + if input.Details != nil && *input.Details { + subSQFunc := func(status string, cntField string) *sqlchemy.SSubQuery { + subQ := SubTaskManager.Query() + if len(status) > 0 { + subQ = subQ.Equals("status", status) + } + subQ = subQ.GroupBy(subQ.Field("task_id")) + subQ = subQ.AppendField(subQ.Field("task_id")) + subQ = subQ.AppendField(sqlchemy.COUNT(cntField)) + return subQ.SubQuery() + } + + { + subSQ := subSQFunc("", "sub_task_count") + q = q.LeftJoin(subSQ, sqlchemy.Equals(subSQ.Field("task_id"), q.Field("id"))) + q = q.AppendField(subSQ.Field("sub_task_count")) + } + + { + failSubSQ := subSQFunc(SUBTASK_FAIL, "fail_sub_task_cnt") + q = q.LeftJoin(failSubSQ, sqlchemy.Equals(failSubSQ.Field("task_id"), q.Field("id"))) + q = q.AppendField(failSubSQ.Field("fail_sub_task_cnt")) + } + + { + succSubSQ := subSQFunc(SUBTASK_SUCC, "succ_sub_task_cnt") + q = q.LeftJoin(succSubSQ, sqlchemy.Equals(succSubSQ.Field("task_id"), q.Field("id"))) + q = q.AppendField(succSubSQ.Field("succ_sub_task_cnt")) + } + + for _, c := range manager.TableSpec().Columns() { + q = q.AppendField(q.Field(c.Name())) + } + } + + // q.DebugQuery2("taskQuery") return q, nil } @@ -1173,8 +1244,8 @@ func (manager *STaskManager) OrderByExtraFields( func (task *STask) SetProgressAndStatus(progress float32, status string) error { _, err := db.Update(task, func() error { - task.SetProgressValue(progress) task.SetStatusValue(status) + task.SetProgressValue(progress) return nil }) if err != nil { @@ -1193,3 +1264,17 @@ func (task *STask) SetProgress(progress float32) error { } return nil } + +func (manager *STaskManager) InitializeData() error { + q := manager.Query().NotIn("stage", []string{TASK_STAGE_FAILED, TASK_STAGE_COMPLETE}) + tasks := make([]STask, 0) + err := db.FetchModelObjects(manager, q, &tasks) + if err != nil { + return errors.Wrap(err, "FetchModelObjects") + } + reason := jsonutils.NewString("service restart") + for i := range tasks { + tasks[i].SetStageFailed(context.Background(), reason) + } + return nil +} diff --git a/pkg/cloudcommon/db/taskman/worker.go b/pkg/cloudcommon/db/taskman/worker.go index b7a6d692b7..d7267d322a 100644 --- a/pkg/cloudcommon/db/taskman/worker.go +++ b/pkg/cloudcommon/db/taskman/worker.go @@ -18,34 +18,45 @@ import ( "context" "fmt" "runtime/debug" + "sync" "yunion.io/x/jsonutils" "yunion.io/x/log" api "yunion.io/x/onecloud/pkg/apis/notify" "yunion.io/x/onecloud/pkg/appsrv" + "yunion.io/x/onecloud/pkg/cloudcommon/consts" "yunion.io/x/onecloud/pkg/cloudcommon/notifyclient" ) -const ( - DEFAULT_WORKER_COUNT = 4 -) - -var taskWorkMan *appsrv.SWorkerManager +var _taskWorkMan *appsrv.SWorkerManager var taskWorkerTable map[string]*appsrv.SWorkerManager +var taskWorkManLock *sync.Mutex func init() { - taskWorkMan = appsrv.NewWorkerManager("TaskWorkerManager", DEFAULT_WORKER_COUNT, 1024, true) taskWorkerTable = make(map[string]*appsrv.SWorkerManager) + taskWorkManLock = &sync.Mutex{} } -func UpdateWorkerCount(workerCount int) error { +func getTaskWorkMan() *appsrv.SWorkerManager { + taskWorkManLock.Lock() + defer taskWorkManLock.Unlock() + + if _taskWorkMan != nil { + return _taskWorkMan + } + log.Infof("TaskWorkerManager %d", consts.TaskWorkerCount()) + _taskWorkMan = appsrv.NewWorkerManager("TaskWorkerManager", consts.TaskWorkerCount(), 1024, true) + return _taskWorkMan +} + +/*func UpdateWorkerCount(workerCount int) error { if workerCount != DEFAULT_WORKER_COUNT { log.Infof("update task work count: %d", workerCount) return taskWorkMan.UpdateWorkerCount(workerCount) } return nil -} +}*/ type taskTask struct { taskId string @@ -65,7 +76,7 @@ func runTask(taskId string, data jsonutils.JSONObject) error { if len(taskName) == 0 { return fmt.Errorf("no such task??? task_id=%s", taskId) } - worker := taskWorkMan + worker := getTaskWorkMan() if workerMan, ok := taskWorkerTable[taskName]; ok { worker = workerMan } diff --git a/pkg/cloudcommon/options/changes.go b/pkg/cloudcommon/options/changes.go index 8b0cfd4a1c..547f719446 100644 --- a/pkg/cloudcommon/options/changes.go +++ b/pkg/cloudcommon/options/changes.go @@ -62,6 +62,12 @@ func OnBaseOptionsChange(oOpts, nOpts interface{}) bool { if oldOpts.ApiServer != newOpts.ApiServer { log.Debugf("api_server changed from %s to %s", oldOpts.ApiServer, newOpts.ApiServer) } + if oldOpts.TaskWorkerCount != newOpts.TaskWorkerCount { + consts.SetTaskWorkerCount(newOpts.TaskWorkerCount) + } + if oldOpts.LocalTaskWorkerCount != newOpts.LocalTaskWorkerCount { + consts.SetLocalTaskWorkerCount(newOpts.LocalTaskWorkerCount) + } return changed } diff --git a/pkg/cloudcommon/options/options.go b/pkg/cloudcommon/options/options.go index 1249ba0ec2..aa1d7d94ee 100644 --- a/pkg/cloudcommon/options/options.go +++ b/pkg/cloudcommon/options/options.go @@ -70,7 +70,9 @@ type BaseOptions struct { ApplicationID string `help:"Application ID"` RequestWorkerCount int `default:"8" help:"Request worker thread count, default is 8"` - TaskWorkerCount int `default:"4" help:"Task manager worker thread count, default is 4"` + + TaskWorkerCount int `default:"4" help:"Task manager worker thread count, default is 4"` + LocalTaskWorkerCount int `default:"4" help:"Worker thread count that runs local tasks, default is 4"` DefaultProcessTimeoutSeconds int `default:"60" help:"request process timeout, default is 60 seconds"` @@ -389,6 +391,9 @@ func parseOptions(optStruct interface{}, args []string, configFileName string, s } consts.SetDomainizedNamespace(optionsRef.DomainizedNamespace) + + consts.SetTaskWorkerCount(optionsRef.TaskWorkerCount) + consts.SetLocalTaskWorkerCount(optionsRef.LocalTaskWorkerCount) } func (self *BaseOptions) HttpTransportProxyFunc() httputils.TransportProxyFunc { diff --git a/pkg/compute/service/service.go b/pkg/compute/service/service.go index 7358112cb0..739aa1871a 100644 --- a/pkg/compute/service/service.go +++ b/pkg/compute/service/service.go @@ -27,6 +27,7 @@ import ( "yunion.io/x/pkg/errors" _ "yunion.io/x/sqlchemy/backends" + "yunion.io/x/onecloud/pkg/apis" api "yunion.io/x/onecloud/pkg/apis/compute" "yunion.io/x/onecloud/pkg/apis/identity" "yunion.io/x/onecloud/pkg/cloudcommon" @@ -77,16 +78,16 @@ func StartServiceWithJobs(jobs func(cron *cronman.SCronJobManager)) { }) common_options.StartOptionManager(opts, opts.ConfigSyncPeriodSeconds, api.SERVICE_TYPE, api.SERVICE_VERSION, options.OnOptionsChange) - serviceUrl, err := auth.GetServiceURL(api.SERVER_TYPE_V2, opts.Region, "", identity.EndpointInterfaceInternal) + serviceUrl, err := auth.GetServiceURL(apis.SERVICE_TYPE_REGION, opts.Region, "", identity.EndpointInterfaceInternal) if err != nil { log.Fatalf("unable to get service url: %v", err) } log.Infof("serviceUrl: %s", serviceUrl) taskman.SetServiceUrl(serviceUrl) - err = taskman.UpdateWorkerCount(opts.TaskWorkerCount) - if err != nil { - log.Fatalf("failed update task manager worker count %s", err) - } + // err = taskman.UpdateWorkerCount(opts.TaskWorkerCount) + // if err != nil { + // log.Fatalf("failed update task manager worker count %s", err) + // } err = esxi.InitEsxiConfig(opts.EsxiOptions) if err != nil { diff --git a/pkg/compute/tasks/guest_save_instance_image_task.go b/pkg/compute/tasks/guest_save_instance_image_task.go index 8c322a51aa..f77dfa40ee 100644 --- a/pkg/compute/tasks/guest_save_instance_image_task.go +++ b/pkg/compute/tasks/guest_save_instance_image_task.go @@ -63,11 +63,14 @@ func (self *GuestSaveGuestImageTask) OnInit(ctx context.Context, obj db.IStandal } func (self *GuestSaveGuestImageTask) OnSaveRootImageComplete(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { - subTasks := taskman.SubTaskManager.GetTotalSubtasks(self.Id, "on_save_root_image_complete", taskman.SUBTASK_FAIL) - - if len(subTasks) > 0 { + subTasksCnt, err := taskman.SubTaskManager.GetSubtasksCount(self.Id, "on_save_root_image_complete", taskman.SUBTASK_FAIL) + if err != nil { + self.taskFailed(ctx, guest, jsonutils.NewString(err.Error())) + return + } else if subTasksCnt > 0 { self.taskFailed(ctx, guest, jsonutils.NewString("subtask failed")) // ??? return ??? + return } if restart, _ := self.GetParams().Bool("auto_start"); restart { diff --git a/pkg/compute/tasks/instance_backup_create_task.go b/pkg/compute/tasks/instance_backup_create_task.go index d842c58dc8..200f389fbd 100644 --- a/pkg/compute/tasks/instance_backup_create_task.go +++ b/pkg/compute/tasks/instance_backup_create_task.go @@ -69,7 +69,7 @@ func (self *InstanceBackupCreateTask) OnInit(ctx context.Context, obj db.IStanda } func (self *InstanceBackupCreateTask) OnKvmDisksSnapshot(ctx context.Context, ib *models.SInstanceBackup, data jsonutils.JSONObject) { - subTasks := taskman.SubTaskManager.GetTotalSubtasks(self.Id, "OnKvmDisksSnapshot", "") + subTasks := taskman.SubTaskManager.GetSubtasks(self.Id, "OnKvmDisksSnapshot", "") guest := models.GuestManager.FetchGuestById(ib.GuestId) self.SetStage("OnInstanceBackup", nil) for i := range subTasks { @@ -107,7 +107,7 @@ func (self *InstanceBackupCreateTask) OnKvmDisksSnapshotFailed(ctx context.Conte } func (self *InstanceBackupCreateTask) OnInstanceBackup(ctx context.Context, ib *models.SInstanceBackup, data jsonutils.JSONObject) { - subTasks := taskman.SubTaskManager.GetTotalSubtasks(self.Id, "OnInstanceBackup", "") + subTasks := taskman.SubTaskManager.GetSubtasks(self.Id, "OnInstanceBackup", "") for i := range subTasks { if subTasks[i].Status == taskman.SUBTASK_SUCC { continue diff --git a/pkg/compute/tasks/instance_backup_syncstatus.go b/pkg/compute/tasks/instance_backup_syncstatus.go index 888b201df9..d485880908 100644 --- a/pkg/compute/tasks/instance_backup_syncstatus.go +++ b/pkg/compute/tasks/instance_backup_syncstatus.go @@ -54,7 +54,7 @@ func (self *InstanceBackupSyncstatusTask) OnInit(ctx context.Context, obj db.ISt } func (self *InstanceBackupSyncstatusTask) OnKvmBackupSyncstatus(ctx context.Context, ib *models.SInstanceBackup, data jsonutils.JSONObject) { - subTasks := taskman.SubTaskManager.GetTotalSubtasks(self.Id, "OnKvmDisksSnapshot", "") + subTasks := taskman.SubTaskManager.GetSubtasks(self.Id, "OnKvmDisksSnapshot", "") for i := range subTasks { log.Infof("subsTask %s result: %s", subTasks[i].SubtaskId, subTasks[i].Result) result, err := jsonutils.ParseString(subTasks[i].Result) diff --git a/pkg/image/drivers/s3/minio.go b/pkg/image/drivers/s3/minio.go index b7821835a5..462a9b0833 100644 --- a/pkg/image/drivers/s3/minio.go +++ b/pkg/image/drivers/s3/minio.go @@ -21,6 +21,7 @@ import ( "os" "yunion.io/x/cloudmux/pkg/cloudprovider" + "yunion.io/x/cloudmux/pkg/multicloud" "yunion.io/x/cloudmux/pkg/multicloud/objectstore" "yunion.io/x/log" "yunion.io/x/pkg/errors" @@ -87,7 +88,7 @@ func ensureBucket() error { return nil } -func PutStream(ctx context.Context, file io.Reader, fSize int64, objName string) (string, error) { +func PutStream(ctx context.Context, file io.Reader, fSize int64, objName string, progresser func(saved int64)) (string, error) { if client == nil { return "", ErrClientNotInit } @@ -96,7 +97,10 @@ func PutStream(ctx context.Context, file io.Reader, fSize int64, objName string) return "", errors.Wrap(err, "client.getBucket") } const blockSizeMB = 100 - err = cloudprovider.UploadObject(ctx, bucket, objName, blockSizeMB*1000*1000, file, fSize, cloudprovider.ACLPrivate, "", nil, false) + pFile := multicloud.NewProgress(fSize, 100, file, func(ratio float32) { + progresser(int64(float64(ratio) * float64(fSize))) + }) + err = cloudprovider.UploadObject(ctx, bucket, objName, blockSizeMB*1000*1000, pFile, fSize, cloudprovider.ACLPrivate, "", nil, false) if err != nil { return "", errors.Wrap(err, "cloudprovider.UploadObject") } @@ -104,7 +108,7 @@ func PutStream(ctx context.Context, file io.Reader, fSize int64, objName string) return client.Location(objName), nil } -func Put(ctx context.Context, filePath, objName string) (string, error) { +func Put(ctx context.Context, filePath, objName string, progresser func(int64)) (string, error) { finfo, err := os.Stat(filePath) if err != nil { return "", errors.Wrap(err, "os.Stat") @@ -115,7 +119,7 @@ func Put(ctx context.Context, filePath, objName string) (string, error) { return "", errors.Wrap(err, "os.Open") } defer file.Close() - return PutStream(ctx, file, fSize, objName) + return PutStream(ctx, file, fSize, objName, progresser) } func Get(ctx context.Context, fileName string) (int64, io.ReadCloser, error) { diff --git a/pkg/image/models/image_subs.go b/pkg/image/models/image_subs.go index 108af5ae66..c19e1be1a6 100644 --- a/pkg/image/models/image_subs.go +++ b/pkg/image/models/image_subs.go @@ -152,7 +152,7 @@ func (self *SImageSubformat) Save(image *SImage) error { log.Errorf("updateStatus fail %s", err) return err } - info, err := storage.ConvertImage(context.Background(), image, self.Format) + info, err := storage.ConvertImage(context.Background(), image, self.Format, nil) if err != nil { return errors.Wrap(err, "unable to ConvertImage") } diff --git a/pkg/image/models/images.go b/pkg/image/models/images.go index a708a202af..fe9548e2a5 100644 --- a/pkg/image/models/images.go +++ b/pkg/image/models/images.go @@ -1827,7 +1827,7 @@ func (image *SImage) doUploadPermanentStorage(ctx context.Context, userCred mccl imagePath := image.GetLocalLocation() image.SetStatus(ctx, userCred, api.IMAGE_STATUS_SAVING, "save image to specific storage") storage := GetStorage() - location, err := storage.SaveImage(ctx, imagePath) + location, err := storage.SaveImage(ctx, imagePath, nil) if err != nil { log.Errorf("Failed save image to specific storage %s", err) errStr := fmt.Sprintf("save image to storage %s: %v", storage.Type(), err) @@ -1870,7 +1870,7 @@ func (image *SImage) doUploadPermanentStorage(ctx context.Context, userCred mccl } else { imagePath := subimgs[i].GetLocalLocation() storage := GetStorage() - location, err := GetStorage().SaveImage(ctx, imagePath) + location, err := GetStorage().SaveImage(ctx, imagePath, nil) if err != nil { log.Errorf("Failed save image to sepcific storage %s", err) subimgs[i].SetStatus(api.IMAGE_STATUS_SAVE_FAIL) diff --git a/pkg/image/models/storage.go b/pkg/image/models/storage.go index fbd89bfa3f..9543c157ff 100644 --- a/pkg/image/models/storage.go +++ b/pkg/image/models/storage.go @@ -31,11 +31,11 @@ import ( "yunion.io/x/onecloud/pkg/util/procutils" ) -var local Storage = &LocalStorage{} -var s3Instance Storage = &S3Storage{} -var storage Storage +var local IImageStorage = &LocalStorage{} +var s3Instance IImageStorage = &S3Storage{} +var storage IImageStorage -func GetStorage() Storage { +func GetStorage() IImageStorage { return storage } @@ -83,15 +83,15 @@ func Init(storageBackend string) { } } -type Storage interface { +type IImageStorage interface { Type() string - SaveImage(context.Context, string) (string, error) + SaveImage(context.Context, string, func(int64)) (string, error) CleanTempfile(string) error GetImage(context.Context, string) (int64, io.ReadCloser, error) RemoveImage(context.Context, string) error IsCheckStatusEnabled() bool - ConvertImage(ctx context.Context, image *SImage, targetFormat string) (*SConverImageInfo, error) + ConvertImage(ctx context.Context, image *SImage, targetFormat string, progresser func(saved int64)) (*SConverImageInfo, error) } type LocalStorage struct{} @@ -100,7 +100,7 @@ func (s *LocalStorage) Type() string { return image.IMAGE_STORAGE_DRIVER_LOCAL } -func (s *LocalStorage) SaveImage(ctx context.Context, imagePath string) (string, error) { +func (s *LocalStorage) SaveImage(ctx context.Context, imagePath string, progresser func(saved int64)) (string, error) { return fmt.Sprintf("%s%s", LocalFilePrefix, imagePath), nil } @@ -120,7 +120,7 @@ func (s *LocalStorage) GetImage(ctx context.Context, imagePath string) (int64, i return fstat.Size(), f, nil } -func (s *LocalStorage) ConvertImage(ctx context.Context, image *SImage, targetFormat string) (*SConverImageInfo, error) { +func (s *LocalStorage) ConvertImage(ctx context.Context, image *SImage, targetFormat string, progresser func(saved int64)) (*SConverImageInfo, error) { location := image.GetPath(targetFormat) img, err := image.getQemuImage() if err != nil { @@ -155,11 +155,11 @@ func (s *S3Storage) Type() string { return image.IMAGE_STORAGE_DRIVER_S3 } -func (s *S3Storage) SaveImage(ctx context.Context, imagePath string) (string, error) { +func (s *S3Storage) SaveImage(ctx context.Context, imagePath string, progresser func(saved int64)) (string, error) { if !fileutils2.IsFile(imagePath) { return "", fmt.Errorf("%s not valid file", imagePath) } - return s3.Put(ctx, imagePath, imagePathToName(imagePath)) + return s3.Put(ctx, imagePath, imagePathToName(imagePath), progresser) } func (s *S3Storage) CleanTempfile(filePath string) error { @@ -191,7 +191,7 @@ type SConverImageInfo struct { SizeBytes int64 } -func (s *S3Storage) ConvertImage(ctx context.Context, image *SImage, targetFormat string) (*SConverImageInfo, error) { +func (s *S3Storage) ConvertImage(ctx context.Context, image *SImage, targetFormat string, progresser func(saved int64)) (*SConverImageInfo, error) { tempDir, err := s.getTempDir() if err != nil { return nil, err @@ -206,7 +206,7 @@ func (s *S3Storage) ConvertImage(ctx context.Context, image *SImage, targetForma return nil, errors.Wrap(err, "unable to img.Clone") } defer s.CleanTempfile(location) - s3Location, err := s.SaveImage(ctx, location) + s3Location, err := s.SaveImage(ctx, location, progresser) if err != nil { return nil, errors.Wrap(err, "unable to SaveImage") } diff --git a/pkg/mcclient/modulebase/modules.go b/pkg/mcclient/modulebase/modules.go index 825b12aa69..eb83b8cbf1 100644 --- a/pkg/mcclient/modulebase/modules.go +++ b/pkg/mcclient/modulebase/modules.go @@ -18,6 +18,7 @@ import ( "fmt" "sort" "strings" + "sync" "yunion.io/x/jsonutils" "yunion.io/x/log" @@ -178,57 +179,107 @@ var ( jointModules map[string][]JointManager ) -func _getJointKey(mod1 Manager, mod2 Manager) string { +func resourceKey(mod IBaseManager) string { + return fmt.Sprintf("%s-%s", mod.ServiceType(), mod.KeyString()) +} + +func resourceKey2(mod IBaseManager) string { + return mod.KeyString() +} + +func jointResourceKey(mod1, mod2 IBaseManager) string { + return fmt.Sprintf("%s-%s-%s", mod1.ServiceType(), mod1.KeyString(), mod2.KeyString()) +} + +func jointResourceKey2(mod1, mod2 IBaseManager) string { return fmt.Sprintf("%s-%s", mod1.KeyString(), mod2.KeyString()) } -func ensureModuleNotRegistered(mod, newMod IBaseManager) { +func ensureModuleNotRegistered(mod, newMod IBaseManager, isFatal bool) bool { modSvcType := mod.ServiceType() newModSvcType := newMod.ServiceType() if mod == newMod { - log.Fatalf("Module %#v duplicate registered, service type: %q", mod, modSvcType) + if isFatal { + log.Fatalf("Module %#v duplicate registered, service type: %q", mod, modSvcType) + } else { + return false + } } if modSvcType != newModSvcType { - log.Fatalf("Module %#v already registered, service type is %q.\nSo new module %#v can't be registered, service type is %q", mod, modSvcType, newMod, newModSvcType) + if isFatal { + log.Fatalf("Module %#v already registered, service type is %q.\nSo new module %#v can't be registered, service type is %q", mod, modSvcType, newMod, newModSvcType) + } else { + return false + } } + return true } -func Register(mod IBaseManager) { +func Register(m IBaseManager) { if modules == nil { modules = make(map[string][]IBaseManager) } - mods, ok := modules[mod.KeyString()] - if !ok { - mods = make([]IBaseManager, 0) + for i, key := range []string{ + resourceKey(m), + resourceKey2(m), + } { + fatal := false + if i == 0 { + fatal = true + } + mods, ok := modules[key] + if !ok { + mods = make([]IBaseManager, 0) + } + skip := false + for i := range mods { + if !ensureModuleNotRegistered(mods[i], m, fatal) { + skip = true + break + } + } + if !skip { + mods = append(mods, m) + modules[key] = mods + } } - for i := range mods { - ensureModuleNotRegistered(mods[i], mod) - } - mods = append(mods, mod) - modules[mod.KeyString()] = mods - // modtable[mod.KeyString()] = append(mods, mod) } func RegisterJointModule(mod IBaseManager) { jointMod, ok := mod.(JointManager) if ok { // also a joint manager - jointKey := _getJointKey(jointMod.MasterManager(), jointMod.SlaveManager()) - // log.Printf("%s(%s) is also a joint module", mod.KeyString(), jointKey) - jointMods, ok := jointModules[jointKey] - if !ok { - jointMods = make([]JointManager, 0) + for _, jointKey := range []string{ + jointResourceKey(jointMod.MasterManager(), jointMod.SlaveManager()), + jointResourceKey2(jointMod.MasterManager(), jointMod.SlaveManager()), + } { + jointMods, ok := jointModules[jointKey] + if !ok { + jointMods = make([]JointManager, 0) + } + skip := false + for i := range jointMods { + if !ensureModuleNotRegistered(jointMods[i], jointMod, false) { + skip = true + break + } + } + if !skip { + jointModules[jointKey] = append(jointMods, jointMod) + } } - for i := range jointMods { - // if m == jointMod { - ensureModuleNotRegistered(jointMods[i], jointMod) - //} - } - // modtable[jointKey] = append(jointMods, jointMod) - jointModules[jointKey] = append(jointMods, jointMod) } } +var jointModulesLock *sync.Mutex + +func init() { + jointModulesLock = &sync.Mutex{} +} + func registerAllJointModules() { + jointModulesLock.Lock() + defer jointModulesLock.Unlock() + if jointModules == nil { jointModules = make(map[string][]JointManager) for modname := range modules { @@ -291,22 +342,28 @@ func GetJointModule(session *mcclient.ClientSession, name string) (JointManager, func GetJointModule2(session *mcclient.ClientSession, mod1 Manager, mod2 Manager) (JointManager, error) { registerAllJointModules() - key := _getJointKey(mod1, mod2) - mods, ok := jointModules[key] - if !ok { - return nil, fmt.Errorf("No such joint module: %s", key) - } - for _, mod := range mods { - url, e := session.GetServiceVersionURL(mod.ServiceType(), mod.EndpointType()) - if e != nil { - return nil, e + + for _, key := range []string{ + jointResourceKey(mod1, mod2), + jointResourceKey2(mod1, mod2), + } { + mods, ok := jointModules[key] + if !ok { + continue } - _, ver := mcclient.SplitVersionedURL(url) - if strings.EqualFold(ver, mod.Version()) { - return mod, nil + for _, mod := range mods { + url, e := session.GetServiceVersionURL(mod.ServiceType(), mod.EndpointType()) + if e != nil { + return nil, e + } + _, ver := mcclient.SplitVersionedURL(url) + if strings.EqualFold(ver, mod.Version()) { + return mod, nil + } } + return nil, fmt.Errorf("Version mismatch") } - return nil, fmt.Errorf("Version mismatch") + return nil, fmt.Errorf("No such joint module: %s", jointResourceKey(mod1, mod2)) } func GetRegisterdModules() ([]string, []string) {