diff --git a/cmd/climc/shell/contacts.go b/cmd/climc/shell/contacts.go index c7c957be2e..8f37c44645 100644 --- a/cmd/climc/shell/contacts.go +++ b/cmd/climc/shell/contacts.go @@ -29,10 +29,11 @@ func init() { * 操作用户的通信地址（如果用户的通信地址不存在则进行添加；如果已存在则进行修改；如果设置空则进行删除。） */ type ContactsUpdateOptions struct { - UID string `help:"The user you wanna add contact to (Keystone User ID)"` - CONTACTTYPE string `help:"The contact type email|mobile" choices:"email|mobile|dingtalk"` - CONTACT string `help:"The contacts details mobile number or email address or dingtalk's userid, if set it the empty str means delete"` - Status string `help:"Enabled or disabled contact status" choices:"enable|disable"` + UID string `help:"The user you wanna add contact to (Keystone User ID)"` + CONTACTTYPE string `help:"The contact type email|mobile" choices:"email|mobile|dingtalk"` + CONTACT string `help:"The contacts details mobile number or email address or dingtalk's userid, if set it the empty str means delete"` + Status string `help:"Enabled or disabled contact status" choices:"enable|disable"` + UpdateDingtalk bool `help:"if update dingtalk"` } R(&ContactsUpdateOptions{}, "contact-update", "Create, delete or update contact for user", func(s *mcclient.ClientSession, args *ContactsUpdateOptions) error { arr := jsonutils.NewArray() @@ -51,6 +52,9 @@ func init() { params := jsonutils.NewDict() params.Add(arr, "contacts") + if args.UpdateDingtalk { + params.Add(jsonutils.JSONTrue, "update_dingtalk") + } contact, err := modules.Contacts.PerformAction(s, args.UID, "update-contact", params) @@ -98,6 +102,7 @@ func init() { } } + params.Add(jsonutils.JSONTrue, "details") result, err := modules.Contacts.List(s, params) if err != nil { @@ -126,6 +131,8 @@ func init() { } } + params.Add(jsonutils.JSONTrue, "details") + result, err := modules.Contacts.Get(s, args.UID, params) if err != nil { return err @@ -133,7 +140,7 @@ func init() { contactsStr, err := result.GetString("details") if err != nil { - return err + return nil } contactsJson, err := jsonutils.ParseString(contactsStr) diff --git a/cmd/climc/shell/notification.go b/cmd/climc/shell/notification.go index 3974c8649c..4803a25892 100644 --- a/cmd/climc/shell/notification.go +++ b/cmd/climc/shell/notification.go @@ -29,20 +29,20 @@ 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"` + 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"` } R(&NotificationCreateOptions{}, "notify", "Send a notification to sb", func(s *mcclient.ClientSession, args *NotificationCreateOptions) error { msg := notify.SNotifyMessage{} if args.Group { - msg.Gid = args.UID + msg.Gid = args.Uid } else { - msg.Uid = args.UID + msg.Uid = args.Uid } msg.ContactType = notify.TNotifyChannel(args.CONTACTTYPE) diff --git a/pkg/cloudcommon/db/keystonecache.go b/pkg/cloudcommon/db/keystonecache.go index 40c4d212a4..3535a6f4ed 100644 --- a/pkg/cloudcommon/db/keystonecache.go +++ b/pkg/cloudcommon/db/keystonecache.go @@ -16,6 +16,8 @@ package db import ( "time" + + "yunion.io/x/onecloud/pkg/cloudcommon/consts" ) type SKeystoneCacheObjectManager struct { @@ -43,3 +45,14 @@ func NewKeystoneCacheObject(id string, name string, domainId string, domain stri obj.DomainId = domainId return obj } + +func (t *SKeystoneCacheObject) IsExpired() bool { + if t.LastCheck.IsZero() { + return true + } + now := time.Now().UTC() + if t.LastCheck.Add(consts.GetTenantCacheExpireSeconds()).Before(now) { + return true + } + return false +} diff --git a/pkg/cloudcommon/db/tenantcache.go b/pkg/cloudcommon/db/tenantcache.go index 07c7e48934..79ba0c2e6f 100644 --- a/pkg/cloudcommon/db/tenantcache.go +++ b/pkg/cloudcommon/db/tenantcache.go @@ -154,7 +154,8 @@ func (manager *STenantCacheManager) FetchTenantById(ctx context.Context, idStr s } func (manager *STenantCacheManager) FetchTenantByIdWithoutExpireCheck(ctx context.Context, idStr string) (*STenant, error) { - return manager.fetchTenantById(ctx, idStr, false) + // noExpireCheck should be true + return manager.fetchTenantById(ctx, idStr, true) } func (manager *STenantCacheManager) fetchTenantById(ctx context.Context, idStr string, noExpireCheck bool) (*STenant, error) { diff --git a/pkg/cloudcommon/notifyclient/notify.go b/pkg/cloudcommon/notifyclient/notify.go index 749df4837f..230e05500d 100644 --- a/pkg/cloudcommon/notifyclient/notify.go +++ b/pkg/cloudcommon/notifyclient/notify.go @@ -96,7 +96,8 @@ func getContent(topic string, contType string, channel notify.TNotifyChannel, da return buf.String(), nil } -func Notify(recipientId string, isGroup bool, priority notify.TNotifyPriority, event string, data jsonutils.JSONObject) { +func Notify(recipientId []string, isGroup bool, priority notify.TNotifyPriority, event string, + data jsonutils.JSONObject) { switch priority { case notify.NotifyPriorityCritical: NotifyCritical(recipientId, isGroup, event, data) @@ -107,7 +108,8 @@ func Notify(recipientId string, isGroup bool, priority notify.TNotifyPriority, e } } -func RawNotify(recipientId string, isGroup bool, channel notify.TNotifyChannel, priority notify.TNotifyPriority, event string, data jsonutils.JSONObject) { +func RawNotify(recipientId []string, isGroup bool, channel notify.TNotifyChannel, priority notify.TNotifyPriority, + event string, data jsonutils.JSONObject) { log.Infof("notify %s event %s priority %s", recipientId, event, priority) msg := notify.SNotifyMessage{} if isGroup { @@ -134,7 +136,7 @@ func RawNotify(recipientId string, isGroup bool, channel notify.TNotifyChannel, }, nil, nil) } -func NotifyNormal(recipientId string, isGroup bool, event string, data jsonutils.JSONObject) { +func NotifyNormal(recipientId []string, isGroup bool, event string, data jsonutils.JSONObject) { for _, c := range []notify.TNotifyChannel{ notify.NotifyByEmail, notify.NotifyByDingTalk, @@ -147,7 +149,7 @@ func NotifyNormal(recipientId string, isGroup bool, event string, data jsonutils } } -func NotifyImportant(recipientId string, isGroup bool, event string, data jsonutils.JSONObject) { +func NotifyImportant(recipientId []string, isGroup bool, event string, data jsonutils.JSONObject) { for _, c := range []notify.TNotifyChannel{ notify.NotifyByEmail, notify.NotifyByDingTalk, @@ -161,7 +163,7 @@ func NotifyImportant(recipientId string, isGroup bool, event string, data jsonut } } -func NotifyCritical(recipientId string, isGroup bool, event string, data jsonutils.JSONObject) { +func NotifyCritical(recipientId []string, isGroup bool, event string, data jsonutils.JSONObject) { for _, c := range []notify.TNotifyChannel{ notify.NotifyByEmail, notify.NotifyByDingTalk, @@ -176,12 +178,11 @@ func NotifyCritical(recipientId string, isGroup bool, event string, data jsonuti } func SystemNotify(priority notify.TNotifyPriority, event string, data jsonutils.JSONObject) { - for _, uid := range notifyAdminUsers { - Notify(uid, false, priority, event, data) - } - for _, gid := range notifyAdminGroups { - Notify(gid, true, priority, event, data) - } + // userId + Notify(notifyAdminUsers, false, priority, event, data) + + // groupId + Notify(notifyAdminGroups, true, priority, event, data) } func NotifyGeneralSystemError(data jsonutils.JSONObject) { diff --git a/pkg/cloudcommon/policy/resources.go b/pkg/cloudcommon/policy/resources.go index 483e8312e9..276582f43d 100644 --- a/pkg/cloudcommon/policy/resources.go +++ b/pkg/cloudcommon/policy/resources.go @@ -51,10 +51,11 @@ var ( notifySystemResources = []string{ "configs", - "contacts", } notifyDomainResources = []string{} - notifyUserResources = []string{} + notifyUserResources = []string{ + "contacts", + } meterSystemResources = []string{ "rates", diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index 5dbbb4865f..b32e90f949 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -727,7 +727,7 @@ func (self *SGuest) NotifyServerEvent( } } } - notifyclient.Notify(userCred.GetUserId(), false, priority, event, kwargs) + notifyclient.Notify([]string{userCred.GetUserId()}, false, priority, event, kwargs) if notifyAdmin { notifyclient.SystemNotify(priority, event, kwargs) } diff --git a/pkg/image/tasks/image_convert_task.go b/pkg/image/tasks/image_convert_task.go index c3cb57b50f..0d688f87be 100644 --- a/pkg/image/tasks/image_convert_task.go +++ b/pkg/image/tasks/image_convert_task.go @@ -62,7 +62,7 @@ func (self *ImageConvertTask) OnInit(ctx context.Context, obj db.IStandaloneMode kwargs.Set("os_type", jsonutils.NewString(osType.Value)) } notifyclient.SystemNotify(notify.NotifyPriorityNormal, notifyclient.IMAGE_ACTIVED, kwargs) - notifyclient.NotifyImportant(self.UserCred.GetUserId(), false, notifyclient.IMAGE_ACTIVED, kwargs) + notifyclient.NotifyImportant([]string{self.UserCred.GetUserId()}, false, notifyclient.IMAGE_ACTIVED, kwargs) } return nil, err }) diff --git a/pkg/mcclient/modules/notify/mod_notification.go b/pkg/mcclient/modules/notify/mod_notification.go index f042177aeb..708b314185 100644 --- a/pkg/mcclient/modules/notify/mod_notification.go +++ b/pkg/mcclient/modules/notify/mod_notification.go @@ -27,8 +27,8 @@ var ( ) type SNotifyMessage struct { - Uid string `json:"uid,omitempty"` - Gid string `json:"gid,omitempty"` + Uid []string `json:"uid,omitempty"` + Gid []string `json:"gid,omitempty"` ContactType TNotifyChannel `json:"contact_type,omitempty"` Topic string `json:"topic,omitempty"` Priority TNotifyPriority `json:"priority,omitempty"` diff --git a/pkg/notify/cache/doc.go b/pkg/notify/cache/doc.go new file mode 100644 index 0000000000..7a00422d98 --- /dev/null +++ b/pkg/notify/cache/doc.go @@ -0,0 +1 @@ +package cache // import "yunion.io/x/onecloud/pkg/notify/cache" diff --git a/pkg/notify/cache/user_group_cache.go b/pkg/notify/cache/user_group_cache.go new file mode 100644 index 0000000000..50fdc5eb6d --- /dev/null +++ b/pkg/notify/cache/user_group_cache.go @@ -0,0 +1,181 @@ +// 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 cache + +import ( + "context" + "time" + + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + "yunion.io/x/pkg/util/compare" + + "yunion.io/x/onecloud/pkg/cloudcommon/consts" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/lockman" + "yunion.io/x/onecloud/pkg/mcclient/auth" + "yunion.io/x/onecloud/pkg/mcclient/modules" +) + +type SUserGroupCacheManager struct { + db.SResourceBaseManager +} + +type SUserGroup struct { + db.SResourceBase + UserId string + GroupId string + LastCheck time.Time `nullable:"false"` +} + +func (ug *SUserGroup) GetModelManager() db.IModelManager { + return UserGroupCacheManager +} + +var UserGroupCacheManager *SUserGroupCacheManager + +func init() { + UserGroupCacheManager = &SUserGroupCacheManager{db.NewResourceBaseManager( + SUserGroup{}, + "user_group_cache_tbl", + "usergroup", + "usergroups", + )} +} + +func (ug *SUserGroup) IsExpired() bool { + if ug.LastCheck.IsZero() { + return true + } + now := time.Now().UTC() + if ug.LastCheck.Add(consts.GetTenantCacheExpireSeconds()).Before(now) { + return true + } + return false +} + +func (manager *SUserGroupCacheManager) FetchByGroupId(ctx context.Context, groupId string) ([]SUserGroup, error) { + q := manager.Query().Equals("gourp_id", groupId) + ugs := make([]SUserGroup, 0) + err := db.FetchModelObjects(manager, q, &ugs) + if err != nil { + return nil, err + } + var needSync bool + if len(ugs) == 0 { + needSync = true + } + now := time.Now().UTC() + expireTime := now.Add(-consts.GetTenantCacheExpireSeconds()) + for i := range ugs { + if ugs[i].LastCheck.Before(expireTime) { + needSync = true + break + } + } + if !needSync { + return ugs, nil + } + ugs, syncResult, err := manager.Sync(ctx, ugs, groupId) + if err != nil { + return nil, err + } + if syncResult.IsError() { + log.Errorf(syncResult.Result()) + } + return ugs, nil +} + +func (manager *SUserGroupCacheManager) Sync(ctx context.Context, ugCache []SUserGroup, groupId string) ([]SUserGroup, + compare.SyncResult, error) { + lockman.LockRawObject(ctx, manager.KeywordPlural(), groupId) + defer lockman.ReleaseRawObject(ctx, manager.KeywordPlural(), groupId) + + s := auth.GetAdminSession(ctx, consts.GetRegion(), "v3") + syncResult := compare.SyncResult{} + users, err := modules.Groups.GetUsers(s, groupId) + if err != nil { + return nil, syncResult, errors.Wrap(err, "fetch users by group id from keystone failed") + } + newUgCache := make([]SUserGroup, len(users.Data)) + for i := range users.Data { + userId, _ := users.Data[i].GetString("id") + newUgCache[i] = SUserGroup{ + UserId: userId, + GroupId: groupId, + } + } + added := make([]SUserGroup, 0) + removed := make([]SUserGroup, 0) + commondb := make([]SUserGroup, 0) + compareSets(ugCache, newUgCache, &added, &removed, &commondb) + now := time.Now().UTC() + for i := range added { + added[i].LastCheck = now + err := manager.TableSpec().Insert(&added[i]) + if err != nil { + syncResult.AddError(err) + } else { + syncResult.Add() + } + } + + for i := range commondb { + ug := &commondb[i] + _, err := db.Update(ug, func() error { + ug.LastCheck = now + return nil + }) + if err != nil { + syncResult.UpdateError(err) + } else { + syncResult.Update() + } + } + + for i := range removed { + ug := &removed[i] + _, err := db.Update(ug, func() error { + return ug.MarkDelete() + }) + if err != nil { + syncResult.DeleteError(err) + } else { + syncResult.Delete() + } + } + + return newUgCache, syncResult, nil +} + +func compareSets(dbs, remotes []SUserGroup, added, removed, commondb *[]SUserGroup) { + dbmap := make(map[string]SUserGroup) + for i := range dbs { + dbmap[dbs[i].UserId] = dbs[i] + } + + for i := range remotes { + userId := remotes[i].UserId + if _, ok := dbmap[userId]; ok { + *commondb = append(*commondb, remotes[i]) + } else { + *added = append(*added, remotes[i]) + } + delete(dbmap, userId) + } + for _, v := range dbmap { + *removed = append(*removed, v) + } +} diff --git a/pkg/notify/cache/usercache.go b/pkg/notify/cache/usercache.go new file mode 100644 index 0000000000..90303ca9aa --- /dev/null +++ b/pkg/notify/cache/usercache.go @@ -0,0 +1,219 @@ +// 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 cache + +import ( + "context" + "database/sql" + "fmt" + "time" + + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + + "yunion.io/x/onecloud/pkg/cloudcommon/consts" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/lockman" + "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/mcclient/auth" + "yunion.io/x/onecloud/pkg/mcclient/modules" + "yunion.io/x/onecloud/pkg/util/httputils" +) + +type SUserCacheManager struct { + db.SKeystoneCacheObjectManager +} + +type SUser struct { + db.SKeystoneCacheObject +} + +func (user *SUser) GetModelManager() db.IModelManager { + return UserCacheManager +} + +var UserCacheManager *SUserCacheManager + +func init() { + UserCacheManager = &SUserCacheManager{ + db.NewKeystoneCacheObjectManager(SUser{}, "users_cache_tbl", "user", "users")} + // log.Debugf("initialize user cache manager %s", UserCacheManager.KeywordPlural()) + UserCacheManager.SetVirtualObject(UserCacheManager) +} + +func RegistUserCredCacheUpdater() { + auth.RegisterAuthHook(onAuthCompleteUpdateCache) +} + +func onAuthCompleteUpdateCache(userCred mcclient.TokenCredential) { + UserCacheManager.updateUserCache(userCred) +} + +func (ucm *SUserCacheManager) updateUserCache(userCred mcclient.TokenCredential) { + ucm.Save(context.Background(), userCred.GetUserId(), userCred.GetUserName(), + userCred.GetDomainId()) +} + +func (ucm *SUserCacheManager) FetchUserByIdOrName(idStr string) (*SUser, error) { + obj, err := ucm.SKeystoneCacheObjectManager.FetchByIdOrName(nil, idStr) + if err != nil { + return nil, err + } + return obj.(*SUser), nil +} + +func (ucm *SUserCacheManager) FetchUserById(idStr string) (*SUser, error) { + obj, err := ucm.SKeystoneCacheObjectManager.FetchById(idStr) + if err != nil { + return nil, err + } + return obj.(*SUser), nil +} + +func (ucm *SUserCacheManager) FetchUserByName(idStr string) (*SUser, error) { + obj, err := ucm.SKeystoneCacheObjectManager.FetchByName(nil, idStr) + if err != nil { + return nil, err + } + return obj.(*SUser), nil +} + +func (ucm *SUserCacheManager) Save(ctx context.Context, idStr string, name string, domainId string) (*SUser, error) { + lockman.LockRawObject(ctx, ucm.KeywordPlural(), idStr) + defer lockman.ReleaseRawObject(ctx, ucm.KeywordPlural(), idStr) + + objo, err := ucm.FetchById(idStr) + if err != nil && err != sql.ErrNoRows { + log.Errorf("FetchTenantbyId fail %s", err) + return nil, err + } + now := time.Now().UTC() + if err == nil { + obj := objo.(*SUser) + if obj.Id == idStr && obj.Name == name && obj.DomainId == domainId { + db.Update(obj, func() error { + obj.LastCheck = now + return nil + }) + return obj, nil + } + _, err = db.Update(obj, func() error { + obj.Id = idStr + obj.Name = name + obj.DomainId = domainId + obj.LastCheck = now + return nil + }) + if err != nil { + return nil, err + } else { + return obj, nil + } + } else { + objm, err := db.NewModelObject(ucm) + obj := objm.(*SUser) + obj.Id = idStr + obj.Name = name + obj.DomainId = domainId + obj.LastCheck = now + err = ucm.TableSpec().Insert(obj) + if err != nil { + return nil, err + } else { + return obj, nil + } + } +} + +func (ucm *SUserCacheManager) fetchUserFromKeystone(ctx context.Context, idStr string) (*SUser, error) { + if len(idStr) == 0 { + return nil, fmt.Errorf("Empty idStr") + } + s := auth.GetAdminSession(ctx, consts.GetRegion(), "v3") + user, err := modules.UsersV3.GetById(s, idStr, nil) + if err != nil { + if je, ok := err.(*httputils.JSONClientError); ok && je.Code == 404 { + return nil, sql.ErrNoRows + } + log.Errorf("fetch project %s fail %s", idStr, err) + return nil, errors.Wrap(err, "modules.Projects.Get") + } + userId, _ := user.GetString("id") + userName, _ := user.GetString("name") + domainId, _ := user.GetString("domain_id") + return ucm.Save(ctx, userId, userName, domainId) +} + +func (ucm *SUserCacheManager) FetchUsersByIDs(ctx context.Context, ids []string) (map[string]SUser, error) { + q := ucm.Query().In("id", ids) + users := make([]SUser, 0) + err := db.FetchModelObjects(ucm, q, &users) + if err != nil { + return nil, err + } + ret := make(map[string]SUser) + + for i := range users { + ret[users[i].Id] = users[i] + } + + // check that id is exist + for _, id := range ids { + if _, ok := ret[id]; ok { + continue + } + user, err := ucm.fetchUserFromKeystone(ctx, id) + if err != nil { + continue + } + ret[id] = *user + } + return ret, nil +} + +func (ucm *SUserCacheManager) FetchUserByID(ctx context.Context, idStr string, noExpireCheck bool) (*SUser, error) { + + q := ucm.Query().Equals("id", idStr) + uobj, err := db.NewModelObject(ucm) + if err != nil { + return nil, errors.Wrap(err, "NewModelObject") + } + err = q.First(uobj) + if err != nil && err != sql.ErrNoRows { + return nil, errors.Wrap(err, "query") + } else if uobj != nil { + user := uobj.(*SUser) + if noExpireCheck || !user.IsExpired() { + return user, nil + } + } + return ucm.fetchUserFromKeystone(ctx, idStr) +} + +func (ucm *SUserCacheManager) FetchUserLikeName(ctx context.Context, name string, noExpireCheck bool) ([]SUser, + error) { + + if !noExpireCheck { + // todo + return nil, fmt.Errorf("FetchUserLikeName with check Not Implement") + } + q := ucm.Query().Like("name", "%"+name+"%") + users := make([]SUser, 0) + err := db.FetchModelObjects(ucm, q, &users) + if err != nil { + return nil, err + } + return users, nil +} diff --git a/pkg/notify/dispatcher.go b/pkg/notify/dispatcher.go index de92af0e1e..c0788edf31 100644 --- a/pkg/notify/dispatcher.go +++ b/pkg/notify/dispatcher.go @@ -15,7 +15,6 @@ package notify import ( - "bytes" "context" "fmt" "net/http" @@ -26,15 +25,18 @@ import ( "yunion.io/x/log" "yunion.io/x/pkg/errors" "yunion.io/x/pkg/utils" + "yunion.io/x/sqlchemy" "yunion.io/x/onecloud/pkg/appctx" "yunion.io/x/onecloud/pkg/appsrv" "yunion.io/x/onecloud/pkg/appsrv/dispatcher" "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/lockman" "yunion.io/x/onecloud/pkg/cloudcommon/policy" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient" "yunion.io/x/onecloud/pkg/notify/models" + noutils "yunion.io/x/onecloud/pkg/notify/utils" "yunion.io/x/onecloud/pkg/util/rbacutils" ) @@ -72,7 +74,7 @@ func (self *NotifyModelDispatcher) DeleteConfig(ctx context.Context, params map[ } userCred := policy.FetchUserCredential(ctx) for i := range configs { - err = DeleteItem(models.ConfigManager, &configs[i], ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) + err = DeleteItem(&configs[i], ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) if err != nil { return errors.Wrapf(err, "Delete part of old one, so please input new data again.") } @@ -84,7 +86,7 @@ func (self *NotifyModelDispatcher) DeleteConfig(ctx context.Context, params map[ func (self *NotifyModelDispatcher) UpdateConfig(ctx context.Context, body jsonutils.JSONObject) error { data := body.(*jsonutils.JSONDict) contactType := data.SortedKeys()[0] - originData, err := models.ConfigManager.GetVauleByType(contactType) + originData, err := models.ConfigManager.GetConfig(contactType) if err != nil { return err } @@ -100,12 +102,13 @@ func (self *NotifyModelDispatcher) UpdateConfig(ctx context.Context, body jsonut return errors.Wrap(err, "Get Config by contactType failed") } for i := range configs { - err = DeleteItem(models.ConfigManager, &configs[i], ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) + err = DeleteItem(&configs[i], ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) if err != nil { return errors.Wrapf(err, "Delete part of old one, so please input new data again.") } } } + config := make(map[string]string) // create for _, key := range data.SortedKeys() { createData := jsonutils.NewDict() @@ -114,10 +117,13 @@ func (self *NotifyModelDispatcher) UpdateConfig(ctx context.Context, body jsonut createData.Add(jsonutils.NewString(key), "key_text") createData.Add(jsonutils.NewString(contactType), "type") _, err := self.Create(ctx, jsonutils.JSONNull, createData, nil) + config[key] = tmp.String() if err != nil { return errors.Wrapf(err, "Create config (%s, %s, %s) failed", contactType, key, tmp) } } + // update config + models.RestartService(config, contactType) return nil } @@ -128,14 +134,15 @@ func (self *NotifyModelDispatcher) CreateNotification(ctx context.Context, data // Get all contacts info of group if data contains "gid". // If no contact, return ErrContactNotFound. contactType, _ := data.GetString("contact_type") - group, id := false, "" + group := false + var ids []string if data.Contains("gid") { group = true - id, _ = data.GetString("gid") + ids = self.getIds(data, "gid") } else { - id, _ = data.GetString("uid") + ids = self.getIds(data, "uid") } - contacts, err := models.ContactManager.GetAllNotify(id, contactType, group) + contacts, err := models.ContactManager.GetAllNotify(ctx, ids, contactType, group) if err != nil { return nil, httperrors.NewGeneralError(errors.Wrap(err, "get all contacts error")) } @@ -149,6 +156,19 @@ func (self *NotifyModelDispatcher) CreateNotification(ctx context.Context, data return ret, nil } +func (self *NotifyModelDispatcher) getIds(data jsonutils.JSONObject, key string) []string { + var ids []string + tmpIds, err := data.GetArray(key) + if err != nil { + id, _ := data.GetString(key) + ids = make([]string, 1) + ids[0] = id + } else { + ids = noutils.JsonArrayToStringArray(tmpIds) + } + return ids +} + // Verify process: // 1.fetch verify by ID; 2.check that if verify is expired; // 3.if not check that if token is correct and update status of contact whose id is verify's CID @@ -181,7 +201,7 @@ func (self *NotifyModelDispatcher) Verify(ctx context.Context, params map[string } // modify contact's status and verified time. data := jsonutils.NewDict() - data.Set("status", jsonutils.NewString(models.CONTACT_VERIFIED)) + data.Set("status", jsonutils.NewString(models.VERIFICATION_VERIFIED)) data.Set("verified_at", jsonutils.NewTimeString(current)) _, err = self.Update(ctx, verifition.CID, jsonutils.JSONNull, data, nil) if err != nil { @@ -213,14 +233,15 @@ func (self *NotifyModelDispatcher) VerifyTrigger(ctx context.Context, params map return nil, httperrors.NewGeneralError(err) } // update contact state - updateDate := jsonutils.NewDict() - updateDate.Set("status", jsonutils.NewString(models.CONTACT_VERIFYING)) - err = UpdateItem(models.ContactManager, &scontact, ctx, userCred, jsonutils.JSONNull, updateDate) + if scontact.Status != models.CONTACT_VERIFYING { + scontact.SetStatus(userCred, models.CONTACT_VERIFYING, "") + } + if err != nil { return nil, httperrors.NewGeneralError(err) } processID := verification.ID - go models.SendVerifyMessage(processID, uid, contactType, contact, verification.Token) + models.SendVerifyMessage(userCred, verification, uid, contactType, contact) ret := map[string]map[string]string{ "contact": { "process_id": processID, @@ -232,10 +253,10 @@ func (self *NotifyModelDispatcher) VerifyTrigger(ctx context.Context, params map return makeNewVerify() } if scontact.Status == models.CONTACT_VERIFYING { - if err != nil { - return nil, errors.Error(fmt.Sprintf(`uid %q don't have contact %q of contact_type %q`, uid, contact, contactType)) - } - verifications, err := models.VerifyManager.FetchByCID(scontact.ID) + verifications, err := models.VerifyManager.FetchByCID(scontact.ID, func(q *sqlchemy.SQuery) *sqlchemy.SQuery { + q = q.Equals("status", models.VERIFICATION_SENT).Desc("created_at") + return q + }) if err != nil { return nil, httperrors.NewGeneralError(err) } @@ -243,7 +264,7 @@ func (self *NotifyModelDispatcher) VerifyTrigger(ctx context.Context, params map for _, verification := range verifications { if current.After(verification.ExpireAt) { //delete old one - err = DeleteItem(models.VerifyManager, &verification, ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) + err = DeleteItem(&verification, ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) if err != nil { return nil, httperrors.NewGeneralError(err) } @@ -271,7 +292,7 @@ func (self *NotifyModelDispatcher) DeleteContacts(ctx context.Context, uids2 []j userCred := policy.FetchUserCredential(ctx) deleteFailed := make([]string, 0, 1) for _, contact := range contacts { - err = DeleteItem(models.ContactManager, &contact, ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) + err = DeleteItem(&contact, ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) if err != nil { deleteFailed = append(deleteFailed, contact.ID) } @@ -283,12 +304,10 @@ func (self *NotifyModelDispatcher) DeleteContacts(ctx context.Context, uids2 []j return nil } -// UpdateContacts analysis the data, update corresponding contacts if they exist in the database or create new ones. -func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr string, query jsonutils.JSONObject, data jsonutils.JSONObject, ctxIds []dispatcher.SResourceContext) (jsonutils.JSONObject, error) { - datas, err := data.GetArray("contacts") - if err != nil { - return nil, httperrors.NewGeneralError(errors.Wrapf(err, `"contacts" not found`)) - } +// UpdateContacts analysis the data and update corresponding contacts if they exist in the database create new ones. +func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr string, query jsonutils.JSONObject, + datas []jsonutils.JSONObject, ctxIds []dispatcher.SResourceContext) error { + type pair struct { contact string enabled string @@ -300,6 +319,9 @@ func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr str contactTypes := make([]string, len(datas)) for i := range datas { contactType, _ := datas[i].GetString("contact_type") + if _, ok := models.UpdateNotAllow[contactType]; ok { + continue + } contact, _ := datas[i].GetString("contact") enabled := "-1" if datas[i].Contains("enabled") { @@ -311,7 +333,7 @@ func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr str records, err := models.ContactManager.FetchByUIDAndCType(idstr, contactTypes) if err != nil { - return nil, httperrors.NewGeneralError(err) + return httperrors.NewGeneralError(err) } // updateFailed record the information of failed update @@ -324,7 +346,7 @@ func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr str pairUpdate := contactInfos[contactType] if len(pairUpdate.contact) == 0 { // delete - err = DeleteItem(models.ContactManager, &records[i], ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) + err = DeleteItem(&records[i], ctx, userCred, jsonutils.JSONNull, jsonutils.JSONNull) if err != nil { deleteFailed = append(deleteFailed, fmt.Sprintf(`uid:%q, contact_type:%q`, idstr, contactType)) } @@ -335,7 +357,11 @@ func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr str if pairUpdate.enabled != "-1" { updateData.Set("enabled", jsonutils.NewString(pairUpdate.enabled)) } - updateData.Set("status", jsonutils.NewString("init")) + if records[i].Contact != pairUpdate.contact { + updateData.Set("status", jsonutils.NewString(models.CONTACT_INIT)) + } + // update is not relational + //updateData.Set("status", jsonutils.NewString("init")) err = UpdateItem(models.ContactManager, &records[i], ctx, userCred, jsonutils.JSONNull, updateData) if err != nil { updateFailed = append(updateFailed, fmt.Sprintf(`uid:%q, contact_type:%q, contact:%q`, idstr, contactType, pairUpdate.contact)) @@ -356,11 +382,6 @@ func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr str if conPair.enabled != "-1" { tmpMap["enabled"] = conPair.enabled } - // dingtalk don't need verify, judge and specified status for now - if conType == "dingtalk" { - tmpMap["status"] = models.CONTACT_VERIFIED - tmpMap["verified_at"] = time.Now() - } newDatas = append(newDatas, tmpMap) } @@ -373,7 +394,7 @@ func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr str // generate error through updateFailed and createFailed if len(updateFailed) != 0 || len(createFailed) != 0 || len(deleteFailed) != 0 { - var errInfoBuffer bytes.Buffer + var errInfoBuffer strings.Builder if len(updateFailed) != 0 { errInfoBuffer.WriteString(strings.Join(updateFailed, "; ")) errInfoBuffer.WriteString(" update failed. ") @@ -387,9 +408,13 @@ func (self *NotifyModelDispatcher) UpdateContacts(ctx context.Context, idstr str errInfoBuffer.WriteString(" create failed. ") } errInfo := errInfoBuffer.String() - return nil, httperrors.NewGeneralError(errors.Error(errInfo)) + return httperrors.NewGeneralError(errors.Error(errInfo)) } - return data, nil + + if query.Contains("update_dingtalk") { + models.UpdateDingtalk(idstr) + } + return nil } // fetchEnv fetch handler, params, query and body from ctx(context.Context) @@ -417,7 +442,9 @@ func mergeQueryParams(params map[string]string, query jsonutils.JSONObject, excl } // DeleteItem delete a database record corresponding to model -func DeleteItem(manager db.IModelManager, model db.IModel, ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) error { +func DeleteItem(model db.IModel, ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) error { + lockman.LockObject(ctx, model) + defer lockman.ReleaseObject(ctx, model) err := model.ValidateDeleteCondition(ctx) if err != nil { log.Errorf("validate delete condition error: %s", err) @@ -440,8 +467,9 @@ func DeleteItem(manager db.IModelManager, model db.IModel, ctx context.Context, // UpdateItem update a database record corresponding to model whose update fields are in data func UpdateItem(manager db.IModelManager, item db.IModel, ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) error { + lockman.LockObject(ctx, item) + defer lockman.ReleaseObject(ctx, item) var err error - err = item.ValidateUpdateCondition(ctx) if err != nil { diff --git a/pkg/notify/handlers.go b/pkg/notify/handlers.go index 4894a616fa..407e43857e 100644 --- a/pkg/notify/handlers.go +++ b/pkg/notify/handlers.go @@ -21,21 +21,25 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/log" - "yunion.io/x/onecloud/pkg/mcclient/modulebase" "yunion.io/x/pkg/errors" "yunion.io/x/onecloud/pkg/appsrv" "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/mcclient/modulebase" + "yunion.io/x/onecloud/pkg/notify/cache" "yunion.io/x/onecloud/pkg/notify/models" "yunion.io/x/onecloud/pkg/notify/utils" ) func InitHandlers(app *appsrv.Application) { + db.AddProjectResourceCountHandler("api/v1", app) db.RegisterModelManager(models.ContactManager) db.RegisterModelManager(models.VerifyManager) db.RegisterModelManager(models.NotificationManager) db.RegisterModelManager(models.ConfigManager) + db.RegisterModelManager(cache.UserCacheManager) + db.RegisterModelManager(cache.UserGroupCacheManager) AddNotifyDispatcher("/api/v1/", app) } @@ -46,105 +50,86 @@ func AddNotifyDispatcher(prefix string, app *appsrv.Application) { // Contact Handler modelDispatcher := NewNotifyModelDispatcher(models.ContactManager) metadata, tags = map[string]interface{}{"manager": modelDispatcher}, map[string]string{"resource": modelDispatcher.KeywordPlural()} - h := app.AddHandler2("POST", + app.AddHandler2("POST", fmt.Sprintf("%s/%s//update-contact", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(contactUpdateHandler), metadata, "contact_update", tags) - modelDispatcher.CustomizeHandlerInfo(h) // List - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s", prefix, modelDispatcher.KeywordPlural()), - modelDispatcher.Filter(listManyHandler), metadata, "list_contacts", tags) - modelDispatcher.CustomizeHandlerInfo(h) + modelDispatcher.Filter(listHandler), metadata, "list_contacts", tags) - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s/", prefix, modelDispatcher.KeywordPlural()), - modelDispatcher.Filter(listOneHandler), metadata, "list_by_uid", tags) - modelDispatcher.CustomizeHandlerInfo(h) + modelDispatcher.Filter(getHandler), metadata, "list_by_uid", tags) - h = app.AddHandler2("POST", + app.AddHandler2("POST", fmt.Sprintf("%s/%s/delete-contact", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(deleteContactHandler), metadata, "delete", tags) - modelDispatcher.CustomizeHandlerInfo(h) // verify-trigger - h = app.AddHandler2("POST", + app.AddHandler2("POST", fmt.Sprintf("%s/%s//verify", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(verifyTriggerHandler), metadata, "verify_trigger", tags) - modelDispatcher.CustomizeHandlerInfo(h) // Verify Handler, this modelDispatcher need db.DBModelDispatcher'Create function to create Contact so this modelDispatcher is // NotifyModelDispatcher whose DBModelDispatcher has modelManager models.ContactManager metadata, tags = map[string]interface{}{"manager": modelDispatcher}, map[string]string{"resource": models.VerifyManager.KeywordPlural()} - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s/", prefix, models.VerifyManager.KeywordPlural()), modelDispatcher.Filter(verifyHandler), metadata, "verify", tags) // notification Handler modelDispatcher = NewNotifyModelDispatcher(models.NotificationManager) metadata, tags = map[string]interface{}{"manager": modelDispatcher}, map[string]string{"resource": modelDispatcher.KeywordPlural()} - h = app.AddHandler2("POST", + app.AddHandler2("POST", fmt.Sprintf("%s/%s/", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(notificationHandler), metadata, "send_notifications", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s/", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(listHandler), metadata, "send_notifications", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s/", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(listHandler), metadata, "list_notification_by_id", tags) - modelDispatcher.CustomizeHandlerInfo(h) // config Handler modelDispatcher = NewNotifyModelDispatcher(models.ConfigManager) metadata, tags = map[string]interface{}{"manager": modelDispatcher}, map[string]string{"resource": modelDispatcher.KeywordPlural()} - h = app.AddHandler2("POST", + app.AddHandler2("POST", fmt.Sprintf("%s/%s/", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(configUpdateHandler), metadata, "update_configs", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s/", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(configGetHandler), metadata, "get_configs", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("DELETE", + app.AddHandler2("DELETE", fmt.Sprintf("%s/%s/", prefix, modelDispatcher.KeywordPlural()), modelDispatcher.Filter(configDeleteHandler), metadata, "delete_configs", tags) - modelDispatcher.CustomizeHandlerInfo(h) // email handler for being compatible - h = app.AddHandler2("POST", + app.AddHandler2("POST", fmt.Sprintf("%s/%s/", prefix, EMAIL_KEYWORDPLURAL), modelDispatcher.Filter(emailConfigUpdateHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s/", prefix, EMAIL_KEYWORDPLURAL), modelDispatcher.Filter(emailConfigGetHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("DELETE", + app.AddHandler2("DELETE", fmt.Sprintf("%s/%s/", prefix, EMAIL_KEYWORDPLURAL), modelDispatcher.Filter(emailConfigDeleteHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("PUT", + app.AddHandler2("PUT", fmt.Sprintf("%s/%s/", prefix, EMAIL_KEYWORDPLURAL), modelDispatcher.Filter(emailConfigUpdateHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("POST", + app.AddHandler2("POST", fmt.Sprintf("%s/%s/", prefix, SMS_KEYWORDPLURAL), modelDispatcher.Filter(smsConfigUpdateHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("GET", + app.AddHandler2("GET", fmt.Sprintf("%s/%s/", prefix, SMS_KEYWORDPLURAL), modelDispatcher.Filter(smsConfigGetHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("DELETE", + app.AddHandler2("DELETE", fmt.Sprintf("%s/%s/", prefix, SMS_KEYWORDPLURAL), modelDispatcher.Filter(smsConfigDeleteHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) - h = app.AddHandler2("PUT", + app.AddHandler2("PUT", fmt.Sprintf("%s/%s/", prefix, SMS_KEYWORDPLURAL), modelDispatcher.Filter(smsConfigUpdateHandler), metadata, "", tags) - modelDispatcher.CustomizeHandlerInfo(h) } @@ -167,13 +152,12 @@ func configGetHandler(ctx context.Context, w http.ResponseWriter, r *http.Reques func configUpdateHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) { manager, _, _, body := fetchEnv(ctx, w, r) - if body.Contains("config") { - body, _ = body.Get("config") + body, err := body.Get(models.ConfigManager.Keyword()) + if err != nil { + httperrors.GeneralServerError(w, httperrors.NewInputParameterError("need config or configs")) + return } - if body.Contains("configs") { - body, _ = body.Get("config") - } - err := manager.UpdateConfig(ctx, body) + err = manager.UpdateConfig(ctx, body) if err != nil { httperrors.GeneralServerError(w, err) } @@ -204,31 +188,30 @@ func verifyHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) // contact update handler func contactUpdateHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) { - manager, params, _, body := fetchEnv(ctx, w, r) + manager, params, query, body := fetchEnv(ctx, w, r) - var data jsonutils.JSONObject - if body != nil { - if body.Contains(manager.Keyword()) { - data, _ = body.Get(manager.Keyword()) - if data == nil { - data = body.(*jsonutils.JSONDict) - } - } else { - data = body - } - } else { - data = jsonutils.NewDict() + var data []jsonutils.JSONObject + data, err := body.GetArray(manager.Keyword(), manager.KeywordPlural()) + if err != nil { + httperrors.GeneralServerError(w, httperrors.NewInputParameterError("need %s or %s", manager.Keyword(), + manager.KeywordPlural())) + return } - // check that if the uid is exist uid := params[""] - _, err := utils.GetUserByID(uid) + _, err = utils.GetUserByID(ctx, uid) if err != nil { log.Errorf(`uid %q not found`, uid) httperrors.NotFoundError(w, "Uid Not Found") return } - _, err = manager.UpdateContacts(ctx, uid, jsonutils.JSONNull, data, nil) + queryDict := mergeQueryParams(params, query) + update, _ := body.Bool(manager.Keyword(), "update_dingtalk") + if update { + dict := queryDict.(*jsonutils.JSONDict) + dict.Add(jsonutils.JSONTrue, "update_dingtalk") + } + err = manager.UpdateContacts(ctx, uid, queryDict, data, nil) if err != nil { log.Errorf(err.Error()) httperrors.BadRequestError(w, "") @@ -280,7 +263,7 @@ func listManyHandler(ctx context.Context, w http.ResponseWriter, r *http.Request httperrors.GeneralServerError(w, err) return } - listResult = arrangeList(listResult) + listResult = arrangeList(ctx, listResult) appsrv.SendJSON(w, modulebase.ListResult2JSONWithKey(listResult, manager.KeywordPlural())) } @@ -295,6 +278,22 @@ func listHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) { appsrv.SendJSON(w, modulebase.ListResult2JSONWithKey(listResult, manager.KeywordPlural())) } +func getHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) { + manager, params, query, _ := fetchEnv(ctx, w, r) + listResult, err := manager.List(ctx, mergeQueryParams(params, query), nil) + if err != nil { + httperrors.GeneralServerError(w, err) + return + } + var data jsonutils.JSONObject + if len(listResult.Data) == 0 { + data = jsonutils.NewDict() + } else { + data = listResult.Data[0] + } + appsrv.SendJSON(w, wrap(data, manager.Keyword())) +} + func listOneHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) { manager, params, query, _ := fetchEnv(ctx, w, r) listResult, err := manager.List(ctx, mergeQueryParams(params, query), nil) @@ -302,7 +301,7 @@ func listOneHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) httperrors.GeneralServerError(w, err) return } - appsrv.SendJSON(w, wrap(arrangeOne(listResult), manager.Keyword())) + appsrv.SendJSON(w, wrap(arrangeOne(ctx, listResult), manager.Keyword())) } func wrap(data jsonutils.JSONObject, key string) jsonutils.JSONObject { @@ -314,7 +313,7 @@ func wrap(data jsonutils.JSONObject, key string) jsonutils.JSONObject { // For limit option, there is a bug but don't fix it for now. // This limit point to contact record, but these contact records whose uid are same // are considered as one record. -func arrangeList(listResult *modulebase.ListResult) *modulebase.ListResult { +func arrangeList(ctx context.Context, listResult *modulebase.ListResult) *modulebase.ListResult { ret := make(map[string]*jsonutils.JSONArray) for _, data := range listResult.Data { uid, _ := data.GetString("uid") @@ -327,7 +326,7 @@ func arrangeList(listResult *modulebase.ListResult) *modulebase.ListResult { data := make([]jsonutils.JSONObject, len(ret)) index := 0 for uid, value := range ret { - cr := models.NewSContactResponse(uid, value.String()) + cr := models.NewSContactResponse(ctx, uid, value.String()) data[index] = jsonutils.Marshal(cr) index++ } @@ -336,7 +335,7 @@ func arrangeList(listResult *modulebase.ListResult) *modulebase.ListResult { return listResult } -func arrangeOne(listResult *modulebase.ListResult) jsonutils.JSONObject { +func arrangeOne(ctx context.Context, listResult *modulebase.ListResult) jsonutils.JSONObject { if len(listResult.Data) == 0 { return jsonutils.NewDict() } @@ -345,5 +344,5 @@ func arrangeOne(listResult *modulebase.ListResult) jsonutils.JSONObject { for _, data := range listResult.Data { details.Add(data) } - return jsonutils.Marshal(models.NewSContactResponse(uid, details.String())) + return jsonutils.Marshal(models.NewSContactResponse(ctx, uid, details.String())) } diff --git a/pkg/notify/interface/doc.go b/pkg/notify/interface/doc.go new file mode 100644 index 0000000000..b3e40a4498 --- /dev/null +++ b/pkg/notify/interface/doc.go @@ -0,0 +1 @@ +package _interface // import "yunion.io/x/onecloud/pkg/notify/interface" diff --git a/pkg/notify/interface/interface.go b/pkg/notify/interface/interface.go new file mode 100644 index 0000000000..97027b669a --- /dev/null +++ b/pkg/notify/interface/interface.go @@ -0,0 +1,37 @@ +// 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 _interface + +import ( + "context" + + "yunion.io/x/onecloud/pkg/mcclient" +) + +type INotifyService interface { + InitAll() error + StopAll() + UpdateServices(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) + RestartService(ctx context.Context, config SConfig, serviceName string) + Send(ctx context.Context, contactType, contact, topic, msg, priority string) error + ContactByMobile(ctx context.Context, mobile, serviceName string) (string, error) +} + +type IServiceConfigStore interface { + GetConfig(serviceName string) (SConfig, error) + SetConfig(serviceName string, config SConfig) error +} + +type SConfig map[string]string diff --git a/pkg/notify/models/consts.go b/pkg/notify/models/consts.go index afc62c88de..a967993c69 100644 --- a/pkg/notify/models/consts.go +++ b/pkg/notify/models/consts.go @@ -19,28 +19,26 @@ const ( MOBILE = "mobile" DINGTALK = "dingtalk" WEBCONSOLE = "webconsole" - // Received a task about sending a notification - NOTIFY_RECEIVED = "received" - // Nofity module hasn't sent the notification - NOTIFY_UNSENT = "unsent" - // Nofity module has sent notification, but result unkown - NOTIFY_SENT = "sent" - // Notification was sent successfully - NOTIFY_OK = "sent_ok" - // That sent a notification is failed - NOTIFY_FAIL = "sent_fail" - // Contact's status is init which means no verifying - CONTACT_INIT = "init" - // Contact's status is verifying - CONTACT_VERIFYING = "verifying" - // Contact's status is verified - CONTACT_VERIFIED = "verified" - // Verification was sent - VERIFICATION_SENT = "sent" - // Verification was verified - VERIFICATION_VERIFIED = "verified" + NOTIFY_RECEIVED = "received" // Received a task about sending a notification + NOTIFY_SENT = "sent" // Nofity module has sent notification, but result unkown + NOTIFY_OK = "sent_ok" // Notification was sent successfully + NOTIFY_FAIL = "sent_fail" // That sent a notification is failed + + CONTACT_INIT = "init" // Contact's status is init which means no verifying + CONTACT_VERIFYING = "verifying" // Contact's status is verifying + CONTACT_VERIFIED = "verified" // Contact's status is verified + + VERIFICATION_SENT = "sent" // Verification was sent + VERIFICATION_SENT_FAIL = "sent_fail" // Verification was sent failed + VERIFICATION_VERIFIED = "verified" // Verification was verified VERIFICATION_TOKEN_EXPIRED = "Verification code expired" - VERIFICATION_TOKEN_INVALID = "Incorrect verification code" ) + +// Dingtalk account will be update automatically as mobile number change so that update dingtalk is not allowed +// In webconsole, uid is the same as contact. +var UpdateNotAllow = map[string]struct{}{ + DINGTALK: {}, + WEBCONSOLE: {}, +} diff --git a/pkg/notify/models/initdb.go b/pkg/notify/models/initdb.go index 20ef68febe..e662fca861 100644 --- a/pkg/notify/models/initdb.go +++ b/pkg/notify/models/initdb.go @@ -16,6 +16,7 @@ package models import ( "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/cloudcommon/db" ) diff --git a/pkg/notify/models/mod_config.go b/pkg/notify/models/mod_config.go index 831c002ddd..55903240f4 100644 --- a/pkg/notify/models/mod_config.go +++ b/pkg/notify/models/mod_config.go @@ -17,12 +17,14 @@ package models import ( "context" "fmt" + "yunion.io/x/jsonutils" "yunion.io/x/pkg/errors" "yunion.io/x/sqlchemy" "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/mcclient" + _interface "yunion.io/x/onecloud/pkg/notify/interface" ) type SConfigManager struct { @@ -80,19 +82,6 @@ func (self *SConfigManager) GetValue(key, contactType string) (*SConfig, error) return &configs[0], nil } -// Get all (k, v) whose type is contactType. -func (self *SConfigManager) GetVauleByType(contactType string) (map[string]string, error) { - configs, err := self.GetConfigByType(contactType) - if err != nil { - return nil, err - } - ret := make(map[string]string) - for i := range configs { - ret[configs[i].KeyText] = configs[i].ValueText - } - return ret, nil -} - 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, "") @@ -117,3 +106,19 @@ func (self *SConfigManager) GetConfigByType(contactType string) ([]SConfig, erro //} return configs, nil } + +func (self *SConfigManager) GetConfig(contactType string) (_interface.SConfig, error) { + configs, err := self.GetConfigByType(contactType) + if err != nil { + return nil, err + } + ret := make(map[string]string) + for i := range configs { + ret[configs[i].KeyText] = configs[i].ValueText + } + return ret, nil +} + +func (self *SConfigManager) SetConfig(contactType string, config _interface.SConfig) error { + return fmt.Errorf("SetConfig Not Implemented") +} diff --git a/pkg/notify/models/mod_contact.go b/pkg/notify/models/mod_contact.go index 20f4c8ad0d..5a2713a8b6 100644 --- a/pkg/notify/models/mod_contact.go +++ b/pkg/notify/models/mod_contact.go @@ -17,14 +17,16 @@ package models import ( "context" "fmt" + "strings" "time" - "yunion.io/x/jsonutils" + "yunion.io/x/pkg/errors" "yunion.io/x/sqlchemy" "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/mcclient" "yunion.io/x/onecloud/pkg/notify/utils" + "yunion.io/x/onecloud/pkg/util/rbacutils" ) type SContactManager struct { @@ -48,21 +50,47 @@ func init() { type SContact struct { SStatusStandaloneResourceBase - UID string `width:"128" nullable:"false" create:"required" list:"user" update:"user"` - ContactType string `width:"16" nullable:"false" create:"required" list:"user" update:"user"` - Contact string `width:"64" nullable:"false" create:"required" list:"user" update:"user"` - Enabled string `width:"5" nullable:"false" default:"1" create:"optional" list:"user" update:"user"` - VerifiedAt time.Time `update:"user" list:"user"` + UID string `width:"128" nullable:"false" create:"required" update:"user"` + ContactType string `width:"16" nullable:"false" create:"required" update:"user"` + Contact string `width:"64" nullable:"false" create:"required" update:"user"` + Enabled string `width:"5" nullable:"false" default:"1" create:"optional" update:"user"` + VerifiedAt time.Time `update:"user"` } func (self *SContactManager) AllowListItems(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) bool { - return db.IsAdminAllowList(userCred, self) + return true } func (self *SContactManager) AllowCreateItem(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool { return true } +func (self *SContactManager) ResourceScope() rbacutils.TRbacScope { + return rbacutils.ScopeUser +} + +func (self *SContactManager) NamespaceScope() rbacutils.TRbacScope { + return rbacutils.ScopeUser +} + +func (self *SContactManager) FetchOwnerId(ctx context.Context, + data jsonutils.JSONObject) (mcclient.IIdentityProvider, error) { + + return db.FetchUserInfo(ctx, data) +} + +func (self *SContactManager) FilterByOwner(q *sqlchemy.SQuery, owner mcclient.IIdentityProvider, + scope rbacutils.TRbacScope) *sqlchemy.SQuery { + if owner != nil { + if scope == rbacutils.ScopeUser { + if len(owner.GetUserId()) > 0 { + q = q.Equals("uid", owner.GetUserId()) + } + } + } + return q +} + func (self *SContactManager) InitializeData() error { sql := fmt.Sprintf("update %s set updated_at=update_at, deleted=is_deleted", self.TableSpec().Name()) q := sqlchemy.NewRawQuery(sql, "") @@ -103,34 +131,103 @@ func (self *SContactManager) FetchByMore(uid, contact, contactType string) ([]SC return records, nil } -func (self *SContactManager) FetchDingtalkContacts(uid string) { - // todo +func (self *SContact) GetCustomizeColumns(ctx context.Context, userCred mcclient.TokenCredential, + query jsonutils.JSONObject) *jsonutils.JSONDict { + + ret, _ := self.getMoreDetail(ctx, userCred, query) + return ret +} + +func (self *SContact) getMoreDetail(ctx context.Context, userCred mcclient.TokenCredential, + query jsonutils.JSONObject) (*jsonutils.JSONDict, error) { + + ret := jsonutils.NewDict() + uname, err := utils.GetUsernameByID(ctx, self.UID) + if err != nil { + return ret, err + } + + q := ContactManager.Query().Equals("uid", self.UID) + contacts := make([]SContact, 0) + err = db.FetchModelObjects(ContactManager, q, &contacts) + if err != nil { + return ret, errors.Wrapf(err, "fetch Contacts of uid %s error", self.UID) + } + ret.Add(jsonutils.NewString(self.UID), "id") + ret.Add(jsonutils.NewString(uname), "name") + ret.Add(jsonutils.NewString(jsonutils.Marshal(contacts).String()), "details") + + return ret, nil +} + +func (self *SContact) GetExtraDetail(ctx context.Context, userCred mcclient.TokenCredential, + query jsonutils.JSONObject) (*jsonutils.JSONDict, error) { + + return self.getMoreDetail(ctx, userCred, query) } func (self *SContactManager) ListItemFilter(ctx context.Context, q *sqlchemy.SQuery, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (*sqlchemy.SQuery, error) { queryDict := query.(*jsonutils.JSONDict) if queryDict.Contains("uid") { uid, _ := queryDict.GetString("uid") - q = q.Filter(sqlchemy.Equals(q.Field("uid"), uid)) + q = q.Equals("uid", uid) } + // for now + if queryDict.Contains("filter") { + filterCon, _ := queryDict.GetString("filter") + queryDict.Remove("filter") + contain := "name.contains(" + index := strings.Index(filterCon, contain) + if index < 0 { + return q, nil + } + filterCon = filterCon[index+len(contain):] + index = strings.Index(filterCon, ")") + if index < 0 { + return q, nil + } + name := filterCon[:index] + ids, err := utils.GetUserIdsLikeName(ctx, name) + if err != nil { + return q, nil + } + q = q.In("uid", ids) + } + + scopeStr, err := query.GetString("scope") + if err != nil { + scopeStr = "system" + } + scope := rbacutils.TRbacScope(scopeStr) + + if !scope.HigherEqual(rbacutils.ScopeSystem) { + q = q.Equals("uid", userCred.GetUserId()) + } + q = q.GroupBy("uid") + return q, nil } -func (self *SContactManager) GetAllNotify(id, contactType string, group bool) ([]SContact, error) { +func (self *SContactManager) GetAllNotify(ctx context.Context, ids []string, contactType string, group bool) ([]SContact, error) { var uids []string var err error q := self.Query() if !group { - q.Filter(sqlchemy.AND(sqlchemy.Equals(q.Field("uid"), id), sqlchemy.Equals(q.Field("contact_type"), contactType), sqlchemy.Equals(q.Field("status"), CONTACT_VERIFIED))) - uids = []string{id} + uids = ids } else { - uids, err = utils.GetUsersByGroupID(id) - if err != nil { - return nil, err + uid := make([]string, 0) + for _, id := range uids { + tmpUids, err := utils.GetUsersByGroupID(ctx, id) + if err != nil { + return nil, err + } + uid = append(uid, tmpUids...) } - q.Filter(sqlchemy.AND(sqlchemy.In(q.Field("uid"), uids), sqlchemy.Equals(q.Field("contact_type"), contactType))) } + q.Filter(sqlchemy.AND(sqlchemy.In(q.Field("uid"), uids), sqlchemy.Equals(q.Field("contact_type"), + contactType), sqlchemy.Equals(q.Field("status"), CONTACT_VERIFIED))) + if contactType == WEBCONSOLE { ret := make([]SContact, len(uids)) for i := range uids { @@ -156,8 +253,8 @@ type SContactResponse struct { Details string } -func NewSContactResponse(uid string, details string) SContactResponse { - name, _ := utils.GetUsernameByID(uid) +func NewSContactResponse(ctx context.Context, uid string, details string) SContactResponse { + name, _ := utils.GetUsernameByID(ctx, uid) return SContactResponse{ Id: uid, Name: name, diff --git a/pkg/notify/models/mod_notification.go b/pkg/notify/models/mod_notification.go index f5734a854e..1e0fda70a1 100644 --- a/pkg/notify/models/mod_notification.go +++ b/pkg/notify/models/mod_notification.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "fmt" + "strings" "sync" "time" @@ -30,7 +31,11 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/policy" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/notify/cache" + _interface "yunion.io/x/onecloud/pkg/notify/interface" + "yunion.io/x/onecloud/pkg/notify/options" "yunion.io/x/onecloud/pkg/notify/utils" + "yunion.io/x/onecloud/pkg/util/rbacutils" ) type SNotificationManager struct { @@ -38,6 +43,7 @@ type SNotificationManager struct { } var NotificationManager *SNotificationManager +var NotifyService _interface.INotifyService func init() { NotificationManager = &SNotificationManager{ @@ -51,17 +57,102 @@ func init() { NotificationManager.SetVirtualObject(NotificationManager) } +func (self *SNotificationManager) ResourceScope() rbacutils.TRbacScope { + return rbacutils.ScopeUser +} + +func (self *SNotificationManager) NamespaceScope() rbacutils.TRbacScope { + return rbacutils.ScopeUser +} + +func (self *SNotificationManager) FetchOwnerId(ctx context.Context, + data jsonutils.JSONObject) (mcclient.IIdentityProvider, error) { + + return db.FetchUserInfo(ctx, data) +} + +func (self *SNotificationManager) FilterByOwner(q *sqlchemy.SQuery, owner mcclient.IIdentityProvider, + scope rbacutils.TRbacScope) *sqlchemy.SQuery { + if owner != nil { + if scope == rbacutils.ScopeUser { + if len(owner.GetUserId()) > 0 { + q = q.Equals("uid", owner.GetUserId()) + } + } + } + return q +} + type SNotification struct { SStatusStandaloneResourceBase UID string `width:"128" nullable:"false" create:"required"` - ContactType string `width:"16" nullable:"false" create:"required"` - Topic string `width:"128" nullable:"false" create:"optional"` - Priority string `width:"16" nullable:"false" create:"optional"` + ContactType string `width:"16" nullable:"false" create:"required" list:"user" index:"true"` + Topic string `width:"128" nullable:"false" create:"optional" list:"user"` + Priority string `width:"16" nullable:"false" create:"optional" list:"user"` Msg string `create:"required"` - ReceivedAt time.Time `nullable:"false"` + ReceivedAt time.Time `nullable:"false" list:"user" create:"optional"` SendAt time.Time `nullable:"false"` SendBy string `width:"128" nullable:"false"` + // ClusterID identify message with same topic, msg, priority + ClusterID string `width:"128" charset:"ascii" primary:"true" create:"optional"` +} + +type UserDetail struct { + Status string + Name string + ReceivedAt time.Time +} + +func (self *SNotification) GetCustomizeColumns(ctx context.Context, userCred mcclient.TokenCredential, + query jsonutils.JSONObject) *jsonutils.JSONDict { + + // collect user infos + scopeStr, err := query.GetString("scope") + if err != nil { + scopeStr = "system" + } + scope := rbacutils.TRbacScope(scopeStr) + + var userDetails []UserDetail + if scope.HigherEqual(rbacutils.ScopeSystem) { + // fetch users from database + userDetails, err = NotificationManager.fetchUserDetailByClusterID(ctx, self.ClusterID) + if err != nil { + log.Errorf(err.Error()) + } + + } else { + userDetail := UserDetail{ + Status: self.Status, + Name: userCred.GetUserId(), + ReceivedAt: self.ReceivedAt, + } + name, err := utils.GetUsernameByID(ctx, self.UID) + if err == nil && len(name) != 0 { + userDetail.Name = name + } + userDetails = []UserDetail{userDetail} + } + ret := jsonutils.NewDict() + ret.Add(jsonutils.Marshal(userDetails), "user_list") + return ret +} + +func (self *SNotification) GetExtraDetails(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (*jsonutils.JSONDict, error) { + userDetail := UserDetail{ + Status: self.Status, + Name: self.UID, + ReceivedAt: self.ReceivedAt, + } + name, err := utils.GetUsernameByID(ctx, self.UID) + if err == nil && len(name) != 0 { + userDetail.Name = name + } + ret := jsonutils.NewDict() + data := jsonutils.Marshal([]UserDetail{userDetail}) + ret.Add(data, "user_list") + return ret, nil } func (self *SNotificationManager) AllowListItems(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) bool { @@ -72,13 +163,120 @@ func (self *SNotificationManager) AllowCreateItem(ctx context.Context, userCred return db.IsAdminAllowCreate(userCred, self) } +type sUpdate struct { + ID string + UID string + Topic string + Priority string + ContactType string +} + func (self *SNotificationManager) InitializeData() error { - sql := fmt.Sprintf("update %s set updated_at=update_at, deleted=is_deleted", self.TableSpec().Name()) - q := sqlchemy.NewRawQuery(sql, "") - q.Row() + scope := time.Duration(options.Options.InitNotificationScope) * time.Hour + time := time.Now().Add(-scope) + q := self.Query("id", "uid", "topic", "priority", "contact_type").GE("created_at", + time).Desc("received_at").Equals("contact_type", "webconsole") + q = q.Filter(sqlchemy.OR(sqlchemy.IsNull(q.Field("cluster_id")), sqlchemy.IsEmpty(q.Field("cluster_id")))) + rows, err := q.Rows() + if err != nil { + return err + } + updates, update := make([]sUpdate, 0, 10), sUpdate{} + for rows.Next() { + err := rows.Scan(&update.ID, &update.UID, &update.Topic, &update.Priority, &update.ContactType) + if err == nil { + updates = append(updates, update) + } + } + log.Debugf("this is total %d updates", len(updates)) + + // updates is too little + //if len(updates) < 100 { + // updates = updates[:0] + // q := self.Query("id", "uid", "topic", "priority", "contact_type").Desc("received_at").Equals("contact_type", + // "webconsole").Limit(500) + // q = q.Filter(sqlchemy.OR(sqlchemy.IsNull(q.Field("cluster_id")), sqlchemy.IsEmpty(q.Field("cluster_id")))) + // rows, err := q.Rows() + // if err != nil { + // return err + // } + // for rows.Next() { + // err := rows.Scan(&update.ID, &update.UID, &update.Topic, &update.Priority, &update.ContactType) + // if err == nil { + // updates = append(updates, update) + // } + // } + // log.Debugf("this is total %d updates", len(updates)) + //} + + cache := make([]string, 0, 10) + if len(updates) > 0 { + cache = append(cache, updates[0].ID) + } + for i := 1; i < len(updates); i++ { + if updates[i].Topic == updates[i-1].Topic && updates[i].Priority == updates[i-1].Priority { + + cache = append(cache, updates[i].ID) + continue + } + err = self.syncDatabase(cache) + if err != nil { + return errors.Wrap(err, "exec sql error") + } + cache = cache[:0] + if i < len(updates)-1 { + cache = append(cache, updates[i].ID) + } + } + if len(cache) == 0 { + return nil + } + err = self.syncDatabase(cache) + if err != nil { + return errors.Wrap(err, "exec sql error") + } + return nil } +func (self *SNotificationManager) syncDatabase(ids []string) error { + + sql := "update %s set updated_at=update_at, deleted=is_deleted, cluster_id='%s' where id in %s" + + newUid := DefaultUUIDGenerator() + + buffer := new(strings.Builder) + buffer.WriteString("(") + for _, id := range ids { + buffer.WriteString("'") + buffer.WriteString(id) + buffer.WriteString("', ") + } + newSql := fmt.Sprintf(sql, self.TableSpec().Name(), newUid, buffer.String()[:buffer.Len()-2]+")") + q := sqlchemy.NewRawQuery(newSql) + rows, err := q.Rows() + defer rows.Close() + return err +} + +func (self *SNotificationManager) ListItemFilter(ctx context.Context, q *sqlchemy.SQuery, userCred mcclient.TokenCredential, + query jsonutils.JSONObject) (*sqlchemy.SQuery, error) { + + // no domainID for now + scopeStr, err := query.GetString("scope") + if err != nil { + scopeStr = "system" + } + scope := rbacutils.TRbacScope(scopeStr) + + if !scope.HigherEqual(rbacutils.ScopeSystem) { + q = q.Equals("uid", userCred.GetUserId()) + } + + q = q.GroupBy("cluster_id") + return q, nil +} + func (self *SNotificationManager) BatchCreate(ctx context.Context, data jsonutils.JSONObject, contacts []SContact) ([]string, error) { userCred := policy.FetchUserCredential(ctx) ownerID, err := utils.FetchOwnerId(ctx, NotificationManager, userCred, jsonutils.JSONNull) @@ -90,15 +288,18 @@ func (self *SNotificationManager) BatchCreate(ctx context.Context, data jsonutil topic, _ := data.GetString("topic") createFailed, createSuccess, contactSuccess := make([]string, 0), make([]*SNotification, 0, len(contacts)/2), make([]string, 0, len(contacts)/2) + now, clusterId := time.Now(), DefaultUUIDGenerator() for i := range contacts { - createData := map[string]string{ - "uid": contacts[i].ID, + createData := map[string]interface{}{ + "uid": contacts[i].UID, "contact_type": contacts[i].ContactType, "topic": topic, "priority": priority, "msg": msg, + "received_at": now, "send_by": userCred.GetUserId(), - "status": NOTIFY_UNSENT, + "status": NOTIFY_RECEIVED, + "cluster_id": clusterId, } model, err := db.DoCreate(self, ctx, userCred, jsonutils.JSONNull, jsonutils.Marshal(createData), ownerID) if err != nil { @@ -108,7 +309,7 @@ func (self *SNotificationManager) BatchCreate(ctx context.Context, data jsonutil contactSuccess = append(contactSuccess, contacts[i].Contact) } } - go send(createSuccess, userCred, contactSuccess) + Send(createSuccess, userCred, contactSuccess) if len(createFailed) != 0 { errInfo := new(bytes.Buffer) errInfo.WriteString("notifications whose uid are ") @@ -128,9 +329,46 @@ func (self *SNotificationManager) BatchCreate(ctx context.Context, data jsonutil return notificationIDs, nil } +func (self *SNotificationManager) fetchUserDetailByClusterID(ctx context.Context, clusterID string) ([]UserDetail, + error) { + q := self.Query("uid", "status", "received_at").Equals("cluster_id", clusterID) + row, err := q.Rows() + if err != nil { + return nil, err + } + ret := make([]UserDetail, 0) + userIds := make([]string, 0) + var userId, status string + var receviedAt time.Time + for row.Next() { + err := row.Scan(&userId, &status, &receviedAt) + if err != nil { + return nil, errors.Wrap(err, "sql.row parse error") + } + userIds = append(userIds, userId) + ret = append(ret, UserDetail{ + Status: status, + Name: userId, + ReceivedAt: receviedAt, + }) + } + + userMap, err := cache.UserCacheManager.FetchUsersByIDs(ctx, userIds) + if err != nil { + return nil, errors.Wrap(err, "fetch users by ids failed") + } + for i := range ret { + if user, ok := userMap[ret[i].Name]; ok { + ret[i].Name = user.Name + } + } + + return ret, nil +} + func (self *SNotificationManager) FetchNotOK(lastTime time.Time) ([]SNotification, error) { q := self.Query() - q.Filter(sqlchemy.AND(sqlchemy.GE(q.Field("created_at"), lastTime), sqlchemy.NotEquals(q.Field("status"), NOTIFY_UNSENT))) + q.Filter(sqlchemy.AND(sqlchemy.GE(q.Field("created_at"), lastTime), sqlchemy.NotEquals(q.Field("status"), NOTIFY_OK))) records := make([]SNotification, 0, 10) err := db.FetchModelObjects(self, q, &records) if err != nil { @@ -139,28 +377,26 @@ func (self *SNotificationManager) FetchNotOK(lastTime time.Time) ([]SNotificatio return records, nil } -func send(notifications []*SNotification, userCred mcclient.TokenCredential, contacts []string) { - var wg sync.WaitGroup - sendone := func(notification *SNotification, contact string) { - err := notification.SetSentAndTime(userCred) - if err != nil { - log.Errorf("Change notification's status failed.") - return - } - err = RpcService.Send(notification.ContactType, contact, notification.Topic, notification.Msg, notification.Priority) - if err != nil { - log.Errorf("Send notification failed because that %s.", err.Error()) - notification.SetStatus(userCred, NOTIFY_FAIL, err.Error()) - } else { - notification.SetStatus(userCred, NOTIFY_OK, "") - } - wg.Done() +func (self *SNotification) SetStatus(userCred mcclient.TokenCredential, status string, reason string) error { + if self.Status == status { + return nil } - for i := range notifications { - wg.Add(1) - go sendone(notifications[i], contacts[i]) + oldStatus := self.Status + _, err := db.Update(self, func() error { + self.Status = status + return nil + }) + if err != nil { + return err } - wg.Wait() + if userCred != nil { + notes := fmt.Sprintf("%s=>%s", oldStatus, status) + if len(reason) > 0 { + notes = fmt.Sprintf("%s: %s", notes, reason) + } + db.OpsLog.LogEvent(self, db.ACT_UPDATE_STATUS, notes, userCred) + } + return nil } func (self *SNotification) SetSentAndTime(userCred mcclient.TokenCredential) error { @@ -211,7 +447,9 @@ func sendWithoutUserCred(notifications []SNotification) { } // sent_at update todo notification.SetStatusWithoutUserCred(NOTIFY_SENT) - err = RpcService.Send(notification.ContactType, contact[0].Contact, notification.Topic, notification.Msg, notification.Priority) + err = NotifyService.Send(context.Background(), notification.ContactType, contact[0].Contact, notification.Topic, + notification.Msg, + notification.Priority) if err == nil { return } @@ -232,16 +470,9 @@ func sendWithoutUserCred(notifications []SNotification) { func ReSend(minutes int) { scope := time.Duration(minutes) * time.Minute - for { - select { - case <-time.After(scope / 2): - //lastTime := time.Now().Add(-scope) - //q := NotificationManager.Query() - notifications, err := NotificationManager.FetchNotOK(time.Now().Add(-scope)) - if err != nil { - break - } - sendWithoutUserCred(notifications) - } + notifications, err := NotificationManager.FetchNotOK(time.Now().Add(-scope)) + if err != nil { + return } + sendWithoutUserCred(notifications) } diff --git a/pkg/notify/models/mod_verify.go b/pkg/notify/models/mod_verify.go index b86c9e5769..bf4fdb32f2 100644 --- a/pkg/notify/models/mod_verify.go +++ b/pkg/notify/models/mod_verify.go @@ -16,18 +16,14 @@ package models import ( "context" - "encoding/json" "fmt" - "strings" "time" "yunion.io/x/jsonutils" - "yunion.io/x/log" "yunion.io/x/sqlchemy" "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/mcclient" - "yunion.io/x/onecloud/pkg/notify/options" "yunion.io/x/onecloud/pkg/notify/utils" ) @@ -77,6 +73,7 @@ func NewSVerify(contactType string, cid string) *SVerify { SendAt: now, } ret.ID = DefaultUUIDGenerator() + ret.SetModelManager(VerifyManager, ret) return ret } @@ -87,9 +84,10 @@ func (self *SVerifyManager) InitializeData() error { return nil } -func (self *SVerifyManager) FetchByCID(cid string) ([]SVerify, error) { +func (self *SVerifyManager) FetchByCID(cid string, filter func(q *sqlchemy.SQuery) *sqlchemy.SQuery) ([]SVerify, error) { q := self.Query() q.Filter(sqlchemy.Equals(q.Field("cid"), cid)) + q = filter(q) records := make([]SVerify, 0, 1) err := db.FetchModelObjects(self, q, &records) if err != nil { @@ -121,33 +119,3 @@ func (self *SVerifyManager) Create(ctx context.Context, userCred mcclient.TokenC } return nil } - -func SendVerifyMessage(processId, uid, contactType, contact, token string) { - var err error - var msg string - if contactType == "email" { - emailUrl := strings.Replace(options.Options.VerifyEmailUrl, "{0}", processId, 1) - emailUrl = strings.Replace(emailUrl, "{1}", token, 1) - - // get uName - uName, err := utils.GetUsernameByID(uid) - if err != nil || len(uName) == 0 { - uName = "用户" - } - data := struct { - Name string - Link string - }{uName, emailUrl} - jsonStr, _ := json.Marshal(data) - msg = string(jsonStr) - } else if contactType == "mobile" { - msg = fmt.Sprintf(`{"code": "%s"}`, token) - } else { - //todo - } - - err = RpcService.Send(contactType, contact, "verify", msg, "") - if err != nil { - log.Errorf("Send verify message failed because that %s.", err.Error()) - } -} diff --git a/pkg/notify/models/send.go b/pkg/notify/models/send.go deleted file mode 100644 index 0111dff007..0000000000 --- a/pkg/notify/models/send.go +++ /dev/null @@ -1,269 +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 models - -import ( - "fmt" - "io/ioutil" - "net/rpc" - "os" - "path/filepath" - "strings" - "sync" - "time" - - "yunion.io/x/log" - "yunion.io/x/pkg/errors" -) - -const ( - // ErrSendServiceNotFound means SRpcService's SendSerivces hasn't this Send Service. - ErrSendServiceNotFound = errors.Error("Send Service Not Found") - NOTINIT = "Send service hasn't been init" -) - -// RpcService is a single case of SRpcService -var RpcService *SRpcService - -// SRpcService provide rpc service about sending message for notify module and manage these services. -// SendServices storage all send service and its name. -// lock protect the SendServices. -type SRpcService struct { - SendServices map[string]*rpc.Client - socketFileDir string - lock sync.RWMutex -} - -// NewSRpcService create a SRpcService -func NewSRpcService(socketFileDir string) *SRpcService { - return &SRpcService{ - SendServices: make(map[string]*rpc.Client), - socketFileDir: socketFileDir, - } -} - -// InitAll init all Send Services, the init process is that: -// find all socket file in directory 'self.socketFileDir', if wrong return error; -// the name of file is the service's name; then try to dial to this rpc service -// through corresponding socket file, if failed, only print log but not return error. -func (self *SRpcService) InitAll() error { - files, err := ioutil.ReadDir(self.socketFileDir) - if err != nil { - return errors.Wrapf(err, "read dir %s failed", self.socketFileDir) - } - for _, file := range files { - filename := file.Name() - if !file.IsDir() && strings.Contains(filename, ".sock") { - serviceName := filename[:len(filename)-5] - self.checkAndAddOne(serviceName) - } - } - if len(self.SendServices) == 0 { - log.Errorf("No available send service.") - } else { - log.Infof("Total %d send service init successful", len(self.SendServices)) - } - return nil -} - -// UpdateServices will detect the self.sockFileDir every delay seconds. -// Add new service and delete disappeared one from self.SendServices. -func (self *SRpcService) UpdateServices(delay int) { - for { - select { - case <-time.After(time.Duration(delay) * time.Second): - err := self.updateService() - if err != nil { - log.Errorf("update services failed because that %s.", err.Error()) - } - } - } -} - -// StopAll stop all send service in self.SenderServices normally which can delete the socket file. -func (self *SRpcService) StopAll() { - for _, service := range self.SendServices { - service.Close() - } -} - -// Send call the corresponding rpc server.Send to send messager. -func (self *SRpcService) Send(contactType, contact, topic, msg, priority string) error { - self.lock.RLock() - sendService, ok := self.SendServices[contactType] - self.lock.RUnlock() - var err error - if !ok { - sendService, err = self.checkAndAddOne(contactType) - if err == ErrDial { - return ErrSendServiceNotFound - } - if err != nil { - return errors.Wrap(err, "Check or Add connection failed") - } - } - args := SSendArgs{ - Contact: contact, - Topic: topic, - Message: msg, - Priority: priority, - } - reply := SSendReply{} - err = sendService.Call("Server.Send", &args, &reply) - if err != nil { - // should check and send again. - // Possible situation: notify always keep connection but remote guy have restarted - // so that connection valid. - sendService, err = self.checkAndAddOne(contactType) - if err != nil { - return errors.Wrap(err, "Check or Add connection failed") - } - err = sendService.Call("Server.Send", &args, &reply) - if err != nil { - return errors.Wrap(err, "Send message failed.") - } - if !reply.Success { - return errors.Error(fmt.Sprintf("Send message failed because that %s.", reply.Msg)) - } - } - if !reply.Success { - if reply.Msg != NOTINIT { - return errors.Error(fmt.Sprintf("Send message failed because that %s.", reply.Msg)) - } - // should check and send again - sendService, err = self.checkAndAddOne(contactType) - if err != nil { - return errors.Wrap(err, "Check or Add connection failed") - } - err = sendService.Call("Server.Send", &args, &reply) - if err != nil { - return errors.Wrap(err, "Send message failed.") - } - if !reply.Success { - return errors.Error(fmt.Sprintf("Send message failed because that %s.", reply.Msg)) - } - } - return nil -} - -// RestartService can restart remote rpc server and pass config info. -// When first init notify Server, must Call this function. -// When accept the request about changing config, must Call this function. -func (self *SRpcService) RestartService(config map[string]string, serviceName string) { - self.lock.RLock() - sendService, ok := self.SendServices[serviceName] - self.lock.RUnlock() - var err error - if !ok { - sendService, err = self.checkAndAddOne(serviceName) - if err != nil { - log.Debugf("Restart Failed: %s", err.Error()) - return - } - } - args := SRestartArgs{Config: config} - - reply := SSendReply{} - err = sendService.Call("Server.UpdateConfig", &args, &reply) - if err != nil || !reply.Success { - log.Errorf("Restart rpc serve whose name is %s failed.", serviceName) - return - } -} - -// CheckAndAddOne check the status of service 'serviceName' -// If fail to dial to service, delete and remove sock file. -// if dial successfully, try to restart the service. -func (self *SRpcService) checkAndAddOne(serviceName string) (*rpc.Client, error) { - // Try to connect again - filename := filepath.Join(self.socketFileDir, serviceName+".sock") - rpcService, err := rpc.Dial("unix", filename) - if err != nil { - log.Debugf("Try to dial to service failed which unix socket file name is %s.", filename) - // This file maybe left behind inadvertently, so we should try to delete it - os.Remove(filename) - self.lock.Lock() - delete(self.SendServices, serviceName) - self.lock.Unlock() - return nil, ErrDial - } - - // GetKeyValue to config rpc Service - config, err := ConfigManager.GetVauleByType(serviceName) - if err != nil { - log.Debugf("Init service error which unix socket file name is %s because that get config about this failed", filename) - return nil, ErrGetConfig - } - args := SRestartArgs{config} - reply := SSendReply{} - rpcService.Call("Server.UpdateConfig", &args, &reply) - if !reply.Success { - log.Debugf("Init service error which unix socket file name is %s because that %s.", filename, reply.Msg) - return nil, ErrUpdateConfig - } - self.lock.Lock() - self.SendServices[serviceName] = rpcService - self.lock.Unlock() - return rpcService, nil -} - -func (self *SRpcService) updateService() error { - files, err := ioutil.ReadDir(self.socketFileDir) - if err != nil { - return errors.Wrapf(err, "read dir %s failed", self.socketFileDir) - } - original := make(map[string]*rpc.Client) - self.lock.RLock() - for serviceName, client := range self.SendServices { - original[serviceName] = client - } - self.lock.RUnlock() - for _, file := range files { - filename := file.Name() - if !file.IsDir() && strings.Contains(filename, ".sock") { - serviceName := filename[:len(filename)-5] - if _, ok := self.SendServices[serviceName]; ok { - delete(original, serviceName) - continue - } - self.checkAndAddOne(serviceName) - } - } - self.lock.Lock() - for serviceName := range original { - delete(self.SendServices, serviceName) - } - self.lock.Unlock() - for _, client := range original { - client.Close() - } - return nil -} - -type SSendArgs struct { - Contact string - Topic string - Message string - Priority string -} - -type SRestartArgs struct { - Config map[string]string -} - -type SSendReply struct { - Success bool - Msg string -} diff --git a/pkg/notify/models/standalone.go b/pkg/notify/models/standalone.go index 270af9de5b..1c20a01bd0 100644 --- a/pkg/notify/models/standalone.go +++ b/pkg/notify/models/standalone.go @@ -32,7 +32,7 @@ var ( type SStandaloneResourceBase struct { SResourceBase - ID string `width:"128" charset:"ascii" primary:"true" list:"user" create:"optional"` + ID string `width:"128" charset:"ascii" primary:"true" create:"optional"` } func (model *SStandaloneResourceBase) BeforeInsert() { diff --git a/pkg/notify/models/statusstandalone.go b/pkg/notify/models/statusstandalone.go index 2c49b7781b..64908a30e9 100644 --- a/pkg/notify/models/statusstandalone.go +++ b/pkg/notify/models/statusstandalone.go @@ -29,7 +29,7 @@ import ( type SStatusStandaloneResourceBase struct { SStandaloneResourceBase - Status string `width:"36" charset:"ascii" nullable:"false" default:"init" list:"user" create:"optional" update:"user"` + Status string `width:"36" charset:"ascii" nullable:"false" default:"init" create:"optional" update:"user"` } type SStatusStandaloneResourceBaseManager struct { diff --git a/pkg/notify/models/worker.go b/pkg/notify/models/worker.go new file mode 100644 index 0000000000..6b40673d0b --- /dev/null +++ b/pkg/notify/models/worker.go @@ -0,0 +1,170 @@ +// 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 ( + "context" + "fmt" + "strings" + "time" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + + "yunion.io/x/onecloud/pkg/appsrv" + "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/notify/options" + "yunion.io/x/onecloud/pkg/notify/utils" +) + +var workMan *appsrv.SWorkerManager + +func init() { + workMan = appsrv.NewWorkerManager("NotifyWokerManager", 16, 512, false) +} + +func Send(notifications []*SNotification, userCred mcclient.TokenCredential, contacts []string) { + + for i := range notifications { + workMan.Run(func() { + sendone(context.Background(), userCred, notifications[i], contacts[i]) + }, nil, nil) + } +} + +func sendone(ctx context.Context, userCred mcclient.TokenCredential, notification *SNotification, contact string) { + err := notification.SetSentAndTime(userCred) + if err != nil { + log.Errorf("Change notification's status failed.") + return + } + err = NotifyService.Send(ctx, notification.ContactType, contact, notification.Topic, notification.Msg, + notification.Priority) + if err != nil { + log.Errorf("Send notification failed because that %s.", err.Error()) + notification.SetStatus(userCred, NOTIFY_FAIL, err.Error()) + } else { + log.Debugf("send notification successfully") + notification.SetStatus(userCred, NOTIFY_OK, "") + } +} + +func RestartService(config map[string]string, serviceName string) { + workMan.Run(func() { + NotifyService.RestartService(context.Background(), config, serviceName) + }, nil, nil) +} + +func SendVerifyMessage(userCred mcclient.TokenCredential, verify *SVerify, uid, contactType, contact string) { + workMan.Run(func() { + sendVerifyMessage(context.Background(), userCred, verify, uid, contactType, contact) + }, nil, nil) +} + +func sendVerifyMessage(ctx context.Context, userCred mcclient.TokenCredential, verify *SVerify, uid, contactType, + contact string) { + var ( + err error + msg string + ) + processId, token := verify.ID, verify.Token + if contactType == "email" { + emailUrl := strings.Replace(options.Options.VerifyEmailUrl, "{0}", processId, 1) + emailUrl = strings.Replace(emailUrl, "{1}", token, 1) + + // get uName + uName, err := utils.GetUsernameByID(ctx, uid) + if err != nil || len(uName) == 0 { + uName = "用户" + } + data := struct { + Name string + Link string + }{uName, emailUrl} + msg = jsonutils.Marshal(data).String() + } else if contactType == "mobile" { + msg = fmt.Sprintf(`{"code": "%s"}`, token) + } else { + // todo + return + } + + err = NotifyService.Send(ctx, contactType, contact, "verify", msg, "") + if err != nil { + verify.SetStatus(userCred, VERIFICATION_SENT_FAIL, "") + log.Errorf("Send verify message failed because that %s.", err.Error()) + return + } + verify.SetStatus(userCred, VERIFICATION_SENT, "") +} + +func UpdateDingtalk(uid string) { + workMan.Run(func() { + updateDingtalk(context.Background(), uid) + }, nil, nil) +} + +func updateDingtalk(ctx context.Context, uid string) { + contacts, err := ContactManager.FetchByUIDAndCType(uid, []string{MOBILE, DINGTALK}) + if err != nil { + log.Errorf("fetch contacts error") + } + if len(contacts) == 0 { + return + } + var mobileContact, dingtalkContact *SContact + for i := range contacts { + if contacts[i].ContactType == MOBILE { + mobileContact = &contacts[i] + } else { + dingtalkContact = &contacts[i] + } + } + if mobileContact == nil { + return + } + + userid, err := NotifyService.ContactByMobile(ctx, mobileContact.Contact, DINGTALK) + if err != nil { + log.Errorf("fetch dingtalk userid by mobile failed: %s", err.Error()) + } + if dingtalkContact != nil { + dingtalkContact.SetModelManager(ContactManager, dingtalkContact) + origin := dingtalkContact.Contact + _, err := db.Update(dingtalkContact, func() error { + dingtalkContact.Contact = userid + return nil + }) + if err != nil { + log.Errorf("update dingtalk userid %s => %s failed", origin, userid) + } + return + } + + contact := SContact{ + UID: uid, + ContactType: DINGTALK, + Contact: userid, + Enabled: "1", + VerifiedAt: time.Now(), + } + contact.Status = CONTACT_VERIFIED + + err = ContactManager.TableSpec().InsertOrUpdate(&contact) + if err != nil { + log.Errorf("create new dingtalk contact failed") + } +} diff --git a/pkg/notify/options/options.go b/pkg/notify/options/options.go index e2cf8e8dfa..370609fd59 100644 --- a/pkg/notify/options/options.go +++ b/pkg/notify/options/options.go @@ -22,11 +22,12 @@ type NotifyOption struct { options.CommonOptions options.DBOptions - 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` - VerifyEmailUrl string - ReSendScope int `help:"Resend all messages that have not been sent successfully within ReSendScope minutes"` + 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` + VerifyEmailUrl string `help:"url of verify email"` + ReSendScope int `help:"Resend all messages that have not been sent successfully within ReSendScope minutes"` + InitNotificationScope int `help:"initialize data of notification with in InitNotificationScope hours" default:100` } var Options NotifyOption diff --git a/pkg/notify/rpc/apis/doc.go b/pkg/notify/rpc/apis/doc.go new file mode 100644 index 0000000000..e27ffe4328 --- /dev/null +++ b/pkg/notify/rpc/apis/doc.go @@ -0,0 +1 @@ +package apis // import "yunion.io/x/onecloud/pkg/notify/rpc/apis" diff --git a/pkg/notify/rpc/apis/send_client.go b/pkg/notify/rpc/apis/send_client.go new file mode 100644 index 0000000000..cae3506e87 --- /dev/null +++ b/pkg/notify/rpc/apis/send_client.go @@ -0,0 +1,29 @@ +// 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 apis + +import "google.golang.org/grpc" + +type SendNotificationClient struct { + sendAgentClient + Conn *grpc.ClientConn +} + +func NewSendNotificationClient(cc *grpc.ClientConn) *SendNotificationClient { + return &SendNotificationClient{ + sendAgentClient: sendAgentClient{cc}, + Conn: cc, + } +} diff --git a/pkg/notify/rpc/apis/send_server.pb.go b/pkg/notify/rpc/apis/send_server.pb.go new file mode 100644 index 0000000000..358a6b1965 --- /dev/null +++ b/pkg/notify/rpc/apis/send_server.pb.go @@ -0,0 +1,438 @@ +// 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. + +// Code generated by protoc-gen-go. DO NOT EDIT. +// source: send_server.proto + +package apis + +import ( + context "context" + fmt "fmt" + proto "github.com/golang/protobuf/proto" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" + math "math" +) + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package + +type SendParams struct { + Contact string `protobuf:"bytes,1,opt,name=contact,proto3" json:"contact,omitempty"` + Topic string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"` + Message string `protobuf:"bytes,3,opt,name=message,proto3" json:"message,omitempty"` + Priority string `protobuf:"bytes,4,opt,name=Priority,proto3" json:"Priority,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *SendParams) Reset() { *m = SendParams{} } +func (m *SendParams) String() string { return proto.CompactTextString(m) } +func (*SendParams) ProtoMessage() {} +func (*SendParams) Descriptor() ([]byte, []int) { + return fileDescriptor_63fdd68f7eb311f9, []int{0} +} + +func (m *SendParams) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_SendParams.Unmarshal(m, b) +} +func (m *SendParams) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_SendParams.Marshal(b, m, deterministic) +} +func (m *SendParams) XXX_Merge(src proto.Message) { + xxx_messageInfo_SendParams.Merge(m, src) +} +func (m *SendParams) XXX_Size() int { + return xxx_messageInfo_SendParams.Size(m) +} +func (m *SendParams) XXX_DiscardUnknown() { + xxx_messageInfo_SendParams.DiscardUnknown(m) +} + +var xxx_messageInfo_SendParams proto.InternalMessageInfo + +func (m *SendParams) GetContact() string { + if m != nil { + return m.Contact + } + return "" +} + +func (m *SendParams) GetTopic() string { + if m != nil { + return m.Topic + } + return "" +} + +func (m *SendParams) GetMessage() string { + if m != nil { + return m.Message + } + return "" +} + +func (m *SendParams) GetPriority() string { + if m != nil { + return m.Priority + } + return "" +} + +type UpdateConfigParams struct { + Configs map[string]string `protobuf:"bytes,1,rep,name=configs,proto3" json:"configs,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *UpdateConfigParams) Reset() { *m = UpdateConfigParams{} } +func (m *UpdateConfigParams) String() string { return proto.CompactTextString(m) } +func (*UpdateConfigParams) ProtoMessage() {} +func (*UpdateConfigParams) Descriptor() ([]byte, []int) { + return fileDescriptor_63fdd68f7eb311f9, []int{1} +} + +func (m *UpdateConfigParams) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_UpdateConfigParams.Unmarshal(m, b) +} +func (m *UpdateConfigParams) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_UpdateConfigParams.Marshal(b, m, deterministic) +} +func (m *UpdateConfigParams) XXX_Merge(src proto.Message) { + xxx_messageInfo_UpdateConfigParams.Merge(m, src) +} +func (m *UpdateConfigParams) XXX_Size() int { + return xxx_messageInfo_UpdateConfigParams.Size(m) +} +func (m *UpdateConfigParams) XXX_DiscardUnknown() { + xxx_messageInfo_UpdateConfigParams.DiscardUnknown(m) +} + +var xxx_messageInfo_UpdateConfigParams proto.InternalMessageInfo + +func (m *UpdateConfigParams) GetConfigs() map[string]string { + if m != nil { + return m.Configs + } + return nil +} + +type UseridByMobileParams struct { + Mobile string `protobuf:"bytes,1,opt,name=mobile,proto3" json:"mobile,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *UseridByMobileParams) Reset() { *m = UseridByMobileParams{} } +func (m *UseridByMobileParams) String() string { return proto.CompactTextString(m) } +func (*UseridByMobileParams) ProtoMessage() {} +func (*UseridByMobileParams) Descriptor() ([]byte, []int) { + return fileDescriptor_63fdd68f7eb311f9, []int{2} +} + +func (m *UseridByMobileParams) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_UseridByMobileParams.Unmarshal(m, b) +} +func (m *UseridByMobileParams) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_UseridByMobileParams.Marshal(b, m, deterministic) +} +func (m *UseridByMobileParams) XXX_Merge(src proto.Message) { + xxx_messageInfo_UseridByMobileParams.Merge(m, src) +} +func (m *UseridByMobileParams) XXX_Size() int { + return xxx_messageInfo_UseridByMobileParams.Size(m) +} +func (m *UseridByMobileParams) XXX_DiscardUnknown() { + xxx_messageInfo_UseridByMobileParams.DiscardUnknown(m) +} + +var xxx_messageInfo_UseridByMobileParams proto.InternalMessageInfo + +func (m *UseridByMobileParams) GetMobile() string { + if m != nil { + return m.Mobile + } + return "" +} + +type Empty struct { + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Empty) Reset() { *m = Empty{} } +func (m *Empty) String() string { return proto.CompactTextString(m) } +func (*Empty) ProtoMessage() {} +func (*Empty) Descriptor() ([]byte, []int) { + return fileDescriptor_63fdd68f7eb311f9, []int{3} +} + +func (m *Empty) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Empty.Unmarshal(m, b) +} +func (m *Empty) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Empty.Marshal(b, m, deterministic) +} +func (m *Empty) XXX_Merge(src proto.Message) { + xxx_messageInfo_Empty.Merge(m, src) +} +func (m *Empty) XXX_Size() int { + return xxx_messageInfo_Empty.Size(m) +} +func (m *Empty) XXX_DiscardUnknown() { + xxx_messageInfo_Empty.DiscardUnknown(m) +} + +var xxx_messageInfo_Empty proto.InternalMessageInfo + +type UseridByMobileReply struct { + Userid string `protobuf:"bytes,1,opt,name=userid,proto3" json:"userid,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *UseridByMobileReply) Reset() { *m = UseridByMobileReply{} } +func (m *UseridByMobileReply) String() string { return proto.CompactTextString(m) } +func (*UseridByMobileReply) ProtoMessage() {} +func (*UseridByMobileReply) Descriptor() ([]byte, []int) { + return fileDescriptor_63fdd68f7eb311f9, []int{4} +} + +func (m *UseridByMobileReply) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_UseridByMobileReply.Unmarshal(m, b) +} +func (m *UseridByMobileReply) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_UseridByMobileReply.Marshal(b, m, deterministic) +} +func (m *UseridByMobileReply) XXX_Merge(src proto.Message) { + xxx_messageInfo_UseridByMobileReply.Merge(m, src) +} +func (m *UseridByMobileReply) XXX_Size() int { + return xxx_messageInfo_UseridByMobileReply.Size(m) +} +func (m *UseridByMobileReply) XXX_DiscardUnknown() { + xxx_messageInfo_UseridByMobileReply.DiscardUnknown(m) +} + +var xxx_messageInfo_UseridByMobileReply proto.InternalMessageInfo + +func (m *UseridByMobileReply) GetUserid() string { + if m != nil { + return m.Userid + } + return "" +} + +func init() { + proto.RegisterType((*SendParams)(nil), "apis.SendParams") + proto.RegisterType((*UpdateConfigParams)(nil), "apis.UpdateConfigParams") + proto.RegisterMapType((map[string]string)(nil), "apis.UpdateConfigParams.ConfigsEntry") + proto.RegisterType((*UseridByMobileParams)(nil), "apis.UseridByMobileParams") + proto.RegisterType((*Empty)(nil), "apis.Empty") + proto.RegisterType((*UseridByMobileReply)(nil), "apis.UseridByMobileReply") +} + +func init() { proto.RegisterFile("send_server.proto", fileDescriptor_63fdd68f7eb311f9) } + +var fileDescriptor_63fdd68f7eb311f9 = []byte{ + // 325 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x92, 0xcd, 0x4a, 0x03, 0x31, + 0x10, 0xc7, 0xd9, 0x7e, 0xda, 0x69, 0x91, 0x1a, 0x8b, 0xc4, 0x3d, 0x95, 0x85, 0x42, 0x2f, 0xee, + 0xa1, 0x22, 0x48, 0x2f, 0xa2, 0x52, 0x3c, 0x09, 0xa5, 0xd2, 0xb3, 0xa4, 0xdd, 0xe9, 0x12, 0xec, + 0x6e, 0x96, 0x24, 0x2d, 0xec, 0x63, 0xf8, 0x2a, 0x3e, 0xa1, 0xe4, 0x63, 0xb5, 0xc5, 0x7a, 0xcb, + 0x6f, 0x66, 0xfe, 0x99, 0xf9, 0x4f, 0x02, 0x17, 0x0a, 0xf3, 0xe4, 0x5d, 0xa1, 0xdc, 0xa3, 0x8c, + 0x0b, 0x29, 0xb4, 0x20, 0x0d, 0x56, 0x70, 0x15, 0x49, 0x80, 0x37, 0xcc, 0x93, 0x39, 0x93, 0x2c, + 0x53, 0x84, 0x42, 0x7b, 0x2d, 0x72, 0xcd, 0xd6, 0x9a, 0x06, 0xc3, 0x60, 0xdc, 0x59, 0x54, 0x48, + 0x06, 0xd0, 0xd4, 0xa2, 0xe0, 0x6b, 0x5a, 0xb3, 0x71, 0x07, 0xa6, 0x3e, 0x43, 0xa5, 0x58, 0x8a, + 0xb4, 0xee, 0xea, 0x3d, 0x92, 0x10, 0xce, 0xe6, 0x92, 0x0b, 0xc9, 0x75, 0x49, 0x1b, 0x36, 0xf5, + 0xc3, 0xd1, 0x67, 0x00, 0x64, 0x59, 0x24, 0x4c, 0xe3, 0xb3, 0xc8, 0x37, 0x3c, 0xf5, 0xcd, 0x1f, + 0x6c, 0xf3, 0x0d, 0x4f, 0x15, 0x0d, 0x86, 0xf5, 0x71, 0x77, 0x32, 0x8a, 0xcd, 0x88, 0xf1, 0xdf, + 0xd2, 0xd8, 0x81, 0x9a, 0xe5, 0x5a, 0x96, 0x8b, 0x4a, 0x15, 0x4e, 0xa1, 0x77, 0x98, 0x20, 0x7d, + 0xa8, 0x7f, 0x60, 0xe9, 0x9d, 0x98, 0xa3, 0x71, 0xb1, 0x67, 0xdb, 0x1d, 0x56, 0x2e, 0x2c, 0x4c, + 0x6b, 0xf7, 0x41, 0x14, 0xc3, 0x60, 0xa9, 0x50, 0xf2, 0xe4, 0xa9, 0x7c, 0x15, 0x2b, 0xbe, 0x45, + 0x3f, 0xd4, 0x15, 0xb4, 0x32, 0xcb, 0xfe, 0x1a, 0x4f, 0x51, 0x1b, 0x9a, 0xb3, 0xac, 0xd0, 0x65, + 0x74, 0x03, 0x97, 0xc7, 0xc2, 0x05, 0x16, 0xdb, 0xd2, 0xe8, 0x76, 0x36, 0x5c, 0xe9, 0x1c, 0x4d, + 0xbe, 0x02, 0xe8, 0x98, 0x85, 0x3f, 0xa6, 0x98, 0x6b, 0x32, 0x82, 0x86, 0x01, 0xd2, 0x77, 0x4e, + 0x7f, 0x5f, 0x22, 0xec, 0xba, 0x88, 0xed, 0x41, 0xee, 0xa0, 0x77, 0xb8, 0x04, 0x42, 0xff, 0x5b, + 0xcc, 0xb1, 0xec, 0x05, 0xce, 0x8f, 0x47, 0x23, 0xa1, 0x17, 0x9e, 0x70, 0x1a, 0x5e, 0x9f, 0xca, + 0x59, 0x33, 0xab, 0x96, 0xfd, 0x31, 0xb7, 0xdf, 0x01, 0x00, 0x00, 0xff, 0xff, 0x49, 0x12, 0x30, + 0x1e, 0x46, 0x02, 0x00, 0x00, +} + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ grpc.ClientConn + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +const _ = grpc.SupportPackageIsVersion4 + +// SendAgentClient is the client API for SendAgent service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream. +type SendAgentClient interface { + Send(ctx context.Context, in *SendParams, opts ...grpc.CallOption) (*Empty, error) + UpdateConfig(ctx context.Context, in *UpdateConfigParams, opts ...grpc.CallOption) (*Empty, error) + UseridByMobile(ctx context.Context, in *UseridByMobileParams, opts ...grpc.CallOption) (*UseridByMobileReply, error) +} + +type sendAgentClient struct { + cc *grpc.ClientConn +} + +func NewSendAgentClient(cc *grpc.ClientConn) SendAgentClient { + return &sendAgentClient{cc} +} + +func (c *sendAgentClient) Send(ctx context.Context, in *SendParams, opts ...grpc.CallOption) (*Empty, error) { + out := new(Empty) + err := c.cc.Invoke(ctx, "/apis.SendAgent/Send", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *sendAgentClient) UpdateConfig(ctx context.Context, in *UpdateConfigParams, opts ...grpc.CallOption) (*Empty, error) { + out := new(Empty) + err := c.cc.Invoke(ctx, "/apis.SendAgent/UpdateConfig", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *sendAgentClient) UseridByMobile(ctx context.Context, in *UseridByMobileParams, opts ...grpc.CallOption) (*UseridByMobileReply, error) { + out := new(UseridByMobileReply) + err := c.cc.Invoke(ctx, "/apis.SendAgent/UseridByMobile", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// SendAgentServer is the server API for SendAgent service. +type SendAgentServer interface { + Send(context.Context, *SendParams) (*Empty, error) + UpdateConfig(context.Context, *UpdateConfigParams) (*Empty, error) + UseridByMobile(context.Context, *UseridByMobileParams) (*UseridByMobileReply, error) +} + +// UnimplementedSendAgentServer can be embedded to have forward compatible implementations. +type UnimplementedSendAgentServer struct { +} + +func (*UnimplementedSendAgentServer) Send(ctx context.Context, req *SendParams) (*Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method Send not implemented") +} +func (*UnimplementedSendAgentServer) UpdateConfig(ctx context.Context, req *UpdateConfigParams) (*Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method UpdateConfig not implemented") +} +func (*UnimplementedSendAgentServer) UseridByMobile(ctx context.Context, req *UseridByMobileParams) (*UseridByMobileReply, error) { + return nil, status.Errorf(codes.Unimplemented, "method UseridByMobile not implemented") +} + +func RegisterSendAgentServer(s *grpc.Server, srv SendAgentServer) { + s.RegisterService(&_SendAgent_serviceDesc, srv) +} + +func _SendAgent_Send_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(SendParams) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SendAgentServer).Send(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/apis.SendAgent/Send", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SendAgentServer).Send(ctx, req.(*SendParams)) + } + return interceptor(ctx, in, info, handler) +} + +func _SendAgent_UpdateConfig_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(UpdateConfigParams) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SendAgentServer).UpdateConfig(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/apis.SendAgent/UpdateConfig", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SendAgentServer).UpdateConfig(ctx, req.(*UpdateConfigParams)) + } + return interceptor(ctx, in, info, handler) +} + +func _SendAgent_UseridByMobile_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(UseridByMobileParams) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SendAgentServer).UseridByMobile(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/apis.SendAgent/UseridByMobile", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SendAgentServer).UseridByMobile(ctx, req.(*UseridByMobileParams)) + } + return interceptor(ctx, in, info, handler) +} + +var _SendAgent_serviceDesc = grpc.ServiceDesc{ + ServiceName: "apis.SendAgent", + HandlerType: (*SendAgentServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "Send", + Handler: _SendAgent_Send_Handler, + }, + { + MethodName: "UpdateConfig", + Handler: _SendAgent_UpdateConfig_Handler, + }, + { + MethodName: "UseridByMobile", + Handler: _SendAgent_UseridByMobile_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "send_server.proto", +} diff --git a/pkg/notify/rpc/apis/send_server.proto b/pkg/notify/rpc/apis/send_server.proto new file mode 100644 index 0000000000..e2060ecfc8 --- /dev/null +++ b/pkg/notify/rpc/apis/send_server.proto @@ -0,0 +1,31 @@ +syntax = "proto3"; + +package apis; + +message SendParams { + string contact = 1; + string topic = 2; + string message = 3; + string Priority = 4; +} + +message UpdateConfigParams { + map configs = 1; +} + +message UseridByMobileParams { + string mobile = 1; +} + +message Empty { +} + +message UseridByMobileReply { + string userid = 1; +} + +service SendAgent { + rpc Send(SendParams) returns (Empty); + rpc UpdateConfig(UpdateConfigParams) returns (Empty); + rpc UseridByMobile(UseridByMobileParams) returns (UseridByMobileReply); +} \ No newline at end of file diff --git a/pkg/notify/rpc/doc.go b/pkg/notify/rpc/doc.go new file mode 100644 index 0000000000..1e79bce4fb --- /dev/null +++ b/pkg/notify/rpc/doc.go @@ -0,0 +1 @@ +package rpc // import "yunion.io/x/onecloud/pkg/notify/rpc" diff --git a/pkg/notify/rpc/send.go b/pkg/notify/rpc/send.go new file mode 100644 index 0000000000..065f662a84 --- /dev/null +++ b/pkg/notify/rpc/send.go @@ -0,0 +1,332 @@ +// 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 rpc + +import ( + "context" + "fmt" + "io/ioutil" + "net" + "os" + "path/filepath" + "strings" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + + "yunion.io/x/onecloud/pkg/mcclient" + _interface "yunion.io/x/onecloud/pkg/notify/interface" + "yunion.io/x/onecloud/pkg/notify/models" + "yunion.io/x/onecloud/pkg/notify/rpc/apis" + "yunion.io/x/onecloud/pkg/util/fileutils2" +) + +const ( + // ErrSendServiceNotFound means SRpcService's SendSerivces hasn't this Send Service. + ErrSendServiceNotFound = errors.Error("No such send service") + ErrSendServiceNotInit = errors.Error("Send service hasn't been init") +) + +// SRpcService provide rpc service about sending message for notify module and manage these services. +// SendServices storage all send service. +type SRpcService struct { + SendServices *ServiceMap + socketFileDir string + configStore _interface.IServiceConfigStore +} + +// NewSRpcService create a SRpcService +func NewSRpcService(socketFileDir string, configStore _interface.IServiceConfigStore) *SRpcService { + return &SRpcService{ + SendServices: NewServiceMap(), + socketFileDir: socketFileDir, + configStore: configStore, + } +} + +// InitAll init all Send Services, the init process is that: +// find all socket file in directory 'self.socketFileDir', if wrong return error; +// the name of file is the service's name; then try to dial to this rpc service +// through corresponding socket file, if failed, only print log but not return error. +func (self *SRpcService) InitAll() error { + files, err := ioutil.ReadDir(self.socketFileDir) + if err != nil { + return errors.Wrapf(err, "read dir %s failed", self.socketFileDir) + } + ctx := context.Background() + for _, file := range files { + filename := file.Name() + if !file.IsDir() && strings.Contains(filename, ".sock") { + serviceName := filename[:len(filename)-5] + self.startNewService(ctx, serviceName) + } + } + if self.SendServices.Len() == 0 { + log.Errorf("No available send service.") + } else { + log.Infof("Total %d send service init successful", self.SendServices.Len()) + } + return nil +} + +// UpdateServices will detect the self.sockFileDir, add new service and +// delete disappeared one from self.SendServices. +func (self *SRpcService) UpdateServices(ctx context.Context, usreCred mcclient.TokenCredential, isStart bool) { + err := self.updateService(ctx) + if err != nil { + log.Errorf("update services failed because that %s.", err.Error()) + } +} + +// StopAll stop all send service in self.SenderServices normally which can delete the socket file. +func (self *SRpcService) StopAll() { + f := func(client *apis.SendNotificationClient) { + client.Conn.Close() + } + self.SendServices.Map(f) +} + +// Send call the corresponding rpc server to send messager. +func (self *SRpcService) Send(ctx context.Context, contactType, contact, topic, msg, priority string) error { + + args := apis.SendParams{ + Contact: contact, + Topic: topic, + Message: msg, + Priority: priority, + } + + f := func(service *apis.SendNotificationClient) (interface{}, error) { + log.Debugf("send one") + return service.Send(ctx, &args) + } + + _, err := self.execute(ctx, f, contactType) + return err +} + +// RestartService can restart remote rpc server and pass config info. +// This function should be call immediately after init notify server firstly +// This function should be call immediately after accept the request about changing config. +func (self *SRpcService) RestartService(ctx context.Context, config _interface.SConfig, serviceName string) { + _, err := self.restartWithConfig(ctx, serviceName, config) + if err != nil { + log.Debugf("restart service failed: %s", err) + } +} + +func (self *SRpcService) ContactByMobile(ctx context.Context, mobile, serviceName string) (string, error) { + + args := apis.UseridByMobileParams{} + args.Mobile = mobile + + f := func(service *apis.SendNotificationClient) (interface{}, error) { + return service.UseridByMobile(ctx, &args) + } + + ret, err := self.execute(ctx, f, serviceName) + if err != nil { + return "", err + } + + reply := ret.(*apis.UseridByMobileReply) + return reply.Userid, nil +} + +// Wrap function to execute function call rpc server +func (self *SRpcService) execute(ctx context.Context, f func(client *apis.SendNotificationClient) (interface{}, error), + serviceName string) (interface{}, error) { + + sendService, ok := self.SendServices.Get(serviceName) + + log.Debugf("get service %s", serviceName) + var err error + if !ok { + log.Debugf("get service first time failed") + sendService, err = self.startNewService(ctx, serviceName) + + if err != nil { + return nil, errors.Wrap(err, "start new service failed") + } + } + + ret, err := f(sendService) + + if err != nil { + // hander error + st := status.Convert(err) + if st.Code() == codes.Unavailable { + // sock is bad + self.closeService(ctx, serviceName) + return nil, ErrSendServiceNotFound + } + + if st.Message() != ErrSendServiceNotInit.Error() { + return nil, errors.Error(fmt.Sprintf("Send message failed because that %s.", st.Message())) + } + + // if NOINIT, try to restart server and send again + sendService, err = self.restartService(ctx, serviceName) + if err != nil { + return nil, errors.Wrapf(err, "restart service %s failed", serviceName) + } + + _, err := f(sendService) + if err != nil { + st := status.Convert(err) + if st.Code() == codes.Unavailable { + // sock is bad + self.closeService(ctx, serviceName) + + return nil, errors.Wrap(ErrSendServiceNotFound, serviceName) + } + return nil, errors.Error(fmt.Sprintf("Send message failed because that %s.", st.Message())) + } + } + return ret, nil +} + +// restartSrevice fetch config from IServiceConfigStore and Call rpc.UpdateConfig +func (self *SRpcService) restartService(ctx context.Context, serviceName string) (*apis.SendNotificationClient, error) { + + config, err := self.configStore.GetConfig(serviceName) + if err != nil { + log.Debugf("getConfig of serveice %s from database error", serviceName) + return nil, models.ErrGetConfig + } + return self.restartWithConfig(ctx, serviceName, config) +} + +func (self *SRpcService) restartWithConfig(ctx context.Context, serviceName string, + config map[string]string) (*apis.SendNotificationClient, error) { + + var ( + sendService *apis.SendNotificationClient + err error + ) + + sendService, ok := self.SendServices.Get(serviceName) + + if !ok { + return nil, fmt.Errorf("no such service, please start new service") + } + + args := apis.UpdateConfigParams{} + args.Configs = config + _, err = sendService.UpdateConfig(ctx, &args) + if err != nil { + st := status.Convert(err) + return nil, errors.Error(st.Message()) + } + + return sendService, nil +} + +// startNewService try to start a new rpc service named serviceName +func (self *SRpcService) startNewService(ctx context.Context, serviceName string) (*apis.SendNotificationClient, error) { + + var ( + sendService *apis.SendNotificationClient + err error + ) + + filename := filepath.Join(self.socketFileDir, serviceName+".sock") + if !fileutils2.Exists(filename) { + return nil, err + } + + grpcConn, err := grpcDialWithUnixSocket(ctx, filename) + if err != nil { + return nil, err + } + sendService = apis.NewSendNotificationClient(grpcConn) + + self.SendServices.Set(sendService, serviceName) + + // get config + config, err := self.configStore.GetConfig(serviceName) + if err != nil { + log.Errorf("getConfig of serveice %s from database error", serviceName) + return nil, models.ErrGetConfig + } + + // update config for service + args := apis.UpdateConfigParams{} + args.Configs = config + _, err = sendService.UpdateConfig(ctx, &args) + if err != nil { + st := status.Convert(err) + if st.Code() == codes.Unavailable { + // no such rpc serve + os.Remove(filename) + return nil, fmt.Errorf("no such rpc serve") + } + return nil, fmt.Errorf(st.Message()) + } + + return sendService, nil +} + +// closeService will remove service record from self.SendServices and try to remove sock file +func (self *SRpcService) closeService(ctx context.Context, serviceName string) { + filename := filepath.Join(self.socketFileDir, serviceName+".sock") + self.SendServices.Remove(serviceName) + os.Remove(filename) +} + +func (self *SRpcService) updateService(ctx context.Context) error { + files, err := ioutil.ReadDir(self.socketFileDir) + if err != nil { + return errors.Wrapf(err, "read dir %s failed", self.socketFileDir) + } + + serviceNames := self.SendServices.ServiceNames() + serviceNameSet := make(map[string]struct{}) + for _, name := range serviceNames { + serviceNameSet[name] = struct{}{} + } + + for _, file := range files { + filename := file.Name() + if !file.IsDir() && strings.Contains(filename, ".sock") { + serviceName := filename[:len(filename)-5] + if self.SendServices.IsExist(serviceName) { + delete(serviceNameSet, serviceName) + continue + } + self.startNewService(ctx, serviceName) + } + } + + serviceNames = serviceNames[:0] + for serviceName := range serviceNameSet { + serviceNames = append(serviceNames, serviceName) + } + + self.SendServices.BatchRemove(serviceNames) + return nil +} + +func grpcDialWithUnixSocket(ctx context.Context, socketPath string) (*grpc.ClientConn, error) { + return grpc.DialContext(ctx, socketPath, grpc.WithInsecure(), grpc.WithTimeout(time.Second*5), grpc.WithDialer( + func(addr string, timeout time.Duration) (net.Conn, error) { + return net.DialTimeout("unix", addr, timeout) + }), + ) +} diff --git a/pkg/notify/rpc/service_map.go b/pkg/notify/rpc/service_map.go new file mode 100644 index 0000000000..951d076e53 --- /dev/null +++ b/pkg/notify/rpc/service_map.go @@ -0,0 +1,97 @@ +// 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 rpc + +import ( + "sync" + + "yunion.io/x/onecloud/pkg/notify/rpc/apis" +) + +// ServiceMap has a map of string and apis.SendNotification's pointer, and a RWMutex lock protect map. +type ServiceMap struct { + serviceMap map[string]*apis.SendNotificationClient + lock sync.RWMutex +} + +func NewServiceMap() *ServiceMap { + return &ServiceMap{serviceMap: make(map[string]*apis.SendNotificationClient)} +} + +func (sm *ServiceMap) Get(serviceName string) (*apis.SendNotificationClient, bool) { + sm.lock.RLock() + defer sm.lock.RUnlock() + client, ok := sm.serviceMap[serviceName] + return client, ok +} + +func (sm *ServiceMap) Set(service *apis.SendNotificationClient, serviceName string) { + sm.lock.Lock() + defer sm.lock.Unlock() + sm.serviceMap[serviceName] = service +} + +func (sm *ServiceMap) Remove(serviceName string) { + sm.lock.Lock() + defer sm.lock.Unlock() + service, ok := sm.serviceMap[serviceName] + if !ok { + return + } + service.Conn.Close() + delete(sm.serviceMap, serviceName) +} + +func (sm *ServiceMap) BatchRemove(serviceNames []string) { + sm.lock.Lock() + defer sm.lock.Unlock() + for _, serviceName := range serviceNames { + service, ok := sm.serviceMap[serviceName] + if !ok { + continue + } + service.Conn.Close() + delete(sm.serviceMap, serviceName) + } +} + +func (sm *ServiceMap) ServiceNames() []string { + sm.lock.RLock() + defer sm.lock.RUnlock() + serviceNames := make([]string, 0, len(sm.serviceMap)) + for serviceName := range sm.serviceMap { + serviceNames = append(serviceNames, serviceName) + } + return serviceNames +} + +func (sm *ServiceMap) IsExist(serviceName string) bool { + sm.lock.RLock() + defer sm.lock.RUnlock() + _, ok := sm.serviceMap[serviceName] + return ok +} + +func (sm *ServiceMap) Len() int { + return len(sm.serviceMap) +} + +func (sm *ServiceMap) Map(f func(*apis.SendNotificationClient)) { + sm.lock.Lock() + defer sm.lock.Unlock() + for _, service := range sm.serviceMap { + f(service) + } +} diff --git a/pkg/notify/service.go b/pkg/notify/service.go index 9aa05ba353..43537916ac 100644 --- a/pkg/notify/service.go +++ b/pkg/notify/service.go @@ -15,18 +15,23 @@ package notify import ( + "context" "os" + "time" _ "github.com/go-sql-driver/mysql" "yunion.io/x/log" "yunion.io/x/onecloud/pkg/cloudcommon" "yunion.io/x/onecloud/pkg/cloudcommon/app" + "yunion.io/x/onecloud/pkg/cloudcommon/cronman" "yunion.io/x/onecloud/pkg/cloudcommon/db" common_options "yunion.io/x/onecloud/pkg/cloudcommon/options" + "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/notify/cache" "yunion.io/x/onecloud/pkg/notify/models" "yunion.io/x/onecloud/pkg/notify/options" - "yunion.io/x/onecloud/pkg/notify/utils" + "yunion.io/x/onecloud/pkg/notify/rpc" ) func StartService() { @@ -42,9 +47,6 @@ func StartService() { log.Infof("Auth complete!") }) - // Session for user manager in keystone - utils.InitSession(commonOpts) - // init handler applicaion := app.InitApp(baseOpts, true) InitHandlers(applicaion) @@ -53,14 +55,23 @@ func StartService() { db.EnsureAppInitSyncDB(applicaion, dbOpts, models.InitDB) defer cloudcommon.CloseDB() - // init rpc service - models.RpcService = models.NewSRpcService(opts.SocketFileDir) - models.RpcService.InitAll() - defer models.RpcService.StopAll() - go models.RpcService.UpdateServices(opts.UpdateInterval) + // init cache + cache.RegistUserCredCacheUpdater() - // start ReSend service - go models.ReSend(opts.ReSendScope) + // init notify service + models.NotifyService = rpc.NewSRpcService(opts.SocketFileDir, models.ConfigManager) + models.NotifyService.InitAll() + defer models.NotifyService.StopAll() + + cron := cronman.GetCronJobManager(true) + // update service + cron.AddJobAtIntervals("UpdateServices", time.Duration(opts.UpdateInterval)*time.Second, models.NotifyService.UpdateServices) + + // resend notifications + resend := func(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) { + models.ReSend(opts.ReSendScope) + } + cron.AddJobAtIntervals("ReSendNotifications", time.Duration(opts.ReSendScope)*time.Minute, resend) app.ServeForever(applicaion, baseOpts) } diff --git a/pkg/notify/utils/keystone.go b/pkg/notify/utils/keystone.go index ace6c416af..d72c0703d1 100644 --- a/pkg/notify/utils/keystone.go +++ b/pkg/notify/utils/keystone.go @@ -17,43 +17,41 @@ package utils import ( "context" - "yunion.io/x/jsonutils" - - "yunion.io/x/onecloud/pkg/cloudcommon/options" - "yunion.io/x/onecloud/pkg/mcclient" - "yunion.io/x/onecloud/pkg/mcclient/auth" - "yunion.io/x/onecloud/pkg/mcclient/modules" + "yunion.io/x/onecloud/pkg/notify/cache" ) -var ( - session *mcclient.ClientSession -) - -func InitSession(options *options.CommonOptions) { - session = auth.GetAdminSession(context.Background(), options.Region, "v3") +func GetUserByID(ctx context.Context, id string) (*cache.SUser, error) { + return cache.UserCacheManager.FetchUserByID(ctx, id, false) } -func GetUserByID(id string) (jsonutils.JSONObject, error) { - return modules.UsersV3.Get(session, id, jsonutils.NewDict()) -} - -func GetUsersByGroupID(gid string) ([]string, error) { - ret, err := modules.Groups.GetUsers(session, gid) +func GetUserIdsLikeName(ctx context.Context, name string) ([]string, error) { + users, err := cache.UserCacheManager.FetchUserLikeName(ctx, name, true) if err != nil { return nil, err } - ids := make([]string, len(ret.Data)) - for i := range ret.Data { - ids[i], _ = ret.Data[i].GetString("id") + ret := make([]string, len(users)) + for i := range users { + ret[i] = users[i].Id + } + return ret, nil +} + +func GetUsersByGroupID(ctx context.Context, gid string) ([]string, error) { + ret, err := cache.UserGroupCacheManager.FetchByGroupId(ctx, gid) + if err != nil { + return nil, err + } + ids := make([]string, len(ret)) + for i := range ret { + ids[i] = ret[i].UserId } return ids, nil } -func GetUsernameByID(id string) (string, error) { - user, err := GetUserByID(id) +func GetUsernameByID(ctx context.Context, id string) (string, error) { + user, err := GetUserByID(ctx, id) if err != nil { return "", err } - name, _ := user.GetString("name") - return name, nil + return user.Name, nil } diff --git a/pkg/notify/utils/others.go b/pkg/notify/utils/others.go index 12098d7177..db07d4448b 100644 --- a/pkg/notify/utils/others.go +++ b/pkg/notify/utils/others.go @@ -58,3 +58,11 @@ func GenerateEmailToken(tokenLen int) string { } return token.String()[:tokenLen] } + +func JsonArrayToStringArray(src []jsonutils.JSONObject) []string { + des := make([]string, len(src)) + for i := range src { + des[i], _ = src[i].GetString() + } + return des +}