fix: trigger project sync by informer (#19836)

Co-authored-by: Qiu Jian <qiujian@yunionyun.com>

pkg/appsrv/ring.go

@@ -78,3 +78,15 @@ func (r *Ring) Size() int {
 		return len(r.buffer) - r.tail + r.header
 	}
 }
+
+func (r *Ring) Range(proc func(obj interface{}) bool) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	for i := r.header; i != r.tail && i != r.header; i = nextPointer(i, len(r.buffer)) {
+		cont := proc(r.buffer[i])
+		if !cont {
+			break
+		}
+	}
+}

pkg/appsrv/workers.go

@@ -84,6 +84,24 @@ func (worker *SWorker) run() {
 		req := worker.manager.queue.Pop()
 		if req != nil {
 			task := req.(*sWorkerTask)
+
+			if worker.manager.cancelPrevIdent {
+				// cancel previous identical tasks
+				findIdent := false
+				worker.manager.queue.Range(func(iReq interface{}) bool {
+					iTask := iReq.(*sWorkerTask)
+					if iTask.task.Dump() == task.task.Dump() {
+						// found idential task
+						findIdent = true
+						return false
+					}
+					return true
+				})
+				if findIdent {
+					continue
+				}
+			}
+
 			if task.worker != nil {
 				task.worker <- worker
 			}
@@ -161,6 +179,8 @@ type SWorkerManager struct {
 	dbWorker       bool
 
 	ignoreOverflow bool
+
+	cancelPrevIdent bool
 }
 
 func NewWorkerManager(name string, workerCount int, backlog int, dbWorker bool) *SWorkerManager {
@@ -179,6 +199,8 @@ func NewWorkerManagerIgnoreOverflow(name string, workerCount int, backlog int, d
 		dbWorker:       dbWorker,
 
 		ignoreOverflow: ignoreOverflow,
+
+		cancelPrevIdent: false,
 	}
 
 	workerManagerLock.Lock()
@@ -200,6 +222,10 @@ type sWorkerTask struct {
 	start   time.Time
 }
 
+func (wm *SWorkerManager) EnableCancelPreviousIdenticalTask() {
+	wm.cancelPrevIdent = true
+}
+
 func (wm *SWorkerManager) UpdateWorkerCount(workerCount int) error {
 	wm.workerLock.Lock()
 	defer wm.workerLock.Unlock()

pkg/cloudcommon/app/endpoint.go

@@ -34,7 +34,7 @@ func newEndpointChangeManager() *SEndpointChangeManager {
 	return man
 }
 
-func (man *SEndpointChangeManager) DoSync(first bool) (time.Duration, error) {
+func (man *SEndpointChangeManager) DoSync(first bool, timeout bool) (time.Duration, error) {
 	// reauth to refresh endpoint list
 	auth.ReAuth()
 	return time.Hour * 2, nil

pkg/cloudcommon/db/cachesync/cachesync.go

@@ -0,0 +1,33 @@
+// 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 cachesync
+
+import (
+	"yunion.io/x/onecloud/pkg/appsrv"
+	identity_modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
+)
+
+var tenantCacheSyncWorkerMan *appsrv.SWorkerManager
+
+func init() {
+	tenantCacheSyncWorkerMan = appsrv.NewWorkerManagerIgnoreOverflow("tenant_cache_sync_worker", 1, 1, true, true)
+	// tenantCacheSyncWorkerMan.EnableCancelPreviousIdenticalTask()
+}
+
+func StartTenantCacheSync(intvalSeconds int) {
+	newResourceChangeManager(identity_modules.Projects, intvalSeconds)
+	newResourceChangeManager(identity_modules.Domains, intvalSeconds)
+	newResourceChangeManager(identity_modules.UsersV3, intvalSeconds)
+}

pkg/cloudcommon/db/cachesync/changeman.go

@@ -0,0 +1,109 @@
+// 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 cachesync
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/syncman/watcher"
+	"yunion.io/x/onecloud/pkg/mcclient/informer"
+	identity_modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
+)
+
+type SResourceChangeManager struct {
+	watcher.SInformerSyncManager
+
+	resMan          informer.IResourceManager
+	intervalSeconds int
+
+	ids     []string
+	idsLock *sync.Mutex
+}
+
+func newResourceChangeManager(resMan informer.IResourceManager, intvalSecs int) *SResourceChangeManager {
+	man := &SResourceChangeManager{
+		resMan:          resMan,
+		intervalSeconds: intvalSecs,
+
+		ids:     make([]string, 0),
+		idsLock: &sync.Mutex{},
+	}
+	man.InitSync(man)
+	man.FirstSync()
+	man.StartWatching(resMan)
+	return man
+}
+
+func (man *SResourceChangeManager) DoSync(first bool, timeout bool) (time.Duration, error) {
+	if first || timeout {
+		// reset id list
+		man.resetId()
+	} else {
+		log.Debugf("to do incremental sync ids %s", jsonutils.Marshal(man.ids))
+	}
+
+	switch man.resMan.KeyString() {
+	case identity_modules.Projects.KeywordPlural:
+		tenantCacheSyncWorkerMan.Run(&tenantCacheSyncWorker{
+			ids: man.ids,
+		}, nil, nil)
+	case identity_modules.Domains.KeywordPlural:
+		tenantCacheSyncWorkerMan.Run(&domainCacheSyncWorker{
+			ids: man.ids,
+		}, nil, nil)
+	case identity_modules.UsersV3.KeywordPlural:
+		tenantCacheSyncWorkerMan.Run(&userCacheSyncWorker{
+			ids: man.ids,
+		}, nil, nil)
+	}
+	man.resetId()
+	log.Debugf("sync DONE, next sync %d seconds later...", man.intervalSeconds*8)
+	return time.Second * time.Duration(man.intervalSeconds) * 8, nil
+}
+
+func (man *SResourceChangeManager) NeedSync(dat *jsonutils.JSONDict) bool {
+	if dat != nil && dat.Contains("id") {
+		idstr, _ := dat.GetString("id")
+		idstr = strings.TrimSpace(idstr)
+		if len(idstr) > 0 {
+			man.addId(idstr)
+		}
+	}
+	return true
+}
+
+func (man *SResourceChangeManager) addId(idstr string) {
+	man.idsLock.Lock()
+	defer man.idsLock.Unlock()
+
+	man.ids = append(man.ids, idstr)
+}
+
+func (man *SResourceChangeManager) resetId() {
+	man.idsLock.Lock()
+	defer man.idsLock.Unlock()
+
+	man.ids = man.ids[0:0]
+}
+
+func (man *SResourceChangeManager) Name() string {
+	return fmt.Sprintf("ResourceChangeManager:%s", man.resMan.GetKeyword())
+}

pkg/cloudcommon/db/cachesync/doc.go

@@ -0,0 +1,15 @@
+// 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 cachesync // import "yunion.io/x/onecloud/pkg/cloudcommon/db/cachesync"

pkg/cloudcommon/db/cachesync/domain.go

@@ -0,0 +1,87 @@
+// 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 cachesync
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/util/rbacscope"
+
+	identityapi "yunion.io/x/onecloud/pkg/apis/identity"
+	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
+	modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
+)
+
+type domainCacheSyncWorker struct {
+	ids []string
+}
+
+func (w *domainCacheSyncWorker) Run() {
+	log.Debugf("[domainCacheSyncWorker] Run domain cache sync worker ...")
+	err := syncDomains(context.Background(), w.ids)
+	if err != nil {
+		log.Errorf("fail to syncDomains %s", err)
+	}
+}
+
+func (w *domainCacheSyncWorker) Dump() string {
+	return "domainCacheSyncWorker"
+}
+
+func syncDomains(ctx context.Context, ids []string) error {
+	s := auth.GetAdminSession(ctx, consts.GetRegion())
+	query := jsonutils.NewDict()
+	query.Add(jsonutils.NewInt(1024), "limit")
+	query.Add(jsonutils.NewString(string(rbacscope.ScopeSystem)), "scope")
+	query.Add(jsonutils.JSONTrue, "details")
+	query.Add(jsonutils.NewString("all"), "pending_delete")
+	query.Add(jsonutils.NewString("all"), "delete")
+	if len(ids) > 0 {
+		query.Add(jsonutils.NewStringArray(ids), "id")
+	}
+	total := -1
+	offset := 0
+	for total < 0 || offset < total {
+		query.Set("offset", jsonutils.NewInt(int64(offset)))
+		results, err := modules.Domains.List(s, query)
+		if err != nil {
+			return errors.Wrap(err, "Domains.List")
+		}
+		total = results.Total
+		for i := range results.Data {
+			// update domain cache
+			item := db.SCachedTenant{}
+			deleted := jsonutils.QueryBoolean(results.Data[i], "deleted", false)
+			err := results.Data[i].Unmarshal(&item)
+			if err == nil && !deleted {
+				item.ProjectDomain = identityapi.KeystoneDomainRoot
+				item.DomainId = identityapi.KeystoneDomainRoot
+				db.TenantCacheManager.Save(ctx, item, true)
+			} else if deleted {
+				tenantObj, _ := db.TenantCacheManager.FetchById(item.Id)
+				if tenantObj != nil {
+					tenantObj.Delete(ctx, nil)
+				}
+			}
+			offset++
+		}
+	}
+	return nil
+}

pkg/cloudcommon/db/cachesync/project.go

@@ -0,0 +1,85 @@
+// 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 cachesync
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/util/rbacscope"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
+	modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
+)
+
+type tenantCacheSyncWorker struct {
+	ids []string
+}
+
+func (w *tenantCacheSyncWorker) Run() {
+	log.Debugf("[tenantCacheSyncWorker] Run project cache sync worker ...")
+	err := syncProjects(context.Background(), w.ids)
+	if err != nil {
+		log.Errorf("fail to syncProjects %s", err)
+	}
+}
+
+func (w *tenantCacheSyncWorker) Dump() string {
+	return "tenantCacheSyncWorker"
+}
+
+func syncProjects(ctx context.Context, ids []string) error {
+	s := auth.GetAdminSession(ctx, consts.GetRegion())
+	query := jsonutils.NewDict()
+	query.Add(jsonutils.NewInt(1024), "limit")
+	query.Add(jsonutils.NewString(string(rbacscope.ScopeSystem)), "scope")
+	query.Add(jsonutils.JSONTrue, "details")
+	query.Add(jsonutils.NewString("all"), "pending_delete")
+	query.Add(jsonutils.NewString("all"), "delete")
+	if len(ids) > 0 {
+		log.Debugf("to syncProjects for %s", jsonutils.Marshal(ids))
+		query.Add(jsonutils.NewStringArray(ids), "id")
+	}
+	total := -1
+	offset := 0
+	for total < 0 || offset < total {
+		query.Set("offset", jsonutils.NewInt(int64(offset)))
+		results, err := modules.Projects.List(s, query)
+		if err != nil {
+			return errors.Wrap(err, "Projects.List")
+		}
+		total = results.Total
+		for i := range results.Data {
+			// update project cache
+			item := db.SCachedTenant{}
+			deleted := jsonutils.QueryBoolean(results.Data[i], "deleted", false)
+			err := results.Data[i].Unmarshal(&item)
+			if err == nil && !deleted {
+				db.TenantCacheManager.Save(ctx, item, true)
+			} else if deleted {
+				tenantObj, _ := db.TenantCacheManager.FetchById(item.Id)
+				if tenantObj != nil {
+					tenantObj.Delete(ctx, nil)
+				}
+			}
+			offset++
+		}
+	}
+	return nil
+}

pkg/cloudcommon/db/cachesync/user.go

@@ -0,0 +1,84 @@
+// 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 cachesync
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/util/rbacscope"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
+	modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
+)
+
+type userCacheSyncWorker struct {
+	ids []string
+}
+
+func (w *userCacheSyncWorker) Run() {
+	log.Debugf("[userCacheSyncWorker] Run project cache sync worker ...")
+	err := syncUsers(context.Background(), w.ids)
+	if err != nil {
+		log.Errorf("fail to syncUsers %s", err)
+	}
+}
+
+func (w *userCacheSyncWorker) Dump() string {
+	return "userCacheSyncWorker"
+}
+
+func syncUsers(ctx context.Context, ids []string) error {
+	s := auth.GetAdminSession(ctx, consts.GetRegion())
+	query := jsonutils.NewDict()
+	query.Add(jsonutils.NewInt(1024), "limit")
+	query.Add(jsonutils.NewString(string(rbacscope.ScopeSystem)), "scope")
+	query.Add(jsonutils.JSONTrue, "details")
+	query.Add(jsonutils.NewString("all"), "pending_delete")
+	query.Add(jsonutils.NewString("all"), "delete")
+	if len(ids) > 0 {
+		query.Add(jsonutils.NewStringArray(ids), "id")
+	}
+	total := -1
+	offset := 0
+	for total < 0 || offset < total {
+		query.Set("offset", jsonutils.NewInt(int64(offset)))
+		results, err := modules.UsersV3.List(s, query)
+		if err != nil {
+			return errors.Wrap(err, "UsersV3.List")
+		}
+		total = results.Total
+		for i := range results.Data {
+			// update user cache
+			item := db.SCachedUser{}
+			deleted := jsonutils.QueryBoolean(results.Data[i], "deleted", false)
+			err := results.Data[i].Unmarshal(&item)
+			if err == nil && !deleted {
+				db.UserCacheManager.Save(ctx, item.Id, item.Name, item.DomainId, item.ProjectDomain, item.Lang)
+			} else if deleted {
+				usrObj, _ := db.UserCacheManager.FetchById(item.Id)
+				if usrObj != nil {
+					usrObj.Delete(ctx, nil)
+				}
+			}
+			offset++
+		}
+	}
+	return nil
+}

pkg/cloudcommon/db/rolecache.go

@@ -61,7 +61,7 @@ var RoleCacheManager *SRoleCacheManager
 
 func init() {
 	RoleCacheManager = &SRoleCacheManager{
-		NewKeystoneCacheObjectManager(SRole{}, "roles_cache_tbl", "role", "roles"), false}
+		NewKeystoneCacheObjectManager(SRole{}, "roles_cache_tbl", "role_cache", "role_caches"), false}
 	// log.Debugf("initialize role cache manager %s", RoleCacheManager.KeywordPlural())
 	RoleCacheManager.SetVirtualObject(RoleCacheManager)
 

pkg/cloudcommon/db/standalone_anon.go

@@ -260,19 +260,6 @@ func (model *SStandaloneAnonResourceBase) SetAllMetadata(ctx context.Context, di
 	return nil
 }
 
-func (model *SStandaloneAnonResourceBase) SetUserMetadataValues(ctx context.Context, dictstore map[string]string, userCred mcclient.TokenCredential) error {
-	dictStore, err := ensurePrefixString(dictstore, USER_TAG_PREFIX)
-	if err != nil {
-		return errors.Wrapf(err, "ensurePrefixString %s", USER_TAG_PREFIX)
-	}
-	err = Metadata.SetValuesWithLog(ctx, model, dictStore, userCred)
-	if err != nil {
-		return errors.Wrap(err, "SetValuesWithLog")
-	}
-	model.GetIStandaloneModel().OnMetadataUpdated(ctx, userCred)
-	return nil
-}
-
 func ensurePrefix(input map[string]interface{}, prefix string) (map[string]interface{}, error) {
 	dictStore := make(map[string]interface{}, len(input))
 	for k, v := range input {
@@ -307,6 +294,22 @@ func ensurePrefixString(input map[string]string, prefix string) (map[string]inte
 	return dictStore, nil
 }
 
+func (model *SStandaloneAnonResourceBase) SetUserMetadataValues(ctx context.Context, dictstore map[string]string, userCred mcclient.TokenCredential) error {
+	dictStore, err := ensurePrefixString(dictstore, USER_TAG_PREFIX)
+	if err != nil {
+		return errors.Wrapf(err, "ensurePrefixString %s", USER_TAG_PREFIX)
+	}
+	err = Metadata.SetValuesWithLog(ctx, model, dictStore, userCred)
+	if err != nil {
+		return errors.Wrap(err, "SetValuesWithLog")
+	}
+	{
+		model.GetModelManager().TableSpec().InformUpdate(ctx, model, jsonutils.Marshal(model).(*jsonutils.JSONDict))
+	}
+	model.GetIStandaloneModel().OnMetadataUpdated(ctx, userCred)
+	return nil
+}
+
 func (model *SStandaloneAnonResourceBase) SetUserMetadataAll(ctx context.Context, dictstore map[string]string, userCred mcclient.TokenCredential) error {
 	var err error
 	dictStore, err := ensurePrefixString(dictstore, USER_TAG_PREFIX)
@@ -317,6 +320,9 @@ func (model *SStandaloneAnonResourceBase) SetUserMetadataAll(ctx context.Context
 	if err != nil {
 		return errors.Wrap(err, "SetAll")
 	}
+	{
+		model.GetModelManager().TableSpec().InformUpdate(ctx, model, jsonutils.Marshal(model).(*jsonutils.JSONDict))
+	}
 	model.GetIStandaloneModel().OnMetadataUpdated(ctx, userCred)
 	return nil
 }

pkg/cloudcommon/db/tablespec.go

@@ -57,6 +57,8 @@ type ITableSpec interface {
 	GetTableSpec() *sqlchemy.STableSpec
 
 	GetDBName() sqlchemy.DBName
+
+	InformUpdate(ctx context.Context, dt interface{}, oldObj *jsonutils.JSONDict)
 }
 
 type sTableSpec struct {
@@ -303,6 +305,7 @@ func (ts *sTableSpec) informUpdate(ctx context.Context, dt interface{}, oldObj *
 			debug.PrintStack()
 			return
 		}
+		debug.PrintStack()
 		if err := informer.Update(ctx, obj, oldObj); err != nil {
 			if errors.Cause(err) == informer.ErrBackendNotInit {
 				log.V(4).Warningf("informer backend not init")
@@ -313,3 +316,7 @@ func (ts *sTableSpec) informUpdate(ctx context.Context, dt interface{}, oldObj *
 	}
 	nopanic.Run(nf)
 }
+
+func (ts *sTableSpec) InformUpdate(ctx context.Context, dt interface{}, oldObj *jsonutils.JSONDict) {
+	ts.informUpdate(ctx, dt, oldObj)
+}

pkg/cloudcommon/db/tenantcache.go

@@ -83,8 +83,8 @@ func init() {
 		SKeystoneCacheObjectManager: NewKeystoneCacheObjectManager(
 			STenant{},
 			"tenant_cache_tbl",
-			"tenant",
-			"tenants",
+			"tenant_cache",
+			"tenant_caches",
 		)}
 	// log.Debugf("Initialize tenant cache manager %s %s", TenantCacheManager.KeywordPlural(), TenantCacheManager)
 

pkg/cloudcommon/db/tenantcachesync.go

@@ -1,182 +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 db
-
-import (
-	"context"
-	"time"
-
-	"yunion.io/x/jsonutils"
-	"yunion.io/x/log"
-	"yunion.io/x/pkg/errors"
-	"yunion.io/x/pkg/util/rbacscope"
-
-	identityapi "yunion.io/x/onecloud/pkg/apis/identity"
-	"yunion.io/x/onecloud/pkg/appsrv"
-	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
-	"yunion.io/x/onecloud/pkg/mcclient/auth"
-	modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
-)
-
-var (
-	tenantCacheSyncWorkerMan = appsrv.NewWorkerManagerIgnoreOverflow("tenant_cache_sync_worker", 1, 1, true, true)
-)
-
-func StartTenantCacheSync(ctx context.Context, intvalSeconds int) {
-	go runTenantCacheSync(ctx, intvalSeconds)
-}
-
-func runTenantCacheSync(ctx context.Context, intvalSeconds int) {
-	for {
-		select {
-		case <-time.After(time.Duration(intvalSeconds) * time.Second):
-			tenantCacheSyncWorkerMan.Run(&tenantCacheSyncWorker{ctx}, nil, nil)
-		}
-	}
-}
-
-type tenantCacheSyncWorker struct {
-	ctx context.Context
-}
-
-func (w *tenantCacheSyncWorker) Run() {
-	log.Debugf("Run project and domain cache sync worker ...")
-	err := syncDomains(w.ctx)
-	if err != nil {
-		log.Errorf("fail to syncDomains %s", err)
-	}
-	err = syncProjects(w.ctx)
-	if err != nil {
-		log.Errorf("fail to syncProjects %s", err)
-	}
-	err = syncUsers(w.ctx)
-	if err != nil {
-		log.Errorf("fail to syncUsers %s", err)
-	}
-}
-
-func (w *tenantCacheSyncWorker) Dump() string {
-	return "tenantCacheSyncWorker"
-}
-
-func syncDomains(ctx context.Context) error {
-	s := auth.GetAdminSession(ctx, consts.GetRegion())
-	query := jsonutils.NewDict()
-	query.Add(jsonutils.NewInt(1024), "limit")
-	query.Add(jsonutils.NewString(string(rbacscope.ScopeSystem)), "scope")
-	query.Add(jsonutils.JSONTrue, "details")
-	query.Add(jsonutils.NewString("all"), "pending_delete")
-	query.Add(jsonutils.NewString("all"), "delete")
-	total := -1
-	offset := 0
-	for total < 0 || offset < total {
-		query.Set("offset", jsonutils.NewInt(int64(offset)))
-		results, err := modules.Domains.List(s, query)
-		if err != nil {
-			return errors.Wrap(err, "Domains.List")
-		}
-		total = results.Total
-		for i := range results.Data {
-			// update domain cache
-			item := SCachedTenant{}
-			deleted := jsonutils.QueryBoolean(results.Data[i], "deleted", false)
-			err := results.Data[i].Unmarshal(&item)
-			if err == nil && !deleted {
-				item.ProjectDomain = identityapi.KeystoneDomainRoot
-				item.DomainId = identityapi.KeystoneDomainRoot
-				TenantCacheManager.Save(ctx, item, true)
-			} else if deleted {
-				tenantObj, _ := TenantCacheManager.FetchById(item.Id)
-				if tenantObj != nil {
-					tenantObj.Delete(ctx, nil)
-				}
-			}
-			offset++
-		}
-	}
-	return nil
-}
-
-func syncProjects(ctx context.Context) error {
-	s := auth.GetAdminSession(ctx, consts.GetRegion())
-	query := jsonutils.NewDict()
-	query.Add(jsonutils.NewInt(1024), "limit")
-	query.Add(jsonutils.NewString(string(rbacscope.ScopeSystem)), "scope")
-	query.Add(jsonutils.JSONTrue, "details")
-	query.Add(jsonutils.NewString("all"), "pending_delete")
-	query.Add(jsonutils.NewString("all"), "delete")
-	total := -1
-	offset := 0
-	for total < 0 || offset < total {
-		query.Set("offset", jsonutils.NewInt(int64(offset)))
-		results, err := modules.Projects.List(s, query)
-		if err != nil {
-			return errors.Wrap(err, "Projects.List")
-		}
-		total = results.Total
-		for i := range results.Data {
-			// update project cache
-			item := SCachedTenant{}
-			deleted := jsonutils.QueryBoolean(results.Data[i], "deleted", false)
-			err := results.Data[i].Unmarshal(&item)
-			if err == nil && !deleted {
-				TenantCacheManager.Save(ctx, item, true)
-			} else if deleted {
-				tenantObj, _ := TenantCacheManager.FetchById(item.Id)
-				if tenantObj != nil {
-					tenantObj.Delete(ctx, nil)
-				}
-			}
-			offset++
-		}
-	}
-	return nil
-}
-
-func syncUsers(ctx context.Context) error {
-	s := auth.GetAdminSession(ctx, consts.GetRegion())
-	query := jsonutils.NewDict()
-	query.Add(jsonutils.NewInt(1024), "limit")
-	query.Add(jsonutils.NewString(string(rbacscope.ScopeSystem)), "scope")
-	query.Add(jsonutils.JSONTrue, "details")
-	query.Add(jsonutils.NewString("all"), "pending_delete")
-	query.Add(jsonutils.NewString("all"), "delete")
-	total := -1
-	offset := 0
-	for total < 0 || offset < total {
-		query.Set("offset", jsonutils.NewInt(int64(offset)))
-		results, err := modules.UsersV3.List(s, query)
-		if err != nil {
-			return errors.Wrap(err, "UsersV3.List")
-		}
-		total = results.Total
-		for i := range results.Data {
-			// update user cache
-			item := SCachedUser{}
-			deleted := jsonutils.QueryBoolean(results.Data[i], "deleted", false)
-			err := results.Data[i].Unmarshal(&item)
-			if err == nil && !deleted {
-				UserCacheManager.Save(ctx, item.Id, item.Name, item.DomainId, item.ProjectDomain, item.Lang)
-			} else if deleted {
-				usrObj, _ := UserCacheManager.FetchById(item.Id)
-				if usrObj != nil {
-					usrObj.Delete(ctx, nil)
-				}
-			}
-			offset++
-		}
-	}
-	return nil
-}

pkg/cloudcommon/db/usercache.go

@@ -55,7 +55,7 @@ var UserCacheManager *SUserCacheManager
 
 func init() {
 	UserCacheManager = &SUserCacheManager{
-		NewKeystoneCacheObjectManager(SUser{}, "users_cache_tbl", "user", "users")}
+		NewKeystoneCacheObjectManager(SUser{}, "users_cache_tbl", "user_cache", "user_caches")}
 	// log.Debugf("initialize user cache manager %s", UserCacheManager.KeywordPlural())
 	UserCacheManager.SetVirtualObject(UserCacheManager)
 

pkg/cloudcommon/options/manager.go

@@ -123,7 +123,7 @@ func optionsEquals(newOpts interface{}, oldOpts interface{}) bool {
 	return true
 }
 
-func (manager *SOptionManager) DoSync(first bool) (time.Duration, error) {
+func (manager *SOptionManager) DoSync(first bool, timeout bool) (time.Duration, error) {
 	newOpts := manager.newOptions()
 	copyOptions(newOpts, manager.options)
 	merged := manager.session.Merge(newOpts, manager.serviceType, manager.serviceVersion)

pkg/cloudcommon/syncman/sync.go

@@ -26,7 +26,7 @@ import (
 )
 
 type ISyncClient interface {
-	DoSync(first bool) (time.Duration, error)
+	DoSync(first bool, timeout bool) (time.Duration, error)
 	NeedSync(dat *jsonutils.JSONDict) bool
 	Name() string
 }
@@ -46,46 +46,48 @@ func (manager *SSyncManager) InitSync(client ISyncClient) {
 	manager.syncWorkerManager = appsrv.NewWorkerManagerIgnoreOverflow(fmt.Sprintf("(%s)sync_worker", client.Name()), 1, 1, true, true)
 }
 
-func (manager *SSyncManager) syncByInterval() error {
+func (manager *SSyncManager) syncInternal(isFirst bool, isTimeout bool) error {
 	if manager.syncTimer != nil {
 		manager.syncTimer.Stop()
 		manager.syncTimer = nil
 	}
-	isFirst := false
-	if manager.lastSync.IsZero() {
-		isFirst = true
-	}
-	next, err := manager.DoSync(isFirst)
+	next, err := manager.DoSync(isFirst, isTimeout)
 	if err == nil {
 		manager.lastSync = time.Now()
 	}
-	manager.syncTimer = time.AfterFunc(next, manager.SyncOnce)
+	manager.syncTimer = time.AfterFunc(next, func() {
+		manager.SyncOnce(false, true)
+	})
 	return err
 }
 
 type SyncTask struct {
-	manager *SSyncManager
+	manager   *SSyncManager
+	isFirst   bool
+	isTimeout bool
 }
 
 func (t *SyncTask) Run() {
 	atomic.StoreInt32(&t.manager.syncOnce, 0)
-	t.manager.syncByInterval()
+	t.manager.syncInternal(t.isFirst, t.isTimeout)
 }
 
 func (t *SyncTask) Dump() string {
-	return ""
+	return "SyncTask"
 }
 
-func (manager *SSyncManager) SyncOnce() {
-	log.Debugf("[%s] SyncOnce", manager.Name())
+func (manager *SSyncManager) SyncOnce(isFirst bool, isTimeout bool) {
+	log.Debugf("[%s] SyncOnce isFirst %v isTimeout %v", manager.Name(), isFirst, isTimeout)
 	if atomic.CompareAndSwapInt32(&manager.syncOnce, 0, 1) {
 		task := SyncTask{
-			manager: manager,
+			manager:   manager,
+			isFirst:   isFirst,
+			isTimeout: isTimeout,
 		}
 		manager.syncWorkerManager.Run(&task, nil, nil)
 	}
 }
 
 func (manager *SSyncManager) FirstSync() error {
-	return manager.syncByInterval()
+	return manager.syncInternal(true, false)
 }

pkg/cloudcommon/syncman/watcher/watcher.go

@@ -39,21 +39,21 @@ type SInformerSyncManager struct {
 func (manager *SInformerSyncManager) OnAdd(obj *jsonutils.JSONDict) {
 	log.Infof("[CREATED]: \n%s", obj.String())
 	if manager.NeedSync(obj) {
-		manager.SyncOnce()
+		manager.SyncOnce(false, false)
 	}
 }
 
 func (manager *SInformerSyncManager) OnUpdate(oldObj, newObj *jsonutils.JSONDict) {
 	log.Infof("[UPDATED]: \n[NEW]: %s\n[OLD]: %s", newObj.String(), oldObj.String())
 	if manager.NeedSync(oldObj) || manager.NeedSync(newObj) {
-		manager.SyncOnce()
+		manager.SyncOnce(false, false)
 	}
 }
 
 func (manager *SInformerSyncManager) OnDelete(obj *jsonutils.JSONDict) {
 	log.Infof("[DELETED]: \n%s", obj.String())
 	if manager.NeedSync(obj) {
-		manager.SyncOnce()
+		manager.SyncOnce(false, false)
 	}
 }
 

pkg/compute/service/service.go

@@ -34,6 +34,7 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
 	"yunion.io/x/onecloud/pkg/cloudcommon/cronman"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db/cachesync"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
 	"yunion.io/x/onecloud/pkg/cloudcommon/elect"
 	"yunion.io/x/onecloud/pkg/cloudcommon/etcd"
@@ -145,7 +146,7 @@ func StartServiceWithJobs(jobs func(cron *cronman.SCronJobManager)) {
 	}
 
 	cronFunc := func() {
-		db.StartTenantCacheSync(app.GetContext(), opts.TenantCacheExpireSeconds)
+		cachesync.StartTenantCacheSync(opts.TenantCacheExpireSeconds)
 
 		cron := cronman.InitCronJobManager(true, options.Options.CronJobWorkerCount)
 		cron.AddJobAtIntervals("CleanPendingDeleteServers", time.Duration(opts.PendingDeleteCheckSeconds)*time.Second, models.GuestManager.CleanPendingDeleteServers)

pkg/image/service/service.go

@@ -34,6 +34,7 @@ import (
 	app_common "yunion.io/x/onecloud/pkg/cloudcommon/app"
 	"yunion.io/x/onecloud/pkg/cloudcommon/cronman"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db/cachesync"
 	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
 	"yunion.io/x/onecloud/pkg/hostman/hostdeployer/deployclient"
 	"yunion.io/x/onecloud/pkg/image/drivers/s3"
@@ -139,6 +140,8 @@ func StartService() {
 	}
 
 	if !opts.IsSlaveNode {
+		cachesync.StartTenantCacheSync(opts.TenantCacheExpireSeconds)
+
 		cron := cronman.InitCronJobManager(true, options.Options.CronJobWorkerCount)
 		cron.AddJobAtIntervals("CleanPendingDeleteImages", time.Duration(options.Options.PendingDeleteCheckSeconds)*time.Second, models.ImageManager.CleanPendingDeleteImages)
 		cron.AddJobAtIntervals("CalculateQuotaUsages", time.Duration(opts.CalculateQuotaUsageIntervalSeconds)*time.Second, models.QuotaManager.CalculateQuotaUsages)

pkg/keystone/models/services.go

@@ -199,7 +199,7 @@ func (service *SService) PerformConfig(ctx context.Context, userCred mcclient.To
 			return nil, httperrors.NewInternalServerError("update config version fail %s", err)
 		}
 		if service.Type == api.SERVICE_TYPE || service.Type == consts.COMMON_SERVICE {
-			options.OptionManager.SyncOnce()
+			options.OptionManager.SyncOnce(false, false)
 		}
 	}
 	return service.GetDetailsConfig(ctx, userCred, query)

pkg/keystone/models/users.go

@@ -24,6 +24,7 @@ import (
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/tristate"
 	"yunion.io/x/pkg/util/rbacscope"
+	"yunion.io/x/pkg/utils"
 	"yunion.io/x/sqlchemy"
 
 	"yunion.io/x/onecloud/pkg/apis"
@@ -1124,12 +1125,17 @@ func (manager *SUserManager) traceLoginEvent(ctx context.Context, token mcclient
 		log.Errorf("fetchUserById fail %s", err)
 		return
 	}
-	db.Update(usr, func() error {
-		usr.LastActiveAt = time.Now().UTC()
-		usr.LastLoginIp = authCtx.Ip
-		usr.LastLoginSource = authCtx.Source
-		return nil
-	})
+
+	// only save web console login record
+	if usr.LastActiveAt.IsZero() || utils.IsInArray(authCtx.Source, []string{mcclient.AuthSourceWeb}) {
+		db.Update(usr, func() error {
+			usr.LastActiveAt = time.Now().UTC()
+			usr.LastLoginIp = authCtx.Ip
+			usr.LastLoginSource = authCtx.Source
+			return nil
+		})
+	}
+
 	db.OpsLog.LogEvent(usr, "auth", &s, token)
 	// to reduce auth event, log web console login only
 	if authCtx.Source == mcclient.AuthSourceWeb && token.GetProjectId() != "" {

pkg/mcclient/auth/auth.go

@@ -277,12 +277,12 @@ func (a *authManager) authAdmin() error {
 	}
 }
 
-func (a *authManager) DoSync(first bool) (time.Duration, error) {
+func (a *authManager) DoSync(first bool, timeout bool) (time.Duration, error) {
 	err := a.authAdmin()
 	if err != nil {
 		return time.Minute, errors.Wrap(err, "authAdmin")
 	} else {
-		return a.adminCredential.GetExpires().Sub(time.Now()) / 2, nil
+		return time.Until(a.adminCredential.GetExpires()) / 2, nil
 	}
 }
 
@@ -295,7 +295,7 @@ func (a *authManager) Name() string {
 }
 
 func (a *authManager) reAuth() {
-	a.SyncOnce()
+	a.SyncOnce(false, false)
 }
 
 func (a *authManager) GetServiceURL(service, region, zone, endpointType string) (string, error) {