diff --git a/pkg/appsrv/ring.go b/pkg/appsrv/ring.go index 60d7a3e1b6..c46735dc60 100644 --- a/pkg/appsrv/ring.go +++ b/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 + } + } +} diff --git a/pkg/appsrv/workers.go b/pkg/appsrv/workers.go index 178c95a387..f0802389e4 100644 --- a/pkg/appsrv/workers.go +++ b/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() diff --git a/pkg/cloudcommon/app/endpoint.go b/pkg/cloudcommon/app/endpoint.go index 8cd65f2519..dbbec59556 100644 --- a/pkg/cloudcommon/app/endpoint.go +++ b/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 diff --git a/pkg/cloudcommon/db/cachesync/cachesync.go b/pkg/cloudcommon/db/cachesync/cachesync.go new file mode 100644 index 0000000000..e965ce9d25 --- /dev/null +++ b/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) +} diff --git a/pkg/cloudcommon/db/cachesync/changeman.go b/pkg/cloudcommon/db/cachesync/changeman.go new file mode 100644 index 0000000000..c12a9b50cd --- /dev/null +++ b/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()) +} diff --git a/pkg/cloudcommon/db/cachesync/doc.go b/pkg/cloudcommon/db/cachesync/doc.go new file mode 100644 index 0000000000..d75b3f6f49 --- /dev/null +++ b/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" diff --git a/pkg/cloudcommon/db/cachesync/domain.go b/pkg/cloudcommon/db/cachesync/domain.go new file mode 100644 index 0000000000..bb99e5f743 --- /dev/null +++ b/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 +} diff --git a/pkg/cloudcommon/db/cachesync/project.go b/pkg/cloudcommon/db/cachesync/project.go new file mode 100644 index 0000000000..905a223d5f --- /dev/null +++ b/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 +} diff --git a/pkg/cloudcommon/db/cachesync/user.go b/pkg/cloudcommon/db/cachesync/user.go new file mode 100644 index 0000000000..ea74896878 --- /dev/null +++ b/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 +} diff --git a/pkg/cloudcommon/db/rolecache.go b/pkg/cloudcommon/db/rolecache.go index ca8f01534b..213922f245 100644 --- a/pkg/cloudcommon/db/rolecache.go +++ b/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) diff --git a/pkg/cloudcommon/db/standalone_anon.go b/pkg/cloudcommon/db/standalone_anon.go index b5506e2d68..4d9f9d1d57 100644 --- a/pkg/cloudcommon/db/standalone_anon.go +++ b/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 } diff --git a/pkg/cloudcommon/db/tablespec.go b/pkg/cloudcommon/db/tablespec.go index d6587be9fb..0c61a37d74 100644 --- a/pkg/cloudcommon/db/tablespec.go +++ b/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) +} diff --git a/pkg/cloudcommon/db/tenantcache.go b/pkg/cloudcommon/db/tenantcache.go index 93dbecf35a..279665f4a1 100644 --- a/pkg/cloudcommon/db/tenantcache.go +++ b/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) diff --git a/pkg/cloudcommon/db/tenantcachesync.go b/pkg/cloudcommon/db/tenantcachesync.go deleted file mode 100644 index 7615d73a12..0000000000 --- a/pkg/cloudcommon/db/tenantcachesync.go +++ /dev/null @@ -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 -} diff --git a/pkg/cloudcommon/db/usercache.go b/pkg/cloudcommon/db/usercache.go index 98070da1ea..d599c3e4dd 100644 --- a/pkg/cloudcommon/db/usercache.go +++ b/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) diff --git a/pkg/cloudcommon/options/manager.go b/pkg/cloudcommon/options/manager.go index a429451c42..964caab9e9 100644 --- a/pkg/cloudcommon/options/manager.go +++ b/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) diff --git a/pkg/cloudcommon/syncman/sync.go b/pkg/cloudcommon/syncman/sync.go index 847faa2eed..a98906d425 100644 --- a/pkg/cloudcommon/syncman/sync.go +++ b/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) } diff --git a/pkg/cloudcommon/syncman/watcher/watcher.go b/pkg/cloudcommon/syncman/watcher/watcher.go index 9ecdf8f488..fd380f199b 100644 --- a/pkg/cloudcommon/syncman/watcher/watcher.go +++ b/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) } } diff --git a/pkg/compute/service/service.go b/pkg/compute/service/service.go index 9c613aae9f..2c93f2aa3c 100644 --- a/pkg/compute/service/service.go +++ b/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) diff --git a/pkg/image/service/service.go b/pkg/image/service/service.go index 7912c50259..8d313c7c0d 100644 --- a/pkg/image/service/service.go +++ b/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) diff --git a/pkg/keystone/models/services.go b/pkg/keystone/models/services.go index c8ec3d9836..3ae73d4fb4 100644 --- a/pkg/keystone/models/services.go +++ b/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) diff --git a/pkg/keystone/models/users.go b/pkg/keystone/models/users.go index cc5207a1d6..d20e57dd41 100644 --- a/pkg/keystone/models/users.go +++ b/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() != "" { diff --git a/pkg/mcclient/auth/auth.go b/pkg/mcclient/auth/auth.go index 604284862a..8c52e54a2b 100644 --- a/pkg/mcclient/auth/auth.go +++ b/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) {