fix(cloudcommon): notify event of specific resources (#15091)

pkg/cloudcommon/db/db_dispatcher.go

@@ -1684,9 +1684,11 @@ func updateItem(manager IModelManager, item IModel, ctx context.Context, userCre
 	if len(diff) > 0 {
 		OpsLog.LogEvent(item, ACT_UPDATE, diff, userCred)
 		logclient.AddActionLogWithContext(ctx, item, logclient.ACT_UPDATE, diff, userCred, true)
-		item.PostUpdate(ctx, userCred, query, data)
+		CallUpdateNotifyHook(ctx, userCred, item)
 	}
 
+	item.PostUpdate(ctx, userCred, query, data)
+
 	return getItemDetails(manager, item, ctx, userCred, query)
 }
 

pkg/cloudcommon/db/notify_hook.go

@@ -0,0 +1,41 @@
+// 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 db
+
+import (
+	"context"
+
+	"yunion.io/x/onecloud/pkg/mcclient"
+)
+
+var (
+	updateNotifyHook updateNotifyHookFunc
+)
+
+type updateNotifyHookFunc func(ctx context.Context, userCred mcclient.TokenCredential, obj IModel)
+
+func SetUpdateNotifyHook(f updateNotifyHookFunc) {
+	if updateNotifyHook != nil {
+		panic("updateNotifyHook already set")
+	}
+	updateNotifyHook = f
+}
+
+func CallUpdateNotifyHook(ctx context.Context, userCred mcclient.TokenCredential, obj IModel) {
+	if updateNotifyHook == nil {
+		return
+	}
+	updateNotifyHook(ctx, userCred, obj)
+}

pkg/cloudcommon/notifyclient/notify.go

@@ -17,6 +17,7 @@ package notifyclient
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"yunion.io/x/jsonutils"
@@ -36,10 +37,30 @@ var (
 
 	notifyAdminUsers  []string
 	notifyAdminGroups []string
+
+	notifyDBHookResources sync.Map
 )
 
 func init() {
 	notifyClientWorkerMan = appsrv.NewWorkerManager("NotifyClientWorkerManager", 1, 50, false)
+
+	// set db notify hook
+	db.SetUpdateNotifyHook(func(ctx context.Context, userCred mcclient.TokenCredential, obj db.IModel) {
+		_, ok := notifyDBHookResources.Load(obj.KeywordPlural())
+		if !ok {
+			return
+		}
+		EventNotify(ctx, userCred, SEventNotifyParam{
+			Obj:    obj,
+			Action: ActionUpdate,
+		})
+	})
+}
+
+func AddNotifyDBHookResources(keywordPlurals ...string) {
+	for _, kp := range keywordPlurals {
+		notifyDBHookResources.Store(kp, true)
+	}
 }
 
 func NotifyWithCtx(ctx context.Context, recipientId []string, isGroup bool, priority npk.TNotifyPriority, event string, data jsonutils.JSONObject) {

pkg/compute/models/dbinstances.go

@@ -74,6 +74,7 @@ func init() {
 		),
 	}
 	DBInstanceManager.SetVirtualObject(DBInstanceManager)
+	notifyclient.AddNotifyDBHookResources(DBInstanceManager.KeywordPlural())
 }
 
 type SDBInstance struct {
@@ -934,7 +935,7 @@ func (self *SDBInstance) PerformReboot(ctx context.Context, userCred mcclient.To
 	return nil, self.StartDBInstanceRebootTask(ctx, userCred, jsonutils.NewDict(), "")
 }
 
-//同步RDS实例状态
+// 同步RDS实例状态
 func (self *SDBInstance) PerformSyncstatus(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
 	return self.PerformSync(ctx, userCred, query, data)
 }
@@ -2114,8 +2115,4 @@ func (self *SDBInstance) GetExpiredAt() time.Time {
 
 func (db *SDBInstance) PostUpdate(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) {
 	db.SVirtualResourceBase.PostUpdate(ctx, userCred, query, data)
-	notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
-		Obj:    db,
-		Action: notifyclient.ActionUpdate,
-	})
 }

pkg/compute/models/elastic_search.go

@@ -61,6 +61,7 @@ func init() {
 		),
 	}
 	ElasticSearchManager.SetVirtualObject(ElasticSearchManager)
+	notifyclient.AddNotifyDBHookResources(ElasticSearchManager.KeywordPlural())
 }
 
 type SElasticSearch struct {
@@ -631,7 +632,7 @@ func (manager *SElasticSearchManager) ListItemExportKeys(ctx context.Context,
 	return q, nil
 }
 
-//同步ElasticSearch实例状态
+// 同步ElasticSearch实例状态
 func (self *SElasticSearch) PerformSyncstatus(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
 	var openTask = true
 	count, err := taskman.TaskManager.QueryTasksOfObject(self, time.Now().Add(-3*time.Minute), &openTask).CountWithError()
@@ -655,8 +656,4 @@ func (self *SElasticSearch) GetDetailsAccessInfo(ctx context.Context, userCred m
 
 func (es *SElasticSearch) PostUpdate(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) {
 	es.SVirtualResourceBase.PostUpdate(ctx, userCred, query, data)
-	notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
-		Obj:    es,
-		Action: notifyclient.ActionUpdate,
-	})
 }

pkg/compute/models/guests.go

@@ -101,6 +101,7 @@ func init() {
 	GuestManager.SetVirtualObject(GuestManager)
 	GuestManager.SetAlias("guest", "guests")
 	GuestManager.NameRequireAscii = false
+	notifyclient.AddNotifyDBHookResources(GuestManager.KeywordPlural(), GuestManager.AliasPlural())
 }
 
 type SGuest struct {
@@ -1888,10 +1889,6 @@ func (self *SGuest) PostUpdate(ctx context.Context, userCred mcclient.TokenCrede
 			log.Errorf("unable to set sshport for guest %s", self.GetId())
 		}
 	}
-	notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
-		Obj:    self,
-		Action: notifyclient.ActionUpdate,
-	})
 }
 
 func (manager *SGuestManager) checkCreateQuota(
@@ -2594,7 +2591,7 @@ func (self *SGuest) getAdminSecgroupName() string {
 	return ""
 }
 
-//获取多个安全组规则，优先级降序排序
+// 获取多个安全组规则，优先级降序排序
 func (self *SGuest) getSecurityGroupsRules() string {
 	secgroups, _ := self.GetSecgroups()
 	secgroupids := []string{}
@@ -3038,7 +3035,7 @@ func (self *SGuest) GetOSProfile() osprofile.SOSProfile {
 
 // Summary of network address allocation strategy
 //
-// IpAddr when specified must be part of the network
+// # IpAddr when specified must be part of the network
 //
 // Use IpAddr without checking if it's already allocated when UseDesignatedIP
 // is true.  See b31bc7fa ("feature: 1. baremetal server reuse host ip...")

pkg/keystone/models/users.go

@@ -62,6 +62,7 @@ func init() {
 	db.InitManager(func() {
 		UserManager.TableSpec().ColumnSpec("lang").SetDefault(options.Options.DefaultUserLanguage)
 	})
+	notifyclient.AddNotifyDBHookResources(UserManager.KeywordPlural())
 }
 
 /*
@@ -239,7 +240,7 @@ func (manager *SUserManager) initSysUser(ctx context.Context) error {
 }
 
 /*
- Fetch extended userinfo by Id or name + domainId or name + domainName
+Fetch extended userinfo by Id or name + domainId or name + domainName
 */
 func (manager *SUserManager) FetchUserExtended(userId, userName, domainId, domainName string) (*api.SUserExtended, error) {
 	if len(userId) == 0 && len(userName) == 0 {
@@ -779,10 +780,6 @@ func (user *SUser) PostUpdate(ctx context.Context, userCred mcclient.TokenCreden
 			log.Errorf("unable to clear failed auth: %v", err)
 		}
 	}
-	notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
-		Obj:    user,
-		Action: notifyclient.ActionUpdate,
-	})
 }
 
 func (user *SUser) ValidateDeleteCondition(ctx context.Context, info jsonutils.JSONObject) error {