fix(monitor): not depend taskman to execute async task (#15065)

pkg/monitor/models/commonalert.go

@@ -35,7 +35,6 @@ import (
 	notiapi "yunion.io/x/onecloud/pkg/apis/notify"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
-	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
 	"yunion.io/x/onecloud/pkg/hostman/hostinfo/hostconsts"
 	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/i18n"
@@ -1142,14 +1141,30 @@ func (alert *SCommonAlert) RealDelete(ctx context.Context, userCred mcclient.Tok
 	return alert.SStandaloneResourceBase.Delete(ctx, userCred)
 }
 
-func (self *SCommonAlert) StartDeleteTask(
-	ctx context.Context, userCred mcclient.TokenCredential) error {
-	task, err := taskman.TaskManager.NewTask(ctx, "DeleteAlertRecordTask", self, userCred, jsonutils.NewDict(), "", "", nil)
-	if err != nil {
-		return err
-	}
-	task.ScheduleRun(nil)
-	return nil
+func (self *SCommonAlert) StartDeleteTask(ctx context.Context, userCred mcclient.TokenCredential) {
+	RunModelTask("DeleteAlertRecordTask", self, func() error {
+		onErr := func(err error) {
+			msg := jsonutils.NewString(err.Error())
+			// db.OpsLog.LogEvent(self, db.ACT_DELETE_FAIL, msg, userCred)
+			logclient.AddActionLogWithContext(ctx, self, logclient.ACT_DELETE, msg, userCred, false)
+		}
+
+		errs := self.DeleteAttachAlertRecords(ctx, userCred)
+		if len(errs) != 0 {
+			err := errors.Wrapf(errors.NewAggregate(errs), "DeleteAttachAlertRecords of %s", self.GetName())
+			onErr(err)
+			return err
+		}
+		if err := self.RealDelete(ctx, userCred); err != nil {
+			err = errors.Wrapf(err, "RealDelete CommonAlert %s", self.GetName())
+			onErr(err)
+			return err
+		}
+
+		// db.OpsLog.LogEvent(self, db.ACT_DELETE, nil, userCred)
+		logclient.AddActionLogWithContext(ctx, self, logclient.ACT_DELETE, nil, userCred, true)
+		return nil
+	})
 }
 
 func (self *SCommonAlert) DeleteAttachAlertRecords(ctx context.Context, userCred mcclient.TokenCredential) (errs []error) {
@@ -1280,7 +1295,7 @@ func (alert *SCommonAlert) PerformConfig(ctx context.Context, userCred mcclient.
 }
 
 func PerformConfigLog(model db.IModel, userCred mcclient.TokenCredential) {
-	db.OpsLog.LogEvent(model, db.ACT_UPDATE_RULE, "", userCred)
+	// db.OpsLog.LogEvent(model, db.ACT_UPDATE_RULE, "", userCred)
 	logclient.AddSimpleActionLog(model, logclient.ACT_UPDATE_RULE, nil, userCred, true)
 }
 
@@ -1306,11 +1321,25 @@ func (alert *SCommonAlert) PerformDisable(ctx context.Context, userCred mcclient
 }
 
 func (alert *SCommonAlert) StartDetachTask(ctx context.Context, userCred mcclient.TokenCredential) error {
-	task, err := taskman.TaskManager.NewTask(ctx, "DetachAlertResourceTask", alert, userCred, jsonutils.NewDict(), "", "", nil)
-	if err != nil {
-		return err
-	}
-	task.ScheduleRun(nil)
+	RunModelTask("DetachAlertResourceTask", alert, func() error {
+		onErr := func(err error) {
+			msg := jsonutils.NewString(err.Error())
+			// db.OpsLog.LogEvent(alert, db.ACT_DETACH, msg, userCred)
+			logclient.AddActionLogWithContext(ctx, alert, logclient.ACT_DETACH_ALERTRESOURCE, msg, userCred, false)
+		}
+		errs := alert.DetachAlertResourceOnDisable(ctx, userCred)
+		if len(errs) != 0 {
+			err := errors.Wrapf(errors.NewAggregate(errs), "DetachAlertResourceOnDisable of alert %s", alert.GetName())
+			onErr(err)
+			return err
+		}
+		if err := MonitorResourceAlertManager.DetachJoint(ctx, userCred,
+			monitor.MonitorResourceJointListInput{AlertId: alert.GetId()}); err != nil {
+			log.Errorf("DetachJoint when alert(%s) disable: %v", alert.GetName(), err)
+		}
+		logclient.AddActionLogWithContext(ctx, alert, logclient.ACT_DETACH_ALERTRESOURCE, nil, userCred, true)
+		return nil
+	})
 	return nil
 }
 
@@ -1335,13 +1364,13 @@ func (manager *SCommonAlertManager) DetachAlertResourceByAlertId(ctx context.Con
 	return
 }
 
-func (alert *SCommonAlert) StartUpdateMonitorAlertJointTask(ctx context.Context, userCred mcclient.TokenCredential) error {
-	task, err := taskman.TaskManager.NewTask(ctx, "UpdateMonitorResourceJointTask", alert, userCred, jsonutils.NewDict(), "", "", nil)
-	if err != nil {
-		return err
-	}
-	task.ScheduleRun(nil)
-	return nil
+func (alert *SCommonAlert) StartUpdateMonitorAlertJointTask(ctx context.Context, userCred mcclient.TokenCredential) {
+	RunModelTask("UpdateMonitorResourceJointTask", alert, func() error {
+		if err := alert.UpdateMonitorResourceJoint(ctx, userCred); err != nil {
+			return errors.Wrapf(err, "UpdateMonitorResourceJoint of alert %s", alert.GetName())
+		}
+		return nil
+	})
 }
 
 func (alert *SCommonAlert) UpdateMonitorResourceJoint(ctx context.Context, userCred mcclient.TokenCredential) error {
@@ -1417,14 +1446,18 @@ jointLoop:
 	return nil
 }
 
-func (alert *SCommonAlert) StartDetachMonitorAlertJointTask(ctx context.Context,
-	userCred mcclient.TokenCredential) error {
-	task, err := taskman.TaskManager.NewTask(ctx, "DetachMonitorResourceJointTask", alert, userCred, jsonutils.NewDict(), "", "", nil)
-	if err != nil {
-		return err
-	}
-	task.ScheduleRun(nil)
-	return nil
+func (alert *SCommonAlert) StartDetachMonitorAlertJointTask(ctx context.Context, userCred mcclient.TokenCredential) {
+	RunModelTask("DetachMonitorResourceJointTask", alert, func() error {
+		if err := alert.DetachMonitorResourceJoint(ctx, userCred); err != nil {
+			err = errors.Wrapf(err, "DetachMonitorResourceJoint of alert %s", alert.GetName())
+			msg := jsonutils.NewString(err.Error())
+			// db.OpsLog.LogEvent(alert, db.ACT_DETACH_MONITOR_RESOURCE_JOINT, msg, userCred)
+			logclient.AddActionLogWithContext(ctx, alert, logclient.ACT_DETACH_MONITOR_RESOURCE_JOINT, msg, userCred, false)
+			return err
+		}
+		logclient.AddActionLogWithContext(ctx, alert, logclient.ACT_DETACH_MONITOR_RESOURCE_JOINT, nil, userCred, true)
+		return nil
+	})
 }
 
 func (alert *SCommonAlert) DetachMonitorResourceJoint(ctx context.Context, userCred mcclient.TokenCredential) error {

pkg/monitor/models/worker.go

@@ -0,0 +1,65 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package models
+
+import (
+	"fmt"
+
+	"yunion.io/x/log"
+
+	"yunion.io/x/onecloud/pkg/appsrv"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+)
+
+var modelTaskMan *appsrv.SWorkerManager
+
+func GetModelTaskManager() *appsrv.SWorkerManager {
+	if modelTaskMan == nil {
+		modelTaskMan = appsrv.NewWorkerManager("ModelTaskWorkerManager", 4, 1024, true)
+	}
+	return modelTaskMan
+}
+
+type modelTask struct {
+	name           string
+	object         db.IStandaloneModel
+	backgroundFunc func() error
+}
+
+func newModelTask(name string, obj db.IStandaloneModel, f func() error) appsrv.IWorkerTask {
+	return &modelTask{
+		name:           name,
+		object:         obj,
+		backgroundFunc: f,
+	}
+}
+
+func (t *modelTask) getObjectDesc() string {
+	return fmt.Sprintf("%s(%s)", t.object.GetName(), t.object.GetId())
+}
+
+func (t *modelTask) Run() {
+	if err := t.backgroundFunc(); err != nil {
+		log.Errorf("execute %s for model %s: %v", t.name, t.getObjectDesc(), err)
+	}
+}
+
+func (t *modelTask) Dump() string {
+	return fmt.Sprintf("Task %s for model %s", t.name, t.getObjectDesc())
+}
+
+func RunModelTask(name string, obj db.IStandaloneModel, f func() error) {
+	GetModelTaskManager().Run(newModelTask(name, obj, f), nil, nil)
+}

pkg/monitor/service/service.go

@@ -40,7 +40,6 @@ import (
 	"yunion.io/x/onecloud/pkg/monitor/options"
 	"yunion.io/x/onecloud/pkg/monitor/registry"
 	"yunion.io/x/onecloud/pkg/monitor/subscriptionmodel"
-	_ "yunion.io/x/onecloud/pkg/monitor/tasks"
 	_ "yunion.io/x/onecloud/pkg/monitor/tsdb/driver/influxdb"
 	"yunion.io/x/onecloud/pkg/monitor/worker"
 )

pkg/monitor/tasks/delete_alertrecord_task.go

@@ -1,62 +0,0 @@
-// Copyright 2019 Yunion
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package tasks
-
-import (
-	"context"
-	"fmt"
-
-	"yunion.io/x/jsonutils"
-	"yunion.io/x/pkg/errors"
-
-	"yunion.io/x/onecloud/pkg/cloudcommon/db"
-	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
-	"yunion.io/x/onecloud/pkg/monitor/models"
-	"yunion.io/x/onecloud/pkg/util/logclient"
-)
-
-type DeleteAlertRecordTask struct {
-	taskman.STask
-}
-
-func init() {
-	taskman.RegisterTask(&DeleteAlertRecordTask{})
-}
-
-func (self *DeleteAlertRecordTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) {
-	alert := obj.(models.ICommonAlert)
-	errs := alert.DeleteAttachAlertRecords(ctx, self.GetUserCred())
-	if len(errs) != 0 {
-		msg := jsonutils.NewString(fmt.Sprintf("fail to DeleteAttachAlertRecords:%s.err:%v", alert.GetName(), errors.NewAggregate(errs)))
-		self.taskFail(ctx, alert, msg)
-		return
-	}
-
-	err := alert.RealDelete(ctx, self.UserCred)
-	if err != nil {
-		msg := fmt.Sprintf("delete SCommonAlert err:%v", err)
-		self.taskFail(ctx, alert, jsonutils.NewString(msg))
-		return
-	}
-	db.OpsLog.LogEvent(alert, db.ACT_DELETE, nil, self.GetUserCred())
-	logclient.AddActionLogWithStartable(self, alert, logclient.ACT_DELETE, nil, self.UserCred, true)
-	self.SetStageComplete(ctx, nil)
-}
-
-func (self *DeleteAlertRecordTask) taskFail(ctx context.Context, alert models.ICommonAlert, msg jsonutils.JSONObject) {
-	db.OpsLog.LogEvent(alert, db.ACT_DELETE_FAIL, msg, self.GetUserCred())
-	logclient.AddActionLogWithStartable(self, alert, logclient.ACT_DELETE, msg, self.UserCred, false)
-	self.SetStageFailed(ctx, msg)
-}

pkg/monitor/tasks/detach_alertresource_task.go

@@ -1,67 +0,0 @@
-// Copyright 2019 Yunion
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package tasks
-
-import (
-	"context"
-	"fmt"
-
-	"yunion.io/x/jsonutils"
-	"yunion.io/x/log"
-	"yunion.io/x/pkg/errors"
-
-	"yunion.io/x/onecloud/pkg/apis/monitor"
-	"yunion.io/x/onecloud/pkg/cloudcommon/db"
-	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
-	"yunion.io/x/onecloud/pkg/monitor/models"
-	"yunion.io/x/onecloud/pkg/util/logclient"
-)
-
-type DetachAlertResourceTask struct {
-	taskman.STask
-}
-
-func init() {
-	taskman.RegisterTask(&DetachAlertResourceTask{})
-}
-
-func (self *DetachAlertResourceTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) {
-	alert := obj.(models.ICommonAlert)
-	errs := alert.DetachAlertResourceOnDisable(ctx, self.GetUserCred())
-	if len(errs) != 0 {
-		msg := jsonutils.NewString(fmt.Sprintf("fail to DetachAlertResourceOnAlertDisable:%s.err:%v", alert.GetName(), errors.NewAggregate(errs)))
-		self.taskFail(ctx, alert, msg)
-		return
-	}
-	// err := models.GetAlertResourceManager().NotifyAlertResourceCount(ctx)
-	// if err != nil {
-	// 	log.Errorf("DetachAlertResourceTask NotifyAlertResourceCount error:%v", err)
-	// }
-	// detach MonitorResourceJoint when alert disabel
-	err := models.MonitorResourceAlertManager.DetachJoint(ctx, self.GetUserCred(),
-		monitor.MonitorResourceJointListInput{AlertId: alert.GetId()})
-	if err != nil {
-		log.Errorf("DetachJoint when alert:%s disable err:%v", alert.GetName(), err)
-	}
-	logclient.AddActionLogWithStartable(self, alert, logclient.ACT_DETACH_ALERTRESOURCE, nil, self.UserCred, true)
-	self.SetStageComplete(ctx, nil)
-}
-
-func (self *DetachAlertResourceTask) taskFail(ctx context.Context, alert models.ICommonAlert, msg jsonutils.JSONObject) {
-	db.OpsLog.LogEvent(alert, db.ACT_DETACH, msg, self.GetUserCred())
-	logclient.AddActionLogWithStartable(self, alert, logclient.ACT_DETACH_ALERTRESOURCE, msg, self.UserCred, false)
-	self.SetStageFailed(ctx, msg)
-	return
-}

pkg/monitor/tasks/detach_monitor_resource_task.go

@@ -1,54 +0,0 @@
-// Copyright 2019 Yunion
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package tasks
-
-import (
-	"context"
-	"fmt"
-
-	"yunion.io/x/jsonutils"
-
-	"yunion.io/x/onecloud/pkg/cloudcommon/db"
-	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
-	"yunion.io/x/onecloud/pkg/monitor/models"
-	"yunion.io/x/onecloud/pkg/util/logclient"
-)
-
-type DetachMonitorResourceJointTask struct {
-	taskman.STask
-}
-
-func init() {
-	taskman.RegisterTask(&DetachMonitorResourceJointTask{})
-}
-
-func (self *DetachMonitorResourceJointTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) {
-	alert := obj.(models.ICommonAlert)
-	err := alert.DetachMonitorResourceJoint(ctx, self.GetUserCred())
-	if err != nil {
-		msg := jsonutils.NewString(fmt.Sprintf("alert:%s DetachMonitorResourceJoint err:%v", alert.GetName(), err))
-		self.taskFail(ctx, alert, msg)
-		return
-	}
-	logclient.AddActionLogWithStartable(self, alert, logclient.ACT_DETACH_MONITOR_RESOURCE_JOINT, nil, self.UserCred, true)
-	self.SetStageComplete(ctx, nil)
-}
-
-func (self *DetachMonitorResourceJointTask) taskFail(ctx context.Context, alert models.ICommonAlert, msg jsonutils.JSONObject) {
-	db.OpsLog.LogEvent(alert, db.ACT_DETACH_MONITOR_RESOURCE_JOINT, msg, self.GetUserCred())
-	logclient.AddActionLogWithStartable(self, alert, logclient.ACT_DETACH_MONITOR_RESOURCE_JOINT, msg, self.UserCred, false)
-	self.SetStageFailed(ctx, msg)
-	return
-}

pkg/monitor/tasks/doc.go

@@ -1,15 +0,0 @@
-// Copyright 2019 Yunion
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package tasks // import "yunion.io/x/onecloud/pkg/monitor/tasks"

pkg/monitor/tasks/update_monitor_resource_task.go

@@ -1,53 +0,0 @@
-// Copyright 2019 Yunion
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package tasks
-
-import (
-	"context"
-	"fmt"
-
-	"yunion.io/x/jsonutils"
-
-	"yunion.io/x/onecloud/pkg/cloudcommon/db"
-	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
-	"yunion.io/x/onecloud/pkg/monitor/models"
-	// "yunion.io/x/onecloud/pkg/util/logclient"
-)
-
-type UpdateMonitorResourceJointTask struct {
-	taskman.STask
-}
-
-func init() {
-	taskman.RegisterTask(&UpdateMonitorResourceJointTask{})
-}
-
-func (self *UpdateMonitorResourceJointTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) {
-	alert := obj.(models.ICommonAlert)
-	err := alert.UpdateMonitorResourceJoint(ctx, self.GetUserCred())
-	if err != nil {
-		msg := jsonutils.NewString(fmt.Sprintf("alert:%s UpdateMonitorResourceJoint err:%v", alert.GetName(), err))
-		self.taskFail(ctx, alert, msg)
-		return
-	}
-	// logclient.AddActionLogWithStartable(self, alert, logclient.ACT_UPDATE_MONITOR_RESOURCE_JOINT, nil, self.UserCred, true)
-	self.SetStageComplete(ctx, nil)
-}
-
-func (self *UpdateMonitorResourceJointTask) taskFail(ctx context.Context, alert models.ICommonAlert, msg jsonutils.JSONObject) {
-	// db.OpsLog.LogEvent(alert, db.ACT_UPDATE_MONITOR_RESOURCE_JOINT, msg, self.GetUserCred())
-	// logclient.AddActionLogWithStartable(self, alert, logclient.ACT_UPDATE_MONITOR_RESOURCE_JOINT, msg, self.UserCred, false)
-	self.SetStageFailed(ctx, msg)
-}