Merge pull request #4232 from rainzm/bugfix/notify_data_init

Fix(notify): Unified the api params and fix some bug.

cmd/climc/shell/contacts.go

@@ -50,15 +50,13 @@ func init() {
 
 		arr.Add(tmpObj)
 
-		tmpObj = jsonutils.NewDict()
-		tmpObj.Add(arr, "contacts")
 		params := jsonutils.NewDict()
-		params.Add(tmpObj, "contact")
+		params.Add(arr, "contacts")
 		if args.UpdateDingtalk {
 			params.Add(jsonutils.JSONTrue, "update_dingtalk")
 		}
 
-		contact, err := modules.Contacts.PerformAction(s, args.UID, "update-contact", params)
+		contact, err := modules.Contacts.CustomizedPerformAction(s, args.UID, "update-contact", params)
 
 		if err != nil {
 			return err
@@ -80,7 +78,7 @@ func init() {
 		arr.Add(tmpObj)
 		params := jsonutils.NewDict()
 		params.Add(arr, "contacts")
-		contact, err := modules.Contacts.PerformAction(s, args.UID, "update-contact", params)
+		contact, err := modules.Contacts.CustomizedPerformAction(s, args.UID, "update-contact", params)
 		if err != nil {
 			return err
 		}
@@ -135,7 +133,7 @@ func init() {
 
 		params.Add(jsonutils.JSONTrue, "details")
 
-		result, err := modules.Contacts.Get(s, args.UID, params)
+		result, err := modules.Contacts.CustomizedGet(s, args.UID, params)
 		if err != nil {
 			return err
 		}
@@ -168,12 +166,10 @@ func init() {
 		CONTACT      string `help:"The contacts details mobile number or email address"`
 	}
 	R(&ContactsVerifyOptions{}, "contact-verify-trigger", "Trigger contact verify", func(s *mcclient.ClientSession, args *ContactsVerifyOptions) error {
-		params := jsonutils.NewDict()
 		tmpDict := jsonutils.NewDict()
 		tmpDict.Add(jsonutils.NewString(args.CONTACT_TYPE), "contact_type")
 		tmpDict.Add(jsonutils.NewString(args.CONTACT), "contact")
-		params.Add(tmpDict, "contact")
-		_, err := modules.Contacts.PerformAction(s, args.UID, "verify", params)
+		_, err := modules.Contacts.CustomizedPerformAction(s, args.UID, "verify", tmpDict)
 		if err != nil {
 			return err
 		}

cmd/climc/shell/notification.go

@@ -29,22 +29,26 @@ func init() {
 	 */
 
 	type NotificationCreateOptions struct {
-		Uid         []string `help:"The user you wanna sent to (Keystone User ID)"`
-		CONTACTTYPE string   `help:"User's contacts type" choices:"email|mobile|dingtalk|webconsole"`
-		TOPIC       string   `help:"Title or topic of the notification"`
-		PRIORITY    string   `help:"Priority of the notification" choices:"normal|important|fatal"`
-		MSG         string   `help:"The content of the notification"`
-		Remark      string   `help:"Remark or description of the notification"`
-		Group       bool     `help:"Send to group"`
+		CONTACTTYPE string `help:"User's contacts type" choices:"email|mobile|dingtalk|webconsole"`
+		TOPIC       string `help:"Title or topic of the notification"`
+		PRIORITY    string `help:"Priority of the notification" choices:"normal|important|fatal"`
+		MSG         string `help:"The content of the notification"`
+		Remark      string `help:"Remark or description of the notification"`
+		Group       bool   `help:"Send to group"`
 	}
-	R(&NotificationCreateOptions{}, "notify", "Send a notification to someones", func(s *mcclient.ClientSession,
-		args *NotificationCreateOptions) error {
+	type NotificationCreateSingleOptions struct {
+		UID string `help:"The user you wanna sent to (Keystone User ID)"`
+		NotificationCreateOptions
+	}
+
+	R(&NotificationCreateSingleOptions{}, "notify", "Send a notification to someones", func(s *mcclient.ClientSession,
+		args *NotificationCreateSingleOptions) error {
 
 		msg := notify.SNotifyMessage{}
 		if args.Group {
-			msg.Gid = args.Uid
+			msg.Gid = []string{args.UID}
 		} else {
-			msg.Uid = args.Uid
+			msg.Uid = []string{args.UID}
 		}
 
 		msg.ContactType = notify.TNotifyChannel(args.CONTACTTYPE)
@@ -59,6 +63,35 @@ func init() {
 		}
 		return nil
 	})
+
+	type NotificationCreateMultiOptions struct {
+		Uid []string `help:"The user you wanna sent to (Keystone User ID)"`
+		NotificationCreateOptions
+	}
+
+	R(&NotificationCreateMultiOptions{}, "notify-batch", "Send a notification to someones",
+		func(s *mcclient.ClientSession, args *NotificationCreateMultiOptions) error {
+
+			msg := notify.SNotifyMessage{}
+			if args.Group {
+				msg.Gid = args.Uid
+			} else {
+				msg.Uid = args.Uid
+			}
+
+			msg.ContactType = notify.TNotifyChannel(args.CONTACTTYPE)
+			msg.Topic = args.TOPIC
+			msg.Priority = notify.TNotifyPriority(args.PRIORITY)
+			msg.Msg = args.MSG
+			msg.Remark = args.Remark
+
+			err := notify.Notifications.Send(s, msg)
+			if err != nil {
+				return err
+			}
+			return nil
+		})
+
 	/**
 	 * 发送全局通知
 	 */
@@ -74,7 +107,7 @@ func init() {
 	R(&NotificationBroadcastOptions{}, "notify-broadcast", "Send a notification to all online users", func(s *mcclient.ClientSession, args *NotificationBroadcastOptions) error {
 		msg := notify.SNotifyMessage{}
 		msg.Broadcast = true
-		msg.ContactType = notify.TNotifyChannel("webconsole")
+		msg.ContactType = notify.NotifyByWebConsole
 		msg.Topic = args.Topic
 		msg.Priority = notify.TNotifyPriority(args.Priority)
 		msg.Msg = args.Msg

cmd/climc/shell/notify_template.go

@@ -25,17 +25,17 @@ import (
 func init() {
 	type NotifyTemplateUpdateOptions struct {
 		CONTACTTYPE  string `help:"the contanct type, such as 'email', 'mobile'"`
-		Topic        string `help:"the topic of temlate, such as 'VERIFY', 'ALARM'"`
-		TemplateType string `help:"the type of template" choices:"content|title|remote"`
-		Content      string `help:"the content of template"`
+		TOPIC        string `help:"the topic of temlate, such as 'VERIFY', 'ALARM'"`
+		TEMPLATETYPE string `help:"the type of template" choices:"content|title|remote"`
+		CONTENT      string `help:"the content of template"`
 	}
 	R(&NotifyTemplateUpdateOptions{}, "notify-template-update", "Create, update contact for user", func(s *mcclient.ClientSession,
 		args *NotifyTemplateUpdateOptions) error {
 		arr := jsonutils.NewArray()
 		tmpObj := jsonutils.NewDict()
-		tmpObj.Add(jsonutils.NewString(args.Topic), "topic")
-		tmpObj.Add(jsonutils.NewString(args.TemplateType), "template_type")
-		tmpObj.Add(jsonutils.NewString(args.Content), "content")
+		tmpObj.Add(jsonutils.NewString(args.TOPIC), "topic")
+		tmpObj.Add(jsonutils.NewString(args.TEMPLATETYPE), "template_type")
+		tmpObj.Add(jsonutils.NewString(args.CONTENT), "content")
 		arr.Add(tmpObj)
 
 		params := jsonutils.NewDict()

pkg/mcclient/modules/mod_contacts.go

@@ -45,14 +45,19 @@ func (this *ContactsManager) DoBatchDeleteContacts(s *mcclient.ClientSession, pa
 	return modulebase.Post(this.ResourceManager, s, path, params, this.Keyword)
 }
 
-func (this *ContactsManager) PerformAction(session *mcclient.ClientSession, id string, action string,
+func (this *ContactsManager) CustomizedPerformAction(session *mcclient.ClientSession, id string, action string,
 	params jsonutils.JSONObject) (jsonutils.JSONObject, error) {
 
+	body := jsonutils.NewDict()
+	if params != nil {
+		body.Add(params, this.Keyword)
+	}
 	path := fmt.Sprintf("/%s/%s/%s?uname=true", this.ContextPath(nil), url.PathEscape(id), url.PathEscape(action))
 	return modulebase.Post(this.ResourceManager, session, path, params, this.KeywordPlural)
 }
 
-func (this *ContactsManager) Get(session *mcclient.ClientSession, id string, params jsonutils.JSONObject) (jsonutils.JSONObject,
+func (this *ContactsManager) CustomizedGet(session *mcclient.ClientSession, id string,
+	params jsonutils.JSONObject) (jsonutils.JSONObject,
 	error) {
 
 	q := params.(*jsonutils.JSONDict)

pkg/mcclient/modules/notify/mod_notification.go

@@ -15,6 +15,8 @@
 package notify
 
 import (
+	"fmt"
+
 	"yunion.io/x/jsonutils"
 
 	"yunion.io/x/onecloud/pkg/mcclient"
@@ -42,7 +44,12 @@ type NotificationManager struct {
 }
 
 func (manager *NotificationManager) Send(s *mcclient.ClientSession, msg SNotifyMessage) error {
-	_, err := manager.Create(s, jsonutils.Marshal(&msg))
+	params := jsonutils.Marshal(&msg)
+	body := jsonutils.NewDict()
+	body.Add(params, manager.Keyword)
+
+	path := fmt.Sprintf("/%s?uname=true", manager.ContextPath(nil))
+	_, err := modulebase.Post(manager.ResourceManager, s, path, body, manager.KeywordPlural)
 	return err
 }
 

pkg/notify/handlers.go

@@ -187,8 +187,8 @@ func templateUpdateHandler(ctx context.Context, w http.ResponseWriter, r *http.R
 	manager, params, query, body := fetchEnv(ctx, w, r)
 	data, err := body.GetArray(manager.Keyword(), manager.KeywordPlural())
 	if err != nil {
-		httperrors.GeneralServerError(w, httperrors.NewInputParameterError("need %s or %s", manager.Keyword(),
-			manager.KeywordPlural()))
+		httperrors.GeneralServerError(w, httperrors.NewInputParameterError("need %s and %s",
+			manager.Keyword(), manager.KeywordPlural()))
 		return
 	}
 	ctype := params["<ctype>"]
@@ -282,8 +282,8 @@ func contactUpdateHandler(ctx context.Context, w http.ResponseWriter, r *http.Re
 	data, err := body.GetArray(manager.Keyword(), manager.KeywordPlural())
 	if err != nil {
 		log.Errorf("body: %s, err: %s\n", body.String(), err)
-		httperrors.GeneralServerError(w, httperrors.NewInputParameterError("need %s or %s", manager.Keyword(),
-			manager.KeywordPlural()))
+		httperrors.GeneralServerError(w, httperrors.NewInputParameterError("need %s and %s",
+			manager.Keyword(), manager.KeywordPlural()))
 		return
 	}
 
@@ -329,7 +329,7 @@ func verifyTriggerHandler(ctx context.Context, w http.ResponseWriter, r *http.Re
 	manager, params, _, body := fetchEnv(ctx, w, r)
 	data, err := body.Get(models.ContactManager.Keyword())
 	if err != nil {
-		httperrors.BadRequestError(w, "request body should have %s", manager.KeywordPlural())
+		httperrors.BadRequestError(w, "request body should have %s", manager.Keyword())
 		return
 	}
 	ret, err := manager.VerifyTrigger(ctx, params, data)

pkg/notify/models/mod_config.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 
 	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/sqlchemy"
 
@@ -52,7 +53,7 @@ func init() {
 type SConfig struct {
 	SStatusStandaloneResourceBase
 
-	Type      string `width:"15" nullable:"false" create:"required" list:"user""`
+	Type      string `width:"15" nullable:"false" create:"required" list:"user"`
 	KeyText   string `width:"30" nullable:"false" create:"required" list:"user"`
 	ValueText string `width:"100" nullable:"false" create:"required" list:"user"`
 }
@@ -84,23 +85,35 @@ func (self *SConfigManager) GetValue(key, contactType string) (*SConfig, error)
 }
 
 func (self *SConfigManager) InitializeData() error {
-	sql := fmt.Sprintf("update %s set updated_at=gmt_modified, deleted=is_deleted, created_at=gmt_create, deleted_at=gmt_deleted, update_by=modified_by, delete_by=deleted_by", self.TableSpec().Name())
-	q := sqlchemy.NewRawQuery(sql, "")
-	q.Row()
-	sql = fmt.Sprintf("update %s set type='mobile' where type='sms_aliyun'", self.TableSpec().Name())
-	q = sqlchemy.NewRawQuery(sql, "")
-	q.Row()
+	q := self.Query()
+	q = q.Filter(sqlchemy.OR(sqlchemy.IsNotNull(q.Field("updated_at")), sqlchemy.IsNotNull(q.Field("created_at")),
+		sqlchemy.IsNotNull(q.Field("deleted_at")), sqlchemy.IsTrue(q.Field("deleted")),
+		sqlchemy.IsNotNull(q.Field("update_by")), sqlchemy.IsNotNull(q.Field("delete_by"))))
+	n, err := q.CountWithError()
+	if err != nil {
+		return err
+	}
+	if n <= 0 {
+		log.Debugf("need to init data for %s", self.TableSpec().Name())
+		// need to update
+		sql := fmt.Sprintf("update %s set updated_at=gmt_modified, deleted=is_deleted, created_at=gmt_create, deleted_at=gmt_deleted, update_by=modified_by, delete_by=deleted_by", self.TableSpec().Name())
+		q := sqlchemy.NewRawQuery(sql, "")
+		q.Row()
+		sql = fmt.Sprintf("update %s set type='mobile' where type='sms_aliyun'", self.TableSpec().Name())
+		q = sqlchemy.NewRawQuery(sql, "")
+		q.Row()
+	}
 
 	// init webconsole's config
 	q = self.Query().Equals("type", "webconsole")
-	n, err := q.CountWithError()
+	n, err = q.CountWithError()
 	if err != nil {
 		return err
 	}
 	if n >= 4 {
 		return nil
 	}
-	sql = fmt.Sprintf("update %s set deleted='1' where type='webconsole'", self.TableSpec().Name())
+	sql := fmt.Sprintf("update %s set deleted='1' where type='webconsole'", self.TableSpec().Name())
 	q = sqlchemy.NewRawQuery(sql)
 	q.Row()
 	configs := []SConfig{

pkg/notify/models/mod_contact.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/util/sets"
 	"yunion.io/x/sqlchemy"
@@ -94,8 +95,20 @@ func (self *SContactManager) FilterByOwner(q *sqlchemy.SQuery, owner mcclient.II
 }
 
 func (self *SContactManager) InitializeData() error {
+	q := self.Query()
+	q = q.Filter(sqlchemy.OR(sqlchemy.IsNotNull(q.Field("updated_at")), sqlchemy.IsTrue(q.Field("deleted"))))
+	n, err := q.CountWithError()
+	if err != nil {
+		return err
+	}
+	if n > 0 {
+		log.Debugf("no need to init data for %s", self.TableSpec().Name())
+		// no need to init data
+		return nil
+	}
+	log.Debugf("need to init data for %s", self.TableSpec().Name())
 	sql := fmt.Sprintf("update %s set updated_at=update_at, deleted=is_deleted", self.TableSpec().Name())
-	q := sqlchemy.NewRawQuery(sql, "")
+	q = sqlchemy.NewRawQuery(sql, "")
 	q.Row()
 	return nil
 }

pkg/notify/models/mod_notification.go

@@ -19,7 +19,6 @@ import (
 	"context"
 	"fmt"
 	"strings"
-	"sync"
 	"time"
 
 	"yunion.io/x/jsonutils"
@@ -435,11 +434,17 @@ func (self *SNotification) SetStatusWithoutUserCred(status string) error {
 	return nil
 }
 func sendWithoutUserCred(notifications []SNotification) {
-	var wg sync.WaitGroup
+	// limit the number of Concurrency
+	Max := 10
+	limit := make(chan struct{}, Max)
 	sendone := func(notification SNotification) {
+		defer func() {
+			<-limit
+		}()
 		// Get contact
 		contact, err := ContactManager.FetchByUIDAndCType(notification.UID, []string{notification.ContactType})
 		if err != nil {
+			log.Debugf("fail to fetch contacts with uid '%s' in ReSend Cron Job", notification.UID)
 			return
 		}
 		if len(contact) == 0 {
@@ -454,18 +459,20 @@ func sendWithoutUserCred(notifications []SNotification) {
 			return
 		}
 		if err != nil {
-			log.Errorf("Send notification failed because that %s.", err.Error())
+			log.Errorf("Send notification failed in ReSend Cron Job: %s.", err.Error())
 			notification.SetStatusWithoutUserCred(NOTIFY_FAIL)
 		} else {
 			notification.SetStatusWithoutUserCred(NOTIFY_OK)
 		}
-		wg.Done()
 	}
 	for i := range notifications {
-		wg.Add(1)
+		limit <- struct{}{}
 		go sendone(notifications[i])
 	}
-	wg.Wait()
+	// wait all finish
+	for i := 0; i < Max; i++ {
+		limit <- struct{}{}
+	}
 }
 
 func ReSend(seconds int) {

pkg/notify/models/mod_template.go

@@ -74,6 +74,7 @@ func (tm *STemplateManager) GetEmailUrl() string {
 }
 
 func (tm *STemplateManager) InitializeData() error {
+	// init email VERIFY template
 	q := tm.Query().Equals("contact_type", "email").Equals("topic", "VERIFY").Equals("template_type", "content")
 	count, _ := q.CountWithError()
 	if count > 0 {
@@ -89,13 +90,13 @@ func (tm *STemplateManager) InitializeData() error {
 	contentTem := STemplate{
 		ContactType:  "email",
 		Topic:        "VERIFY",
-		TemplateType: "content",
+		TemplateType: TEMPLATE_TYPE_CONTENT,
 		Content:      string(content),
 	}
 	titleTem := STemplate{
 		ContactType:  "email",
 		Topic:        "VERIFY",
-		TemplateType: "title",
+		TemplateType: TEMPLATE_TYPE_TITLE,
 		Content:      template.EMAIL_VERIFY_TITLE,
 	}
 	err = tm.TableSpec().InsertOrUpdate(&contentTem)
@@ -106,6 +107,32 @@ func (tm *STemplateManager) InitializeData() error {
 	if err != nil {
 		return errors.Wrap(err, "sqlchemy.TableSpec.InsertOrUpdate")
 	}
+	// init email IMAGE_ACTIVED template
+	q = tm.Query().Equals("contact_type", "email").Equals("topic", "IMAGE_ACTIVED")
+	count, _ = q.CountWithError()
+	if count > 1 {
+		return nil
+	}
+	titleTem = STemplate{
+		ContactType:  "email",
+		Topic:        "IMAGE_ACTIVED",
+		TemplateType: TEMPLATE_TYPE_TITLE,
+		Content:      template.IMAGE_ACTIVED_TITLE,
+	}
+	contentTem = STemplate{
+		ContactType:  "email",
+		Topic:        "IMAGE_ACTIVED",
+		TemplateType: TEMPLATE_TYPE_CONTENT,
+		Content:      template.IMAGE_ACTIVED_CONTENT,
+	}
+	err = tm.TableSpec().InsertOrUpdate(&titleTem)
+	if err != nil {
+		return errors.Wrapf(err, "fail to InsertOrUpdate for email IMAGE_ACTIVED title template")
+	}
+	err = tm.TableSpec().InsertOrUpdate(&contentTem)
+	if err != nil {
+		return errors.Wrapf(err, "fail to InsertOrUpdate for email IMAGE_ACTIVED content template")
+	}
 	return nil
 }
 

pkg/notify/models/mod_verify.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
 	"yunion.io/x/sqlchemy"
 
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
@@ -78,8 +79,20 @@ func NewSVerify(contactType string, cid string) *SVerify {
 }
 
 func (self *SVerifyManager) InitializeData() error {
+	q := self.Query()
+	q = q.Filter(sqlchemy.OR(sqlchemy.IsNotNull(q.Field("updated_at")), sqlchemy.IsTrue(q.Field("deleted"))))
+	n, err := q.CountWithError()
+	if err != nil {
+		return err
+	}
+	if n > 0 {
+		log.Debugf("no need to init data for %s", self.TableSpec().Name())
+		// no need to init data
+		return nil
+	}
+	log.Debugf("need to init data for %s", self.TableSpec().Name())
 	sql := fmt.Sprintf("update %s set updated_at=update_at, deleted=is_deleted", self.TableSpec().Name())
-	q := sqlchemy.NewRawQuery(sql, "")
+	q = sqlchemy.NewRawQuery(sql, "")
 	q.Row()
 	return nil
 }

pkg/notify/options/options.go

@@ -24,7 +24,7 @@ type NotifyOption struct {
 
 	DingtalkEnabled bool   `help:"Enable dingtalk"`
 	SocketFileDir   string `help:"Socket file directory" default:"/etc/yunion/notify"`
-	UpdateInterval  int    `help:"Update send services interval(unit:s)" default:"30"`
+	UpdateInterval  int    `help:"Update send services interval(unit:min)" default:"30"`
 	VerifyEmailUrl  string `help:"url of verify email"`
 	ReSendScope     int    `help:"Resend all messages that have not been sent successfully within ReSendScope
 seconds" default:"30"`

pkg/notify/service.go

@@ -66,9 +66,9 @@ func StartService() {
 
 	cron := cronman.InitCronJobManager(true, 2)
 	// update service
-	cron.AddJobAtIntervals("UpdateServices", time.Duration(opts.UpdateInterval)*time.Second, models.NotifyService.UpdateServices)
+	cron.AddJobAtIntervals("UpdateServices", time.Duration(opts.UpdateInterval)*time.Minute, models.NotifyService.UpdateServices)
 
-	// resend notifications
+	// wrapped func to resend notifications
 	resend := func(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) {
 		models.ReSend(opts.ReSendScope)
 	}

pkg/notify/template/template.go

@@ -17,4 +17,7 @@ package template
 const (
 	EMAIL_VERIFY_CONTENT_PATH = "/opt/yunion/share/notify/email_verify_template"
 	EMAIL_VERIFY_TITLE        = "Yunion Verify"
+
+	IMAGE_ACTIVED_TITLE   = "镜像 {{.name}} 上传完成"
+	IMAGE_ACTIVED_CONTENT = "{{.os_type}} 镜像 {{.name}} 上传完成"
 )