opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-24 16:03:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: task improvements (#20379)

Browse Source 
Co-authored-by: Qiu Jian <qiujian@yunionyun.com>
This commit is contained in:
Jian Qiu
2024-05-29 14:41:21 +08:00
committed by GitHub
co-authored by Qiu Jian
parent d8c4a28b77
commit 8abda21433
18 changed files with 394 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/appsrv/workers.go
+24
View File
@@ -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
pkg/cloudcommon/consts/db.go
+19
View File
@@ -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
}
pkg/cloudcommon/db/taskman/localtaskworker.go
+17 -5
View File
@@ -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())
}
pkg/cloudcommon/db/taskman/subtasks.go
+32 -4
View File
@@ -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 {
pkg/cloudcommon/db/taskman/taskobjs.go
+25
View File
@@ -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)
}
pkg/cloudcommon/db/taskman/tasks.go
+110 -25
View File
@@ -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
}
pkg/cloudcommon/db/taskman/worker.go
+20 -9
View File
@@ -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
}
pkg/cloudcommon/options/changes.go
+6
View File
@@ -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
}
pkg/cloudcommon/options/options.go
+6 -1
View File
@@ -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 {
pkg/compute/service/service.go
+6 -5
View File
@@ -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 {
pkg/compute/tasks/guest_save_instance_image_task.go
+6 -3
View File
@@ -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 {
pkg/compute/tasks/instance_backup_create_task.go
+2 -2
View File
@@ -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
pkg/compute/tasks/instance_backup_syncstatus.go
+1 -1
View File
@@ -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)
pkg/image/drivers/s3/minio.go
+8 -4
View File
@@ -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) {
pkg/image/models/image_subs.go
+1 -1
View File
@@ -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")
}
pkg/image/models/images.go
+2 -2
View File
@@ -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)
pkg/image/models/storage.go
+13 -13
View File
@@ -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")
}
pkg/mcclient/modulebase/modules.go
+96 -39
View File
@@ -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) {