opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-01 15:07:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(common): dump task info when stucking

Browse Source
This commit is contained in:
Qu Xuan
2021-01-18 17:21:19 +08:00
parent ecae338a49
commit 7b845d17c7
25 changed files with 809 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/appsrv/appsrv.go
+74 -48
View File
@@ -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)
pkg/appsrv/workers.go
+29 -7
View File
@@ -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()
}
}
}
pkg/appsrv/workers_test.go
+57 -20
View File
@@ -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")
}
pkg/baremetal/handler/delay.go
+23 -13
View File
@@ -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)
}
pkg/baremetal/tasks/worker.go
+18 -3
View File
@@ -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{}) {
pkg/cloudcommon/cronman/cronman.go
+10 -3
View File
@@ -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) {
pkg/cloudcommon/db/quotas/usageworker.go
+50 -30
View File
@@ -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) {
pkg/cloudcommon/db/taskman/localtaskworker.go
+30 -17
View File
@@ -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)) {
pkg/cloudcommon/db/taskman/worker.go
+20 -3
View File
@@ -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 {
pkg/cloudcommon/informer/worker.go
+22 -8
View File
@@ -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
}
pkg/cloudcommon/notifyclient/notify.go
+25 -11
View File
@@ -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) {
pkg/cloudcommon/notifyclient/notify_internal.go
+58 -41
View File
@@ -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)
}
pkg/cloudcommon/policy/policy.go
+41 -21
View File
@@ -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
}
pkg/cloudcommon/syncman/sync.go
+17 -4
View File
@@ -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)
}
}
pkg/cloudcommon/workmanager/manager.go
+88 -50
View File
@@ -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() {
pkg/cloudid/models/cloudgroup.go
+3
View File
@@ -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)
}
pkg/cloudid/models/clouduser.go
+2
View File
@@ -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 {
pkg/compute/models/snapshotpolicycache.go
+36 -17
View File
@@ -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()))
pkg/compute/models/syncworkers.go
+23 -2
View File
@@ -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)
})
}
pkg/hostman/guestman/qemu-kvm.go
+26 -11
View File
@@ -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) {
pkg/image/torrent/torrent.go
+24 -7
View File
@@ -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
}
pkg/keystone/models/syncworker.go
+50 -32
View File
@@ -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)
}
pkg/mcclient/catalog.go
+36 -8
View File
@@ -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 {
pkg/monitor/service/dispatcher.go
+21 -5
View File
@@ -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)
pkg/util/logclient/logclient.go
+26 -7
View File
@@ -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)
}