Merge pull request #13217 from swordqiu/feature/qj-full-sync-project-domain-meta

fix: full sync of project and domain metadata

pkg/cloudcommon/db/metadataresource.go

@@ -31,7 +31,6 @@ import (
 type SMetadataResourceBaseModelManager struct{}
 
 func ObjectIdQueryWithTagFilters(q *sqlchemy.SQuery, idField string, modelName string, filters tagutils.STagFilters) *sqlchemy.SQuery {
-	log.Debugf("Filters: %s", jsonutils.Marshal(filters))
 	if len(filters.Filters) > 0 {
 		sq := objIdQueryWithTags(modelName, filters.Filters)
 		if sq != nil {

pkg/cloudcommon/db/tenantcache.go

@@ -336,6 +336,9 @@ func (manager *STenantCacheManager) Save(ctx context.Context, item SCachedTenant
 				obj.LastCheck = now
 				return nil
 			})
+			if saveMeta {
+				Metadata.rawSetValues(ctx, item.objType(), item.Id, item.Metadata, true, "")
+			}
 			return obj, nil
 		}
 		_, err = Update(obj, func() error {

pkg/cloudcommon/db/tenantcachesync.go

@@ -0,0 +1,124 @@
+// 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"
+
+	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"
+	"yunion.io/x/onecloud/pkg/util/rbacutils"
+)
+
+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)
+	}
+}
+
+func (w *tenantCacheSyncWorker) Dump() string {
+	return "tenantCacheSyncWorker"
+}
+
+func syncDomains(ctx context.Context) error {
+	s := auth.GetAdminSession(ctx, consts.GetRegion(), "v1")
+	query := jsonutils.NewDict()
+	query.Add(jsonutils.NewInt(1024), "limit")
+	query.Add(jsonutils.NewString(string(rbacutils.ScopeSystem)), "scope")
+	query.Add(jsonutils.JSONTrue, "details")
+	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 {
+			log.Errorf("syncDomain error %s", err)
+			return errors.Wrap(err, "Domains.List")
+		}
+		total = results.Total
+		for i := range results.Data {
+			// update domain cache
+			item := SCachedTenant{}
+			results.Data[i].Unmarshal(&item)
+			item.ProjectDomain = identityapi.KeystoneDomainRoot
+			item.DomainId = identityapi.KeystoneDomainRoot
+			TenantCacheManager.Save(ctx, item, true)
+			offset++
+		}
+	}
+	return nil
+}
+
+func syncProjects(ctx context.Context) error {
+	s := auth.GetAdminSession(ctx, consts.GetRegion(), "v1")
+	query := jsonutils.NewDict()
+	query.Add(jsonutils.NewInt(1024), "limit")
+	query.Add(jsonutils.NewString(string(rbacutils.ScopeSystem)), "scope")
+	query.Add(jsonutils.JSONTrue, "details")
+	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 {
+			log.Errorf("syncProjects error %s", err)
+			return errors.Wrap(err, "Projects.List")
+		}
+		total = results.Total
+		for i := range results.Data {
+			// update project cache
+			item := SCachedTenant{}
+			results.Data[i].Unmarshal(&item)
+			TenantCacheManager.Save(ctx, item, true)
+			offset++
+		}
+	}
+	return nil
+}

pkg/compute/service/service.go

@@ -117,6 +117,8 @@ func StartService() {
 	}
 
 	if !opts.IsSlaveNode {
+		db.StartTenantCacheSync(app.GetContext(), opts.TenantCacheExpireSeconds)
+
 		cron := cronman.InitCronJobManager(true, options.Options.CronJobWorkerCount)
 		cron.AddJobAtIntervals("CleanPendingDeleteServers", time.Duration(opts.PendingDeleteCheckSeconds)*time.Second, models.GuestManager.CleanPendingDeleteServers)
 		cron.AddJobAtIntervals("CleanPendingDeleteDisks", time.Duration(opts.PendingDeleteCheckSeconds)*time.Second, models.DiskManager.CleanPendingDeleteDisks)