From 7b845d17c730c0043e4a5388db75289b2639a9ee Mon Sep 17 00:00:00 2001 From: Qu Xuan Date: Mon, 18 Jan 2021 17:21:19 +0800 Subject: [PATCH] fix(common): dump task info when stucking --- pkg/appsrv/appsrv.go | 122 ++++++++++------ pkg/appsrv/workers.go | 36 ++++- pkg/appsrv/workers_test.go | 77 +++++++--- pkg/baremetal/handler/delay.go | 36 +++-- pkg/baremetal/tasks/worker.go | 21 ++- pkg/cloudcommon/cronman/cronman.go | 13 +- pkg/cloudcommon/db/quotas/usageworker.go | 80 ++++++---- pkg/cloudcommon/db/taskman/localtaskworker.go | 47 +++--- pkg/cloudcommon/db/taskman/worker.go | 23 ++- pkg/cloudcommon/informer/worker.go | 30 +++- pkg/cloudcommon/notifyclient/notify.go | 36 +++-- .../notifyclient/notify_internal.go | 99 +++++++------ pkg/cloudcommon/policy/policy.go | 62 +++++--- pkg/cloudcommon/syncman/sync.go | 21 ++- pkg/cloudcommon/workmanager/manager.go | 138 +++++++++++------- pkg/cloudid/models/cloudgroup.go | 3 + pkg/cloudid/models/clouduser.go | 2 + pkg/compute/models/snapshotpolicycache.go | 53 ++++--- pkg/compute/models/syncworkers.go | 25 +++- pkg/hostman/guestman/qemu-kvm.go | 37 +++-- pkg/image/torrent/torrent.go | 31 +++- pkg/keystone/models/syncworker.go | 82 +++++++---- pkg/mcclient/catalog.go | 44 +++++- pkg/monitor/service/dispatcher.go | 26 +++- pkg/util/logclient/logclient.go | 33 ++++- 25 files changed, 809 insertions(+), 368 deletions(-) diff --git a/pkg/appsrv/appsrv.go b/pkg/appsrv/appsrv.go index eed62459b7..a9bd0afdf1 100644 --- a/pkg/appsrv/appsrv.go +++ b/pkg/appsrv/appsrv.go @@ -252,6 +252,49 @@ func (app *Application) handleCORS(w http.ResponseWriter, r *http.Request) bool } } +type appTask struct { + ctx context.Context + hand *SHandlerInfo + rid string + params map[string]string + appParams *SAppParams + app *Application + fw responseWriterChannel + r *http.Request + segs []string + to time.Duration + cancel context.CancelFunc +} + +func (t *appTask) Run() { + if t.ctx.Err() == nil { + t.ctx = context.WithValue(t.ctx, appctx.APP_CONTEXT_KEY_REQUEST_ID, t.rid) + t.ctx = context.WithValue(t.ctx, appctx.APP_CONTEXT_KEY_CUR_ROOT, t.hand.path) + t.ctx = context.WithValue(t.ctx, appctx.APP_CONTEXT_KEY_CUR_PATH, t.segs[len(t.hand.path):]) + t.ctx = context.WithValue(t.ctx, appctx.APP_CONTEXT_KEY_PARAMS, t.params) + t.ctx = context.WithValue(t.ctx, appctx.APP_CONTEXT_KEY_START_TIME, time.Now().UTC()) + if t.hand.metadata != nil { + t.ctx = context.WithValue(t.ctx, appctx.APP_CONTEXT_KEY_METADATA, t.hand.metadata) + } + t.ctx = context.WithValue(t.ctx, APP_CONTEXT_KEY_APP_PARAMS, t.appParams) + func() { + span := trace.StartServerTrace(&t.fw, t.r, t.appParams.Name, t.app.GetName(), t.hand.GetTags()) + defer func() { + if !t.appParams.SkipTrace { + span.EndTrace() + } + }() + t.ctx = context.WithValue(t.ctx, appctx.APP_CONTEXT_KEY_TRACE, span) + t.hand.handler(t.ctx, &t.fw, t.r) + }() + } // otherwise, the task has been timeout + t.fw.closeChannels() +} + +func (t *appTask) Dump() string { + return fmt.Sprintf("%s %s", t.r.Method, t.r.URL.String()) +} + func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, rid string) (*SHandlerInfo, *SAppParams) { segs := SplitPath(r.URL.EscapedPath()) for i := range segs { @@ -268,24 +311,30 @@ func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, ri // log.Print("Found handler", params) hand, ok := handler.(*SHandlerInfo) if ok { - fw := newResponseWriterChannel(w) - currentWorker := make(chan *SWorker, 1) // make it a buffered channel - to := hand.FetchProcessTimeout(r) - if to == 0 { - to = app.processTimeout + task := &appTask{ + ctx: app.context, + hand: hand, + rid: rid, + params: params, + app: app, + fw: newResponseWriterChannel(w), + r: r, + segs: segs, + to: hand.FetchProcessTimeout(r), + cancel: nil, } - var ( - ctx = app.context - cancel context.CancelFunc = nil - ) - if to > 0 { - ctx, cancel = context.WithTimeout(app.context, to) + currentWorker := make(chan *SWorker, 1) // make it a buffered channel + if task.to == 0 { + task.to = app.processTimeout } - if cancel != nil { - defer cancel() + if task.to > 0 { + task.ctx, task.cancel = context.WithTimeout(task.ctx, task.to) } - ctx = i18n.WithRequestLang(ctx, r) + if task.cancel != nil { + defer task.cancel() + } + task.ctx = i18n.WithRequestLang(task.ctx, r) session := hand.workerMan if session == nil { if r.Method == "GET" || r.Method == "HEAD" { @@ -294,52 +343,29 @@ func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, ri session = app.session } } - appParams := hand.GetAppParams(params, segs) - appParams.Request = r - appParams.Response = w + task.appParams = hand.GetAppParams(params, segs) + task.appParams.Request = r + task.appParams.Response = w session.Run( - func() { - if ctx.Err() == nil { - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_REQUEST_ID, rid) - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_CUR_ROOT, hand.path) - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_CUR_PATH, segs[len(hand.path):]) - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_PARAMS, params) - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_START_TIME, time.Now().UTC()) - if hand.metadata != nil { - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_METADATA, hand.metadata) - } - ctx = context.WithValue(ctx, APP_CONTEXT_KEY_APP_PARAMS, appParams) - func() { - span := trace.StartServerTrace(&fw, r, appParams.Name, app.GetName(), hand.GetTags()) - defer func() { - if !appParams.SkipTrace { - span.EndTrace() - } - }() - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_TRACE, span) - hand.handler(ctx, &fw, r) - }() - } // otherwise, the task has been timeout - fw.closeChannels() - }, + task, currentWorker, func(err error) { - httperrors.InternalServerError(ctx, &fw, "Internal server error: %s", err) - fw.closeChannels() + httperrors.InternalServerError(task.ctx, &task.fw, "Internal server error: %s", err) + task.fw.closeChannels() }, ) - runErr := fw.wait(ctx, currentWorker) + runErr := task.fw.wait(task.ctx, currentWorker) if runErr != nil { switch runErr.(type) { case *httputils.JSONClientError: je := runErr.(*httputils.JSONClientError) - httperrors.GeneralServerError(ctx, w, je) + httperrors.GeneralServerError(task.ctx, w, je) default: - httperrors.InternalServerError(ctx, w, "Internal server error") + httperrors.InternalServerError(task.ctx, w, "Internal server error") } } - fw.closeChannels() - return hand, appParams + task.fw.closeChannels() + return hand, task.appParams } else { ctx := i18n.WithRequestLang(context.TODO(), r) httperrors.InternalServerError(ctx, w, "Invalid handler %s", r.URL) diff --git a/pkg/appsrv/workers.go b/pkg/appsrv/workers.go index a6defdcb7d..b8cef02f3e 100644 --- a/pkg/appsrv/workers.go +++ b/pkg/appsrv/workers.go @@ -21,6 +21,7 @@ import ( "net/http" "runtime/debug" "sync" + "time" "yunion.io/x/jsonutils" "yunion.io/x/log" @@ -53,6 +54,7 @@ type SWorker struct { id uint64 state int container *list.Element + task *sWorkerTask manager *SWorkerManager } @@ -84,6 +86,8 @@ func (worker *SWorker) run() { if task.worker != nil { task.worker <- worker } + task.start = time.Now() + worker.task = task execCallback(task) } else { break @@ -99,7 +103,7 @@ func (worker *SWorker) Detach(reason string) { worker.manager.activeWorker.removeWithLock(worker) worker.manager.detachedWorker.addWithLock(worker) - log.Warningf("detach worker %s due to reason %s", worker, reason) + log.Warningf("detach worker %s(%s) due to reason %s after %s", worker, worker.task.task.Dump(), reason, time.Now().Sub(worker.task.start)) worker.manager.scheduleWithLock() } @@ -113,7 +117,11 @@ func (worker *SWorker) StateStr() string { } func (worker *SWorker) String() string { - return fmt.Sprintf("#%d(%p, %s)", worker.id, worker, worker.StateStr()) + workerInfo := "" + if worker.task != nil { + workerInfo = worker.task.task.Dump() + } + return fmt.Sprintf("#%d(%p, %s) %s", worker.id, worker, worker.StateStr(), workerInfo) } type SWorkerList struct { @@ -179,17 +187,23 @@ func NewWorkerManagerIgnoreOverflow(name string, workerCount int, backlog int, d return &manager } +type IWorkerTask interface { + Run() + Dump() string +} + type sWorkerTask struct { - task func() + task IWorkerTask worker chan *SWorker onError func(error) + start time.Time } func (wm *SWorkerManager) String() string { return wm.name } -func (wm *SWorkerManager) Run(task func(), worker chan *SWorker, onErr func(error)) bool { +func (wm *SWorkerManager) Run(task IWorkerTask, worker chan *SWorker, onErr func(error)) bool { ret := wm.queue.Push(&sWorkerTask{task: task, worker: worker, onError: onErr}) if ret { wm.schedule() @@ -220,7 +234,7 @@ func execCallback(task *sWorkerTask) { debug.PrintStack() } }() - task.task() + task.task.Run() } func (wm *SWorkerManager) schedule() { @@ -231,7 +245,8 @@ func (wm *SWorkerManager) schedule() { } func (wm *SWorkerManager) scheduleWithLock() { - if wm.activeWorker.size() < wm.workerCount && wm.queue.Size() > 0 { + queueSize := wm.queue.Size() + if wm.activeWorker.size() < wm.workerCount && queueSize > 0 { wm.workerId += 1 worker := newWorker(wm.workerId, wm) wm.activeWorker.addWithLock(worker) @@ -239,8 +254,15 @@ func (wm *SWorkerManager) scheduleWithLock() { log.Debugf("no enough worker, add new worker %s", worker) } go worker.run() - } else if wm.queue.Size() > 10 { + } else if queueSize > 10 { log.Warningf("[%s] BUSY activeWork %d detachedWork %d max %d queue: %d", wm, wm.ActiveWorkerCount(), wm.DetachedWorkerCount(), wm.workerCount, wm.queue.Size()) + } else if queueSize > 50 { + w := wm.activeWorker.list.Front() + for w != nil { + worker := w.Value.(*SWorker) + log.Warningf("work [%s]%s stucking for a while", worker.task.start, worker.task.task.Dump()) + w = w.Next() + } } } diff --git a/pkg/appsrv/workers_test.go b/pkg/appsrv/workers_test.go index d007b6b62e..6164e826e5 100644 --- a/pkg/appsrv/workers_test.go +++ b/pkg/appsrv/workers_test.go @@ -15,22 +15,35 @@ package appsrv import ( + "fmt" "sync" "testing" "time" ) +type workerTask struct { + counter int +} + +func (t *workerTask) Run() { + t.counter += 1 + time.Sleep(1 * time.Second) +} + +func (t *workerTask) Dump() string { + return fmt.Sprintf("counter: %d", t.counter) +} + func TestWorkerManager(t *testing.T) { enableDebug() startTime := time.Now() // end := make(chan int) wm := NewWorkerManager("testwm", 2, 10, false) - counter := 0 + task := &workerTask{ + counter: 0, + } for i := 0; i < 10; i += 1 { - wm.Run(func() { - counter += 1 - time.Sleep(1 * time.Second) - }, nil, nil) + wm.Run(task, nil, nil) } for wm.ActiveWorkerCount() != 0 { time.Sleep(time.Second) @@ -40,6 +53,31 @@ func TestWorkerManager(t *testing.T) { } } +type errWorkerTask struct { + wg *sync.WaitGroup +} + +func (t *errWorkerTask) Run() { + t.wg.Done() +} + +func (t *errWorkerTask) Dump() string { + return "" +} + +type panicWorkerTask struct { + wg *sync.WaitGroup +} + +func (t *panicWorkerTask) Run() { + defer t.wg.Done() + panic("panic inside worker") +} + +func (t *panicWorkerTask) Dump() string { + return "" +} + func TestWorkerManagerError(t *testing.T) { wm := NewWorkerManager("testwm", 2, 10, false) errCbFactory := func(wg *sync.WaitGroup, errMark *bool) func(error) { @@ -51,28 +89,27 @@ func TestWorkerManagerError(t *testing.T) { } } t.Run("normal", func(t *testing.T) { - wg := &sync.WaitGroup{} + task := &errWorkerTask{ + wg: &sync.WaitGroup{}, + } errMark := false - errCb := errCbFactory(wg, &errMark) - wg.Add(1) - wm.Run(func() { - defer wg.Done() - }, nil, errCb) - wg.Wait() + errCb := errCbFactory(task.wg, &errMark) + task.wg.Add(1) + wm.Run(task, nil, errCb) + task.wg.Wait() if errMark { t.Errorf("should be normal") } }) t.Run("panic", func(t *testing.T) { - wg := &sync.WaitGroup{} + task := &panicWorkerTask{ + wg: &sync.WaitGroup{}, + } errMark := false - errCb := errCbFactory(wg, &errMark) - wg.Add(2) // 1 for errCb - wm.Run(func() { - defer wg.Done() - panic("panic inside worker") - }, nil, errCb) - wg.Wait() + errCb := errCbFactory(task.wg, &errMark) + task.wg.Add(2) // 1 for errCb + wm.Run(task, nil, errCb) + task.wg.Wait() if !errMark { t.Errorf("expecting error") } diff --git a/pkg/baremetal/handler/delay.go b/pkg/baremetal/handler/delay.go index c716aaabd1..6e067768b5 100644 --- a/pkg/baremetal/handler/delay.go +++ b/pkg/baremetal/handler/delay.go @@ -15,6 +15,8 @@ package handler import ( + "fmt" + "yunion.io/x/jsonutils" "yunion.io/x/onecloud/pkg/appsrv" @@ -37,6 +39,21 @@ type delayTask struct { data jsonutils.JSONObject } +func (t *delayTask) Run() { + ret, err := t.process(t.data) + if err != nil { + modules.ComputeTasks.TaskFailed(t.session, t.taskId, err) + return + } + if len(t.taskId) > 0 { + modules.ComputeTasks.TaskComplete(t.session, t.taskId, ret) + } +} + +func (t *delayTask) Dump() string { + return fmt.Sprintf("taskId: %s data: %v", t.taskId, t.data) +} + func newDelayTask(process ProcessFunc, session *mcclient.ClientSession, taskId string, data jsonutils.JSONObject) *delayTask { return &delayTask{ process: process, @@ -47,18 +64,11 @@ func newDelayTask(process ProcessFunc, session *mcclient.ClientSession, taskId s } func DelayProcess(process ProcessFunc, session *mcclient.ClientSession, taskId string, data jsonutils.JSONObject) { - delayTaskWorkerMan.Run(func() { - executeDelayProcess(newDelayTask(process, session, taskId, data)) - }, nil, nil) -} - -func executeDelayProcess(task *delayTask) { - ret, err := task.process(task.data) - if err != nil { - modules.ComputeTasks.TaskFailed(task.session, task.taskId, err) - return - } - if len(task.taskId) > 0 { - modules.ComputeTasks.TaskComplete(task.session, task.taskId, ret) + task := &delayTask{ + process: process, + taskId: taskId, + session: session, + data: data, } + delayTaskWorkerMan.Run(task, nil, nil) } diff --git a/pkg/baremetal/tasks/worker.go b/pkg/baremetal/tasks/worker.go index 1ac0c084f3..26e884f507 100644 --- a/pkg/baremetal/tasks/worker.go +++ b/pkg/baremetal/tasks/worker.go @@ -44,10 +44,25 @@ func OnStop() { } } +type baremtalTask struct { + task ITask + args interface{} +} + +func (t *baremtalTask) Run() { + executeTask(t.task, t.args) +} + +func (t *baremtalTask) Dump() string { + return fmt.Sprintf("Task %s(%s) params: %v", t.task.GetName(), t.task.GetTaskId(), t.args) +} + func ExecuteTask(task ITask, args interface{}) { - baremetalTaskWorkerMan.Run(func() { - executeTask(task, args) - }, nil, nil) + t := &baremtalTask{ + task: task, + args: args, + } + baremetalTaskWorkerMan.Run(t, nil, nil) } func executeTask(task ITask, args interface{}) { diff --git a/pkg/cloudcommon/cronman/cronman.go b/pkg/cloudcommon/cronman/cronman.go index e65af0f38b..a9f9a8c941 100644 --- a/pkg/cloudcommon/cronman/cronman.go +++ b/pkg/cloudcommon/cronman/cronman.go @@ -377,10 +377,17 @@ func (self *SCronJobManager) runJobs(now time.Time) { } } +func (job *SCronJob) Run() { + job.runJobInWorker(job.StartRun) +} + +func (job *SCronJob) Dump() string { + return "" +} + func (job *SCronJob) runJob(isStart bool) { - manager.workers.Run(func() { - job.runJobInWorker(isStart) - }, nil, nil) + job.StartRun = isStart + manager.workers.Run(job, nil, nil) } func (job *SCronJob) runJobInWorker(isStart bool) { diff --git a/pkg/cloudcommon/db/quotas/usageworker.go b/pkg/cloudcommon/db/quotas/usageworker.go index 5f9ff4d72d..0b71c249ca 100644 --- a/pkg/cloudcommon/db/quotas/usageworker.go +++ b/pkg/cloudcommon/db/quotas/usageworker.go @@ -61,6 +61,49 @@ func isDirty(key string) bool { return false } +type quotaTask struct { + manager *SQuotaBaseManager + keys IQuotaKeys + usageChan chan IQuota + key string +} + +func (t *quotaTask) Run() { + ctx := context.Background() + + usage := t.manager.newQuota() + + if !isDirty(t.key) { + if t.usageChan != nil { + t.manager.usageStore.GetQuota(ctx, t.keys, usage) + t.usageChan <- usage + } + return + } + + usage.SetKeys(t.keys) + err := usage.FetchUsage(ctx) + if err != nil { + log.Debugf("usage.FetchUsage fail %s", err) + if t.usageChan != nil { + t.usageChan <- nil + } + return + } + + t.manager.usageStore.SetQuota(ctx, nil, usage) + + clearDirty(t.key) + + if t.usageChan != nil { + t.usageChan <- usage + } +} + +func (t *quotaTask) Dump() string { + return "" +} + func (manager *SQuotaBaseManager) PostUsageJob(keys IQuotaKeys, usageChan chan IQuota, realTime bool) { if !consts.EnableQuotaCheck() { go func() { @@ -78,37 +121,14 @@ func (manager *SQuotaBaseManager) PostUsageJob(keys IQuotaKeys, usageChan chan I worker = usageCalculateWorker } - worker.Run(func() { - ctx := context.Background() + task := quotaTask{ + manager: manager, + keys: keys, + usageChan: usageChan, + key: key, + } - usage := manager.newQuota() - - if !isDirty(key) { - if usageChan != nil { - manager.usageStore.GetQuota(ctx, keys, usage) - usageChan <- usage - } - return - } - - usage.SetKeys(keys) - err := usage.FetchUsage(ctx) - if err != nil { - log.Debugf("usage.FetchUsage fail %s", err) - if usageChan != nil { - usageChan <- nil - } - return - } - - manager.usageStore.SetQuota(ctx, nil, usage) - - clearDirty(key) - - if usageChan != nil { - usageChan <- usage - } - }, nil, nil) + worker.Run(&task, nil, nil) } func (manager *SQuotaBaseManager) CalculateQuotaUsages(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) { diff --git a/pkg/cloudcommon/db/taskman/localtaskworker.go b/pkg/cloudcommon/db/taskman/localtaskworker.go index 7f68ffb4fa..75e1365ba5 100644 --- a/pkg/cloudcommon/db/taskman/localtaskworker.go +++ b/pkg/cloudcommon/db/taskman/localtaskworker.go @@ -37,27 +37,40 @@ func Error2TaskData(err error) jsonutils.JSONObject { return errJson } -func LocalTaskRunWithWorkers(task ITask, proc func() (jsonutils.JSONObject, error), wm *appsrv.SWorkerManager) { - wm.Run(func() { +type localTask struct { + task ITask + proc func() (jsonutils.JSONObject, error) +} - log.Debugf("XXXXXXXXXXXXXXXXXXLOCAL TASK RUN STARTXXXXXXXXXXXXXXXXX") - defer log.Debugf("XXXXXXXXXXXXXXXXXXLOCAL TASK RUN END XXXXXXXXXXXXXXXXX") +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 { - log.Errorf("LocalTaskRun error: %s", r) - debug.PrintStack() - task.ScheduleRun(Error2TaskData(fmt.Errorf("LocalTaskRun error: %s", r))) - } - }() - data, err := proc() - if err != nil { - task.ScheduleRun(Error2TaskData(err)) - } else { - task.ScheduleRun(data) + defer func() { + if r := recover(); r != nil { + log.Errorf("LocalTaskRun error: %s", r) + debug.PrintStack() + t.task.ScheduleRun(Error2TaskData(fmt.Errorf("LocalTaskRun error: %s", r))) } + }() + data, err := t.proc() + if err != nil { + t.task.ScheduleRun(Error2TaskData(err)) + } else { + t.task.ScheduleRun(data) + } +} - }, nil, nil) +func (t *localTask) Dump() string { + return fmt.Sprintf("StartTime: %s TaskId: %s Params: %s", t.task.GetStartTime(), t.task.GetTaskId(), t.task.GetParams()) +} + +func LocalTaskRunWithWorkers(task ITask, proc func() (jsonutils.JSONObject, error), wm *appsrv.SWorkerManager) { + t := localTask{ + task: task, + proc: proc, + } + wm.Run(&t, nil, nil) } func LocalTaskRun(task ITask, proc func() (jsonutils.JSONObject, error)) { diff --git a/pkg/cloudcommon/db/taskman/worker.go b/pkg/cloudcommon/db/taskman/worker.go index f791525434..0b5feb6877 100644 --- a/pkg/cloudcommon/db/taskman/worker.go +++ b/pkg/cloudcommon/db/taskman/worker.go @@ -32,6 +32,19 @@ func init() { taskWorkerTable = make(map[string]*appsrv.SWorkerManager) } +type taskTask struct { + taskId string + data jsonutils.JSONObject +} + +func (t *taskTask) Run() { + TaskManager.execTask(t.taskId, t.data) +} + +func (t *taskTask) Dump() string { + return jsonutils.Marshal(t).PrettyString() +} + func runTask(taskId string, data jsonutils.JSONObject) error { taskName := TaskManager.getTaskName(taskId) if len(taskName) == 0 { @@ -41,9 +54,13 @@ func runTask(taskId string, data jsonutils.JSONObject) error { if workerMan, ok := taskWorkerTable[taskName]; ok { worker = workerMan } - isOk := worker.Run(func() { - TaskManager.execTask(taskId, data) - }, nil, func(err error) { + + task := &taskTask{ + taskId: taskId, + data: data, + } + + isOk := worker.Run(task, nil, func(err error) { panicutils.SendPanicMessage(context.TODO(), err) }) if !isOk { diff --git a/pkg/cloudcommon/informer/worker.go b/pkg/cloudcommon/informer/worker.go index 68afba9f94..018b5a244e 100644 --- a/pkg/cloudcommon/informer/worker.go +++ b/pkg/cloudcommon/informer/worker.go @@ -51,19 +51,33 @@ func (c noCancel) Value(key interface{}) interface{} { return c.ctx.Value(key) }*/ +type informerTask struct { + be IInformerBackend + f func(ctx context.Context, be IInformerBackend) error +} + +func (t *informerTask) Run() { + nopanic.Run(func() { + // outside context ignored cause of run in worker + if err := t.f(context.Background(), t.be); err != nil { + log.Errorf("run informer error: %v", err) + } + }) +} + +func (t *informerTask) Dump() string { + return "" +} + func run(ctx context.Context, f func(ctx context.Context, be IInformerBackend) error) error { be := GetDefaultBackend() if be == nil { return ErrBackendNotInit } - wf := func() { - nopanic.Run(func() { - // outside context ignored cause of run in worker - if err := f(context.Background(), be); err != nil { - log.Errorf("run informer error: %v", err) - } - }) + task := informerTask{ + f: f, + be: be, } - informerWorkerMan.Run(wf, nil, nil) + informerWorkerMan.Run(&task, nil, nil) return nil } diff --git a/pkg/cloudcommon/notifyclient/notify.go b/pkg/cloudcommon/notifyclient/notify.go index fac49629c8..4ff2e2e569 100644 --- a/pkg/cloudcommon/notifyclient/notify.go +++ b/pkg/cloudcommon/notifyclient/notify.go @@ -16,6 +16,7 @@ package notifyclient import ( "context" + "fmt" "time" "yunion.io/x/jsonutils" @@ -218,6 +219,26 @@ type SEventNotifyParam struct { AdvanceDays int } +type eventTask struct { + params api.NotificationManagerEventNotifyInput +} + +func (t *eventTask) Dump() string { + return fmt.Sprintf("eventTask params: %v", t.params) +} + +func (t *eventTask) Run() { + s, err := AdminSessionGenerator(context.Background(), "", "") + if err != nil { + log.Errorf("unable to get admin session: %v", err) + return + } + _, err = modules.Notification.PerformClassAction(s, "event-notify", jsonutils.Marshal(t.params)) + if err != nil { + log.Errorf("unable to EventNotify: %s", err) + } +} + func EventNotify(ctx context.Context, userCred mcclient.TokenCredential, ep SEventNotifyParam) { ret, err := db.FetchCustomizeColumns(ep.Obj.GetModelManager(), ctx, userCred, jsonutils.NewDict(), []interface{}{ep.Obj}, stringutils2.SSortedStrings{}, false) if err != nil { @@ -249,17 +270,10 @@ func EventNotify(ctx context.Context, userCred mcclient.TokenCredential, ep SEve ProjectId: projectId, ProjectDomainId: projectDomainId, } - notifyClientWorkerMan.Run(func() { - s, err := AdminSessionGenerator(context.Background(), "", "") - if err != nil { - log.Errorf("unable to get admin session: %v", err) - return - } - _, err = modules.Notification.PerformClassAction(s, "event-notify", jsonutils.Marshal(params)) - if err != nil { - log.Errorf("unable to EventNotify: %s", err) - } - }, nil, nil) + t := eventTask{ + params: params, + } + notifyClientWorkerMan.Run(&t, nil, nil) } func RawNotifyWithCtx(ctx context.Context, recipientId []string, isGroup bool, channel npk.TNotifyChannel, priority npk.TNotifyPriority, event string, data jsonutils.JSONObject) { diff --git a/pkg/cloudcommon/notifyclient/notify_internal.go b/pkg/cloudcommon/notifyclient/notify_internal.go index 511a7de8dc..9c93aee904 100644 --- a/pkg/cloudcommon/notifyclient/notify_internal.go +++ b/pkg/cloudcommon/notifyclient/notify_internal.go @@ -2,6 +2,7 @@ package notifyclient import ( "context" + "fmt" "regexp" "strings" "time" @@ -301,6 +302,57 @@ const noSuchReceiver = `no such receiver whose uid is '(.*)'` var noSuchReceiverRegexp = regexp.MustCompile(noSuchReceiver) +type notifyTask struct { + ctx context.Context + msg npk.SNotifyMessage + createReceiver bool +} + +func (t *notifyTask) Dump() string { + return fmt.Sprintf("msg: %v createReceiver: %v", t.msg, t.createReceiver) +} + +func (t *notifyTask) Run() { + s, err := AdminSessionGenerator(t.ctx, consts.GetRegion(), "") + if err != nil { + log.Errorf("fail to get session: %v", err) + } + for { + err := npk.Notifications.Send(s, t.msg) + if err == nil { + break + } + if !t.createReceiver { + log.Errorf("unable to send notification: %v", err) + break + } + jerr, ok := err.(*httputils.JSONClientError) + if !ok { + log.Errorf("unable to send notification: %v", err) + break + } + if jerr.Code > 500 { + log.Errorf("unable to send notification: %v", err) + break + } + match := noSuchReceiverRegexp.FindStringSubmatch(jerr.Details) + if match == nil || len(match) <= 1 { + log.Errorf("unable to send notification: %v", err) + break + } + receiverId := match[1] + createData := jsonutils.NewDict() + createData.Set("uid", jsonutils.NewString(receiverId)) + _, err = modules.NotifyReceiver.Create(s, createData) + if err != nil { + log.Errorf("try to create receiver %q, but failed: %v", receiverId, err) + break + } + log.Infof("create receiver %q successfully", receiverId) + } + return +} + func intelliNotify(ctx context.Context, p sNotifyParams) { log.Infof("recipientId: %v, contacts: %v, event %s priority %s", p.recipientId, p.contacts, p.event, p.priority) msgs, err := genMsgViaLang(ctx, p) @@ -308,47 +360,12 @@ func intelliNotify(ctx context.Context, p sNotifyParams) { log.Errorf("unable send notification: %v", err) } for i := range msgs { - msg := msgs[i] - notifyClientWorkerMan.Run(func() { - s, err := AdminSessionGenerator(context.Background(), consts.GetRegion(), "") - if err != nil { - log.Errorf("fail to get session: %v", err) - } - for { - err := npk.Notifications.Send(s, msg) - if err == nil { - break - } - if !p.createReceiver { - log.Errorf("unable to send notification: %v", err) - break - } - jerr, ok := err.(*httputils.JSONClientError) - if !ok { - log.Errorf("unable to send notification: %v", err) - break - } - if jerr.Code > 500 { - log.Errorf("unable to send notification: %v", err) - break - } - match := noSuchReceiverRegexp.FindStringSubmatch(jerr.Details) - if match == nil || len(match) <= 1 { - log.Errorf("unable to send notification: %v", err) - break - } - receiverId := match[1] - createData := jsonutils.NewDict() - createData.Set("uid", jsonutils.NewString(receiverId)) - _, err = modules.NotifyReceiver.Create(s, createData) - if err != nil { - log.Errorf("try to create receiver %q, but failed: %v", receiverId, err) - break - } - log.Infof("create receiver %q successfully", receiverId) - } - }, nil, nil) - + t := notifyTask{ + ctx: context.Background(), + createReceiver: p.createReceiver, + msg: msgs[i], + } + notifyClientWorkerMan.Run(&t, nil, nil) } // log.Debugf("send notification %s %s", topic, body) } diff --git a/pkg/cloudcommon/policy/policy.go b/pkg/cloudcommon/policy/policy.go index ef6fd2d197..1843c795ae 100644 --- a/pkg/cloudcommon/policy/policy.go +++ b/pkg/cloudcommon/policy/policy.go @@ -222,32 +222,52 @@ func (manager *SPolicyManager) Allow(targetScope rbacutils.TRbacScope, userCred return rbacutils.Deny } +type fetchResult struct { + output *mcclient.SFetchMatchPoliciesOutput + err error +} + +type policyTask struct { + manager *SPolicyManager + key string + userCred mcclient.TokenCredential + resChan chan fetchResult +} + +func (t *policyTask) Run() { + val := t.manager.policyCache.Get(t.key) + result := fetchResult{} + if gotypes.IsNil(val) { + pg, err := DefaultPolicyFetcher(context.Background(), t.userCred) + if err != nil { + result.err = errors.Wrap(err, "DefaultPolicyFetcher") + } else { + t.manager.policyCache.Set(t.key, pg) + result.output = pg + } + } else { + result.output = val.(*mcclient.SFetchMatchPoliciesOutput) + } + t.resChan <- result +} + +func (t *policyTask) Dump() string { + return "" +} + func (manager *SPolicyManager) fetchMatchedPolicies(userCred mcclient.TokenCredential) (*mcclient.SFetchMatchPoliciesOutput, error) { key := policyKey(userCred) - type fetchResult struct { - output *mcclient.SFetchMatchPoliciesOutput - err error + task := policyTask{ + manager: manager, + key: key, + userCred: userCred, } - resChan := make(chan fetchResult) - manager.fetchWorker.Run(func() { - val := manager.policyCache.Get(key) - result := fetchResult{} - if gotypes.IsNil(val) { - pg, err := DefaultPolicyFetcher(context.Background(), userCred) - if err != nil { - result.err = errors.Wrap(err, "DefaultPolicyFetcher") - } else { - manager.policyCache.Set(key, pg) - result.output = pg - } - } else { - result.output = val.(*mcclient.SFetchMatchPoliciesOutput) - } - resChan <- result - }, nil, nil) + task.resChan = make(chan fetchResult) - res := <-resChan + manager.fetchWorker.Run(&task, nil, nil) + + res := <-task.resChan return res.output, res.err } diff --git a/pkg/cloudcommon/syncman/sync.go b/pkg/cloudcommon/syncman/sync.go index 3bfecb0497..847faa2eed 100644 --- a/pkg/cloudcommon/syncman/sync.go +++ b/pkg/cloudcommon/syncman/sync.go @@ -63,13 +63,26 @@ func (manager *SSyncManager) syncByInterval() error { return err } +type SyncTask struct { + manager *SSyncManager +} + +func (t *SyncTask) Run() { + atomic.StoreInt32(&t.manager.syncOnce, 0) + t.manager.syncByInterval() +} + +func (t *SyncTask) Dump() string { + return "" +} + func (manager *SSyncManager) SyncOnce() { log.Debugf("[%s] SyncOnce", manager.Name()) if atomic.CompareAndSwapInt32(&manager.syncOnce, 0, 1) { - manager.syncWorkerManager.Run(func() { - atomic.StoreInt32(&manager.syncOnce, 0) - manager.syncByInterval() - }, nil, nil) + task := SyncTask{ + manager: manager, + } + manager.syncWorkerManager.Run(&task, nil, nil) } } diff --git a/pkg/cloudcommon/workmanager/manager.go b/pkg/cloudcommon/workmanager/manager.go index f9482cef8d..af9e58cc42 100644 --- a/pkg/cloudcommon/workmanager/manager.go +++ b/pkg/cloudcommon/workmanager/manager.go @@ -57,6 +57,50 @@ func (w *SWorkManager) DelayTaskWithWorker( w.delayTask(ctx, task, params, worker) } +type workerTask struct { + ctx context.Context + w *SWorkManager + task DelayTaskFunc + params interface{} +} + +func (t *workerTask) Run() { + defer t.w.done() + defer func() { + if r := recover(); r != nil { + log.Errorf("DelayTask panic: %s", r) + debug.PrintStack() + switch val := r.(type) { + case string: + t.w.onFailed(t.ctx, val) + case error: + t.w.onFailed(t.ctx, val.Error()) + default: + t.w.onFailed(t.ctx, "Unknown panic") + } + } + }() + + // HACK: callback only + if t.task == nil { + t.w.onCompleted(t.ctx, nil) + return + } + + res, err := t.task(t.ctx, t.params) + if err != nil { + log.Infof("DelayTask failed: %s", err) + t.w.onFailed(t.ctx, err.Error()) + } else { + log.Infof("DelayTask complete: %v", res) + t.w.onCompleted(t.ctx, res) + } +} + +func (t *workerTask) Dump() string { + return "" +} + // If delay task is not panic and task func return err is nil // task complete will be called, otherwise called task failed // Params is interface for receive any type, task func should do type assertion @@ -68,38 +112,13 @@ func (w *SWorkManager) delayTask(ctx context.Context, task DelayTaskFunc, params // delayTask should have a new context.Context with value 'taskid' ctx = context.WithValue(context.Background(), appctx.APP_CONTEXT_KEY_TASK_ID, ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID)) w.add() - worker.Run(func() { - defer w.done() - defer func() { - if r := recover(); r != nil { - log.Errorf("DelayTask panic: %s", r) - debug.PrintStack() - switch val := r.(type) { - case string: - w.onFailed(ctx, val) - case error: - w.onFailed(ctx, val.Error()) - default: - w.onFailed(ctx, "Unknown panic") - } - } - }() - - // HACK: callback only - if task == nil { - w.onCompleted(ctx, nil) - return - } - - res, err := task(ctx, params) - if err != nil { - log.Infof("DelayTask failed: %s", err) - w.onFailed(ctx, err.Error()) - } else { - log.Infof("DelayTask complete: %v", res) - w.onCompleted(ctx, res) - } - }, nil, nil) + t := workerTask{ + ctx: ctx, + w: w, + task: task, + params: params, + } + worker.Run(&t, nil, nil) } } @@ -107,6 +126,36 @@ func (w *SWorkManager) DelayTaskWithoutReqctx(ctx context.Context, task DelayTas w.delayTaskWithoutReqctx(ctx, task, params, w.worker) } +type delayWorkerTask struct { + w *SWorkManager + task DelayTaskFunc + ctx context.Context + params interface{} +} + +func (t *delayWorkerTask) Run() { + defer t.w.done() + defer func() { + if r := recover(); r != nil { + log.Errorln("DelayTaskWithoutReqctx panic: ", r) + debug.PrintStack() + } + }() + + if t.task == nil { + return + } + + if _, err := t.task(t.ctx, t.params); err != nil { + log.Errorln("DelayTaskWithoutReqctx error: ", err) + t.w.onFailed(t.ctx, err.Error()) + } +} + +func (t *delayWorkerTask) Dump() string { + return "" +} + // response task by self, did not callback func (w *SWorkManager) delayTaskWithoutReqctx( ctx context.Context, task DelayTaskFunc, params interface{}, worker *appsrv.SWorkerManager, @@ -118,24 +167,13 @@ func (w *SWorkManager) delayTaskWithoutReqctx( } ctx = newCtx w.add() - w.worker.Run(func() { - defer w.done() - defer func() { - if r := recover(); r != nil { - log.Errorln("DelayTaskWithoutReqctx panic: ", r) - debug.PrintStack() - } - }() - - if task == nil { - return - } - - if _, err := task(ctx, params); err != nil { - log.Errorln("DelayTaskWithoutReqctx error: ", err) - w.onFailed(ctx, err.Error()) - } - }, nil, nil) + t := delayWorkerTask{ + w: w, + task: task, + ctx: ctx, + params: params, + } + w.worker.Run(&t, nil, nil) } func (w *SWorkManager) Stop() { diff --git a/pkg/cloudid/models/cloudgroup.go b/pkg/cloudid/models/cloudgroup.go index 4dbc830b25..9420604a7d 100644 --- a/pkg/cloudid/models/cloudgroup.go +++ b/pkg/cloudid/models/cloudgroup.go @@ -17,6 +17,7 @@ package models import ( "context" "database/sql" + "time" "gopkg.in/fatih/set.v0" @@ -72,6 +73,8 @@ func (manager *SCloudgroupManager) ListItemFilter(ctx context.Context, q *sqlche return nil, err } + time.Sleep(time.Minute) + if len(query.Provider) > 0 { q = q.In("provider", query.Provider) } diff --git a/pkg/cloudid/models/clouduser.go b/pkg/cloudid/models/clouduser.go index 341303f3d4..fc4fc96ca8 100644 --- a/pkg/cloudid/models/clouduser.go +++ b/pkg/cloudid/models/clouduser.go @@ -18,6 +18,7 @@ import ( "context" "database/sql" "fmt" + "time" "gopkg.in/fatih/set.v0" @@ -125,6 +126,7 @@ func (manager *SClouduserManager) ListItemFilter(ctx context.Context, q *sqlchem if err != nil { return nil, err } + time.Sleep(time.Minute) q, err = manager.SCloudaccountResourceBaseManager.ListItemFilter(ctx, q, userCred, query.CloudaccountResourceListInput) if err != nil { diff --git a/pkg/compute/models/snapshotpolicycache.go b/pkg/compute/models/snapshotpolicycache.go index 68a688671e..f638788f08 100644 --- a/pkg/compute/models/snapshotpolicycache.go +++ b/pkg/compute/models/snapshotpolicycache.go @@ -401,6 +401,32 @@ func (spcm *SSnapshotPolicyCacheManager) UpdateCloudSnapshotPolicy(snapshotPolic return fmt.Errorf("Not implement") } +type snapshotPolicyTask struct { + ctx context.Context + userCred mcclient.TokenCredential + spc SSnapshotPolicyCache + retChan chan sOperaResult +} + +func (t *snapshotPolicyTask) Run() { + err := t.spc.DeleteCloudSnapshotPolicy() + if err != nil { + t.retChan <- sOperaResult{err, t.spc.GetId()} + return + } + err = t.spc.RealDetele(t.ctx, t.userCred) + if err != nil { + t.retChan <- sOperaResult{errors.Wrap(err, "delete cache in database failed"), t.spc.GetId()} + return + } + t.retChan <- sOperaResult{nil, t.spc.GetId()} + +} + +func (t *snapshotPolicyTask) Dump() string { + return "" +} + func (spcm *SSnapshotPolicyCacheManager) DeleteCloudSnapshotPolices(ctx context.Context, userCred mcclient.TokenCredential, snapshotPolicyId string) error { @@ -414,30 +440,23 @@ func (spcm *SSnapshotPolicyCacheManager) DeleteCloudSnapshotPolices(ctx context. } wm := appsrv.NewWorkerManager("delete-cloud-snapshotpolices", len(spCaches), 1, false) - retChan := make(chan sOperaResult) + + task := &snapshotPolicyTask{ + ctx: ctx, + userCred: userCred, + retChan: make(chan sOperaResult), + } for i := range spCaches { - spc := spCaches[i] - wm.Run(func() { - err := spc.DeleteCloudSnapshotPolicy() - if err != nil { - retChan <- sOperaResult{err, spc.GetId()} - return - } - err = spc.RealDetele(ctx, userCred) - if err != nil { - retChan <- sOperaResult{errors.Wrap(err, "delete cache in database failed"), spc.GetId()} - return - } - retChan <- sOperaResult{nil, spc.GetId()} - }, nil, func(e error) { - retChan <- sOperaResult{e, spc.GetId()} + task.spc = spCaches[i] + wm.Run(task, nil, func(e error) { + task.retChan <- sOperaResult{e, task.spc.GetId()} }) } failedRecord := make([]string, 0) for i := 0; i < len(spCaches); i++ { - ret := <-retChan + ret := <-task.retChan if ret.err != nil { failedRecord = append(failedRecord, fmt.Sprintf("%s failed because that %s", ret.snapshotPolicyId, ret.err.Error())) diff --git a/pkg/compute/models/syncworkers.go b/pkg/compute/models/syncworkers.go index 0413b2c3e8..68d06aa60e 100644 --- a/pkg/compute/models/syncworkers.go +++ b/pkg/compute/models/syncworkers.go @@ -54,17 +54,38 @@ func InitSyncWorkers(count int) { ) } +type resSyncTask struct { + syncFunc func() + key string +} + +func (t *resSyncTask) Run() { + t.syncFunc() +} + +func (t *resSyncTask) Dump() string { + return fmt.Sprintf("key: %s", t.key) +} + func RunSyncCloudproviderRegionTask(ctx context.Context, key string, syncFunc func()) { nodeIdxStr, _ := syncWorkerRing.GetNode(key) nodeIdx, _ := strconv.Atoi(nodeIdxStr) + task := resSyncTask{ + syncFunc: syncFunc, + key: key, + } log.Debugf("run sync task at %d len %d", nodeIdx, len(syncWorkers)) - syncWorkers[nodeIdx].Run(syncFunc, nil, func(err error) { + syncWorkers[nodeIdx].Run(&task, nil, func(err error) { panicutils.SendPanicMessage(ctx, err) }) } func RunSyncCloudAccountTask(ctx context.Context, probeFunc func()) { - syncAccountWorker.Run(probeFunc, nil, func(err error) { + task := resSyncTask{ + syncFunc: probeFunc, + key: "AccountProb", + } + syncAccountWorker.Run(&task, nil, func(err error) { panicutils.SendPanicMessage(ctx, err) }) } diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index c0211e057a..e718bca508 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -279,14 +279,13 @@ func (s *SKVMGuestInstance) asyncScriptStart(ctx context.Context, params interfa s.syncMeta = s.CleanImportMetadata() s.StartMonitor(ctx) return nil, nil - } else { - log.Infof("Async start server %s failed: %s!!!", s.GetName(), err) - if ctx != nil && len(appctx.AppContextTaskId(ctx)) >= 0 { - hostutils.TaskFailed(ctx, fmt.Sprintf("Async start server failed: %s", err)) - } - s.SyncStatus() - return nil, err } + log.Infof("Async start server %s failed: %s!!!", s.GetName(), err) + if ctx != nil && len(appctx.AppContextTaskId(ctx)) >= 0 { + hostutils.TaskFailed(ctx, fmt.Sprintf("Async start server failed: %s", err)) + } + s.SyncStatus() + return nil, nil } func (s *SKVMGuestInstance) saveScripts(data *jsonutils.JSONDict) error { @@ -819,12 +818,28 @@ func (s *SKVMGuestInstance) GetVpcNIC() jsonutils.JSONObject { return nil } +type guestStartTask struct { + s *SKVMGuestInstance + + ctx context.Context + params *jsonutils.JSONDict +} + +func (t *guestStartTask) Run() { + t.s.asyncScriptStart(t.ctx, t.params) +} + +func (t *guestStartTask) Dump() string { + return fmt.Sprintf("guest %s params: %v", t.s.Id, t.params) +} + func (s *SKVMGuestInstance) StartGuest(ctx context.Context, params *jsonutils.JSONDict) { - if params == nil { - params = jsonutils.NewDict() + task := &guestStartTask{ + s: s, + ctx: ctx, + params: params, } - s.manager.GuestStartWorker.Run( - func() { s.asyncScriptStart(ctx, params) }, nil, nil) + s.manager.GuestStartWorker.Run(task, nil, nil) } func (s *SKVMGuestInstance) DeployFs(deployInfo *deployapi.DeployInfo) (jsonutils.JSONObject, error) { diff --git a/pkg/image/torrent/torrent.go b/pkg/image/torrent/torrent.go index dec787d125..56de07c308 100644 --- a/pkg/image/torrent/torrent.go +++ b/pkg/image/torrent/torrent.go @@ -69,14 +69,31 @@ func GetTrackers() []string { return urls } +type torrentTask struct { + torrentpath string + imageId string + format string +} + +func (t *torrentTask) Run() { + log.Infof("Start seed %s ...", t.torrentpath) + err := seedTorrent(t.torrentpath, t.imageId, t.format) + if err == nil { + time.Sleep(10 * time.Second) + } +} + +func (t *torrentTask) Dump() string { + return fmt.Sprintf("torrentpath: %s imageId: %s.%s", t.torrentpath, t.imageId, t.format) +} + func SeedTorrent(torrentpath string, imageId, format string) error { - seedTaskWorkerMan.Run(func() { - log.Infof("Start seed %s ...", torrentpath) - err := seedTorrent(torrentpath, imageId, format) - if err == nil { - time.Sleep(10 * time.Second) - } - }, nil, nil) + task := &torrentTask{ + torrentpath: torrentpath, + imageId: imageId, + format: format, + } + seedTaskWorkerMan.Run(task, nil, nil) return nil } diff --git a/pkg/keystone/models/syncworker.go b/pkg/keystone/models/syncworker.go index d109a67495..33db539c99 100644 --- a/pkg/keystone/models/syncworker.go +++ b/pkg/keystone/models/syncworker.go @@ -16,7 +16,9 @@ package models import ( "context" + "fmt" + "yunion.io/x/jsonutils" "yunion.io/x/log" api "yunion.io/x/onecloud/pkg/apis/identity" @@ -38,38 +40,54 @@ func InitSyncWorkers() { ) } +type syncTask struct { + ctx context.Context + userCred mcclient.TokenCredential + idp *SIdentityProvider +} + +func (t *syncTask) Run() { + t.idp.SetSyncStatus(t.ctx, t.userCred, api.IdentitySyncStatusSyncing) + defer t.idp.SetSyncStatus(t.ctx, t.userCred, api.IdentitySyncStatusIdle) + + conf, err := GetConfigs(t.idp, true, nil, nil) + if err != nil { + log.Errorf("GetConfig for idp %s fail %s", t.idp.Name, err) + t.idp.MarkDisconnected(t.ctx, t.userCred, err) + return + } + driver, err := driver.GetDriver(t.idp.Driver, t.idp.Id, t.idp.Name, t.idp.Template, t.idp.TargetDomainId, conf) + if err != nil { + log.Errorf("GetDriver for idp %s fail %s", t.idp.Name, err) + t.idp.MarkDisconnected(t.ctx, t.userCred, err) + return + } + err = driver.Probe(t.ctx) + if err != nil { + log.Errorf("Probe for idp %s fail %s", t.idp.Name, err) + t.idp.MarkDisconnected(t.ctx, t.userCred, err) + return + } + + t.idp.MarkConnected(t.ctx, t.userCred) + + err = driver.Sync(t.ctx) + if err != nil { + log.Errorf("Sync for idp %s fail %s", t.idp.Name, err) + return + } +} + +func (t *syncTask) Dump() string { + return fmt.Sprintf("idp %s", jsonutils.Marshal(t.idp).String()) +} + func submitIdpSyncTask(ctx context.Context, userCred mcclient.TokenCredential, idp *SIdentityProvider) { idp.SetSyncStatus(ctx, userCred, api.IdentitySyncStatusQueued) - syncWorker.Run(func() { - idp.SetSyncStatus(ctx, userCred, api.IdentitySyncStatusSyncing) - defer idp.SetSyncStatus(ctx, userCred, api.IdentitySyncStatusIdle) - - conf, err := GetConfigs(idp, true, nil, nil) - if err != nil { - log.Errorf("GetConfig for idp %s fail %s", idp.Name, err) - idp.MarkDisconnected(ctx, userCred, err) - return - } - driver, err := driver.GetDriver(idp.Driver, idp.Id, idp.Name, idp.Template, idp.TargetDomainId, conf) - if err != nil { - log.Errorf("GetDriver for idp %s fail %s", idp.Name, err) - idp.MarkDisconnected(ctx, userCred, err) - return - } - err = driver.Probe(ctx) - if err != nil { - log.Errorf("Probe for idp %s fail %s", idp.Name, err) - idp.MarkDisconnected(ctx, userCred, err) - return - } - - idp.MarkConnected(ctx, userCred) - - err = driver.Sync(ctx) - if err != nil { - log.Errorf("Sync for idp %s fail %s", idp.Name, err) - return - } - - }, nil, nil) + task := &syncTask{ + ctx: ctx, + userCred: userCred, + idp: idp, + } + syncWorker.Run(task, nil, nil) } diff --git a/pkg/mcclient/catalog.go b/pkg/mcclient/catalog.go index 93e82b566c..524aa9cbaf 100644 --- a/pkg/mcclient/catalog.go +++ b/pkg/mcclient/catalog.go @@ -27,22 +27,50 @@ type IServiceCatalogChangeListener interface { OnServiceCatalogChange(catalog IServiceCatalog) } +type cliTask struct { + cli *Client + l IServiceCatalogChangeListener + catalog IServiceCatalog + + taskType string +} + +func (t *cliTask) Run() { + switch t.taskType { + case "GetServiceCatalog": + t.l.OnServiceCatalogChange(t.cli.GetServiceCatalog()) + case "OnServiceCatalogChange": + for i := range t.cli.catalogListeners { + t.cli.catalogListeners[i].OnServiceCatalogChange(t.catalog) + } + } +} + +func (t *cliTask) Dump() string { + return "" +} + func (cli *Client) RegisterCatalogListener(l IServiceCatalogChangeListener) { cli.catalogListeners = append(cli.catalogListeners, l) + task := &cliTask{ + cli: cli, + l: l, + taskType: "GetServiceCatalog", + } if cli.GetServiceCatalog() != nil { - listenerWorker.Run(func() { - l.OnServiceCatalogChange(cli.GetServiceCatalog()) - }, nil, nil) + listenerWorker.Run(task, nil, nil) } } func (cli *Client) SetServiceCatalog(catalog IServiceCatalog) { cli._serviceCatalog = catalog - listenerWorker.Run(func() { - for i := range cli.catalogListeners { - cli.catalogListeners[i].OnServiceCatalogChange(catalog) - } - }, nil, nil) + task := &cliTask{ + cli: cli, + catalog: catalog, + taskType: "OnServiceCatalogChange", + } + + listenerWorker.Run(task, nil, nil) } func (this *Client) GetServiceCatalog() IServiceCatalog { diff --git a/pkg/monitor/service/dispatcher.go b/pkg/monitor/service/dispatcher.go index e2fa37fae2..7b23ba001c 100644 --- a/pkg/monitor/service/dispatcher.go +++ b/pkg/monitor/service/dispatcher.go @@ -35,14 +35,30 @@ func addCommonAlertDispatcher(prefix string, app *appsrv.Application) { performHandler, metadata, "perform_class_subscription", tags) } +type subscriptionTask struct { + ctx context.Context + query jsonutils.JSONObject + body []sub.Point +} + +func (t *subscriptionTask) Run() { + t.ctx = context.WithValue(context.Background(), appctx.APP_CONTEXT_KEY_AUTH_TOKEN, auth.AdminCredential()) + subscriptionmodel.SubscriptionManager.PerformWrite(t.ctx, auth.AdminCredential(), t.query, t.body) +} + +func (t *subscriptionTask) Dump() string { + return "" +} + func performHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) { _, query, body := fetchEnv(ctx, w, r) appsrv.SendJSON(w, wrap(jsonutils.NewDict(), "subscription")) - SubscriptionWorkerManager.Run(func() { - ctx = context.WithValue(context.Background(), appctx.APP_CONTEXT_KEY_AUTH_TOKEN, auth.AdminCredential()) - subscriptionmodel.SubscriptionManager.PerformWrite(ctx, auth.AdminCredential(), query, body) - }, nil, nil) - + task := &subscriptionTask{ + ctx: ctx, + query: query, + body: body, + } + SubscriptionWorkerManager.Run(task, nil, nil) } // fetchEnv fetch handler, params, query and body from ctx(context.Context) diff --git a/pkg/util/logclient/logclient.go b/pkg/util/logclient/logclient.go index 9baa5181cf..3e6ba160a2 100644 --- a/pkg/util/logclient/logclient.go +++ b/pkg/util/logclient/logclient.go @@ -16,6 +16,7 @@ package logclient import ( "context" + "fmt" "strings" "time" @@ -160,11 +161,29 @@ func addLog(model IObject, action string, iNotes interface{}, userCred mcclient. logentry.Add(jsonutils.NewString(notes), "notes") - logclientWorkerMan.Run(func() { - s := DefaultSessionGenerator(context.Background(), auth.AdminCredential(), "", "") - _, err := api.Create(s, logentry) - if err != nil { - log.Errorf("create action log %s failed %s", logentry, err) - } - }, nil, nil) + task := &logTask{ + userCred: auth.AdminCredential(), + api: api, + logentry: logentry, + } + + logclientWorkerMan.Run(task, nil, nil) +} + +type logTask struct { + userCred mcclient.TokenCredential + api IModule + logentry *jsonutils.JSONDict +} + +func (t *logTask) Run() { + s := DefaultSessionGenerator(context.Background(), t.userCred, "", "") + _, err := t.api.Create(s, t.logentry) + if err != nil { + log.Errorf("create action log %s failed %s", t.logentry, err) + } +} + +func (t *logTask) Dump() string { + return fmt.Sprintf("logTask %v %s", t.api, t.logentry) }