Merge pull request #3466 from tb365/feature/tb-redis-sku-sync

redis sku sync

pkg/compute/models/cloudsync.go

@@ -93,7 +93,7 @@ func syncRegionZones(ctx context.Context, userCred mcclient.TokenCredential, syn
 	return localZones, remoteZones, nil
 }
 
-func syncRegionSkus(ctx context.Context, localRegion *SCloudregion) {
+func syncRegionSkus(ctx context.Context, userCred mcclient.TokenCredential, localRegion *SCloudregion) {
 	if localRegion == nil {
 		log.Debugf("local region is nil, skipp...")
 		return
@@ -127,7 +127,7 @@ func syncRegionSkus(ctx context.Context, localRegion *SCloudregion) {
 		log.Errorf("SchedManager SyncSku %s", err)
 	}
 
-	syncElasticCacheSkusByRegion(localRegion)
+	syncElasticCacheSkusByRegion(ctx, userCred, localRegion)
 }
 
 func syncProjects(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, driver cloudprovider.ICloudProvider, provider *SCloudprovider) {
@@ -990,7 +990,7 @@ func syncPublicCloudProviderInfo(
 	localZones, remoteZones, _ := syncRegionZones(ctx, userCred, syncResults, provider, localRegion, remoteRegion)
 
 	if !driver.GetFactory().NeedSyncSkuFromCloud() {
-		syncRegionSkus(ctx, localRegion)
+		syncRegionSkus(ctx, userCred, localRegion)
 	}
 
 	// no need to lock public cloud region as cloud region for public cloud is readonly

pkg/compute/models/elasticcache_skus.go

@@ -16,15 +16,16 @@ package models
 
 import (
 	"context"
-	"database/sql"
-	"fmt"
 
 	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/util/compare"
 	"yunion.io/x/sqlchemy"
 
 	api "yunion.io/x/onecloud/pkg/apis/compute"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
 	"yunion.io/x/onecloud/pkg/cloudcommon/validators"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/util/stringutils2"
@@ -87,6 +88,10 @@ type SElasticcacheSku struct {
 	Provider string `width:"32" charset:"ascii" nullable:"false" list:"user" create:"admin_required" update:"admin"` // 公有云厂商	Aliyun/Azure/AWS/Qcloud/...
 }
 
+func (self SElasticcacheSku) GetGlobalId() string {
+	return self.ExternalId
+}
+
 func (self *SElasticcacheSku) GetCustomizeColumns(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) *jsonutils.JSONDict {
 	return self.SStatusStandaloneResourceBase.GetCustomizeColumns(ctx, userCred, query)
 }
@@ -170,106 +175,96 @@ func (manager *SElasticcacheSkuManager) ListItemFilter(ctx context.Context, q *s
 	return q, err
 }
 
-// 获取所有Available状态的sku id
-func (manager *SElasticcacheSkuManager) FetchAllAvailableSkuId() ([]string, error) {
-	q := manager.Query()
-	q = q.Filter(sqlchemy.OR(
-		sqlchemy.Equals(q.Field("prepaid_status"), api.SkuStatusAvailable),
-		sqlchemy.Equals(q.Field("postpaid_status"), api.SkuStatusAvailable)))
-
-	skus := make([]SElasticcacheSku, 0)
-	err := db.FetchModelObjects(manager, q, &skus)
-	if err != nil {
-		return nil, err
-	}
-
-	ids := make([]string, len(skus))
-	for i := range skus {
-		ids[i] = skus[i].GetId()
-	}
-
-	return ids, nil
-}
-
 // 获取region下所有Available状态的sku id
-func (manager *SElasticcacheSkuManager) FetchAllAvailableSkuIdByRegion(regionID string) ([]string, error) {
+func (manager *SElasticcacheSkuManager) FetchSkusByRegion(regionID string) ([]SElasticcacheSku, error) {
 	q := manager.Query()
-	q = q.Filter(sqlchemy.OR(
-		sqlchemy.Equals(q.Field("prepaid_status"), api.SkuStatusAvailable),
-		sqlchemy.Equals(q.Field("postpaid_status"), api.SkuStatusAvailable)))
 	q = q.Equals("cloudregion_id", regionID)
 
 	skus := make([]SElasticcacheSku, 0)
 	err := db.FetchModelObjects(manager, q, &skus)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "ElasticcacheSkuManager.FetchSkusByRegion")
 	}
 
-	ids := make([]string, len(skus))
-	for i := range skus {
-		ids[i] = skus[i].GetId()
-	}
-
-	return ids, nil
+	return skus, nil
 }
 
-func (manager *SElasticcacheSkuManager) InitializeData() error {
-	count, err := manager.Query().Limit(1).CountWithError()
-	if err != nil && err != sql.ErrNoRows {
-		return err
-	}
+func (manager *SElasticcacheSkuManager) syncDBInstanceSkus(ctx context.Context, userCred mcclient.TokenCredential, region *SCloudregion, extSkuMeta *SSkuResourcesMeta) compare.SyncResult {
+	lockman.LockClass(ctx, manager, db.GetLockClassKey(manager, userCred))
+	defer lockman.ReleaseClass(ctx, manager, db.GetLockClassKey(manager, userCred))
 
-	if count > 0 {
-		return nil
-	}
+	syncResult := compare.SyncResult{}
 
-	count, err = CloudaccountManager.Query().IsTrue("is_public_cloud").CountWithError()
-	if count > 0 {
-		SyncElasticCacheSkus(nil, nil, true)
-	}
-
-	return nil
-}
-
-// sku标记为soldout状态。
-func (manager *SElasticcacheSkuManager) MarkAsSoldout(id string) error {
-	if len(id) == 0 {
-		log.Debugf("MarkAsSoldout sku id should not be emtpy")
-		return nil
-	}
-
-	isku, err := manager.FetchById(id)
+	extSkuMeta.SetRegionFilter(region)
+	extSkus, err := extSkuMeta.GetElasticCacheSkus()
 	if err != nil {
-		return err
+		syncResult.Error(err)
+		return syncResult
 	}
 
-	sku, ok := isku.(*SServerSku)
-	if !ok {
-		return fmt.Errorf("%s is not a sku object", id)
+	dbSkus, err := manager.FetchSkusByRegion(region.GetId())
+	if err != nil {
+		syncResult.Error(err)
+		return syncResult
 	}
 
-	_, err = manager.TableSpec().Update(sku, func() error {
-		sku.PrepaidStatus = api.SkuStatusSoldout
-		sku.PostpaidStatus = api.SkuStatusSoldout
+	removed := make([]SElasticcacheSku, 0)
+	commondb := make([]SElasticcacheSku, 0)
+	commonext := make([]SElasticcacheSku, 0)
+	added := make([]SElasticcacheSku, 0)
+
+	err = compare.CompareSets(dbSkus, extSkus, &removed, &commondb, &commonext, &added)
+	if err != nil {
+		syncResult.Error(err)
+		return syncResult
+	}
+
+	for i := 0; i < len(removed); i += 1 {
+		err = removed[i].MarkAsSoldout(ctx)
+		if err != nil {
+			syncResult.DeleteError(err)
+		} else {
+			syncResult.Delete()
+		}
+	}
+	for i := 0; i < len(commondb); i += 1 {
+		err = commondb[i].syncWithCloudSku(ctx, userCred, commonext[i])
+		if err != nil {
+			syncResult.UpdateError(err)
+		} else {
+			syncResult.Update()
+		}
+	}
+	for i := 0; i < len(added); i += 1 {
+		err = manager.newFromCloudSku(ctx, userCred, added[i])
+		if err != nil {
+			syncResult.AddError(err)
+		} else {
+			syncResult.Add()
+		}
+	}
+	return syncResult
+}
+
+func (self *SElasticcacheSku) MarkAsSoldout(ctx context.Context) error {
+	_, err := db.UpdateWithLock(ctx, self, func() error {
+		self.PrepaidStatus = api.SkuStatusSoldout
+		self.PostpaidStatus = api.SkuStatusSoldout
 		return nil
 	})
 
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return errors.Wrap(err, "ElasticcacheSku.MarkAsSoldout")
 }
 
-// sku标记为soldout状态。
-func (manager *SElasticcacheSkuManager) MarkAllAsSoldout(ids []string) error {
-	var err error
-	for _, id := range ids {
-		err = manager.MarkAsSoldout(id)
-		if err != nil {
-			return err
-		}
-	}
+func (self *SElasticcacheSku) syncWithCloudSku(ctx context.Context, userCred mcclient.TokenCredential, extSku SElasticcacheSku) error {
+	_, err := db.Update(self, func() error {
+		self.PrepaidStatus = extSku.PrepaidStatus
+		self.PostpaidStatus = extSku.PostpaidStatus
+		return nil
+	})
+	return err
+}
 
-	return nil
+func (manager *SElasticcacheSkuManager) newFromCloudSku(ctx context.Context, userCred mcclient.TokenCredential, extSku SElasticcacheSku) error {
+	return manager.TableSpec().Insert(&extSku)
 }

pkg/compute/models/skus_tools.go

@@ -0,0 +1,291 @@
+package models
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/compute/options"
+	"yunion.io/x/onecloud/pkg/mcclient"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
+	"yunion.io/x/onecloud/pkg/mcclient/modules"
+)
+
+/*
+资源套餐下载连接信息
+server: 虚拟机
+elasticcache: 弹性缓存(redis&memcached)
+*/
+type SSkuResourcesMeta struct {
+	region       *SCloudregion
+	caches       map[string][]jsonutils.JSONObject
+	zoneCaches   map[string]*SZone
+	regionCaches map[string]*SCloudregion
+
+	Server       string
+	ElasticCache string
+}
+
+// todo: 待测试
+func (self *SSkuResourcesMeta) GetServerSkus() ([]SServerSku, error) {
+	result := []SServerSku{}
+	err := self.getSkus(self.Server, &result)
+	if err != nil {
+		return nil, errors.Wrap(err, "SkuResourcesMeta.GetServerSkus")
+	}
+
+	// todo: process data here
+	return result, nil
+}
+
+func (self *SSkuResourcesMeta) GetElasticCacheSkus() ([]SElasticcacheSku, error) {
+	result := []SElasticcacheSku{}
+	err := self.getSkus(self.ElasticCache, &result)
+	if err != nil {
+		return nil, errors.Wrap(err, "SkuResourcesMeta.GetElasticCacheSkus")
+	}
+
+	// 处理数据
+	for i := range result {
+		provider := result[i].Provider
+		region := result[i].CloudregionId
+
+		result[i].Id = ""
+		r, err := self.fetchRegion(provider, region)
+		if err != nil {
+			return nil, errors.Wrap(err, "SkuResourcesMeta.GetElasticCacheSkus.fetchRegion")
+		}
+		result[i].CloudregionId = r.GetId()
+
+		if len(result[i].ZoneId) > 0 {
+			zone, err := self.fetchZone(provider, region, result[i].ZoneId)
+			if err != nil {
+				return nil, errors.Wrap(err, "SkuResourcesMeta.GetElasticCacheSkus.MasterZone")
+			}
+
+			result[i].ZoneId = zone.GetId()
+		}
+
+		if len(result[i].SlaveZoneId) > 0 {
+			zone, err := self.fetchZone(provider, region, result[i].SlaveZoneId)
+			if err != nil {
+				return nil, errors.Wrap(err, "SkuResourcesMeta.GetElasticCacheSkus.SlaveZone")
+			}
+
+			result[i].SlaveZoneId = zone.GetId()
+		}
+	}
+
+	return result, nil
+}
+
+func (self *SSkuResourcesMeta) fetchZone(provider, region, zone string) (*SZone, error) {
+	if self.zoneCaches == nil {
+		self.zoneCaches = map[string]*SZone{}
+	}
+
+	zoneExternalId := strings.Join([]string{provider, region, zone}, "/")
+	if z, ok := self.zoneCaches[zoneExternalId]; ok {
+		return z, nil
+	}
+
+	_zone, err := db.FetchByExternalId(ZoneManager, zoneExternalId)
+	if err != nil {
+		return nil, errors.Wrap(err, "SkuResourcesMeta.fetchZone.FetchByExternalId")
+	}
+
+	z := _zone.(*SZone)
+	self.zoneCaches[zoneExternalId] = z
+	return z, nil
+}
+
+func (self *SSkuResourcesMeta) fetchRegion(provider, region string) (*SCloudregion, error) {
+	if self.regionCaches == nil {
+		self.regionCaches = map[string]*SCloudregion{}
+	}
+
+	regionExternalId := strings.Join([]string{provider, region}, "/")
+	if r, ok := self.regionCaches[regionExternalId]; ok {
+		return r, nil
+	}
+
+	_region, err := db.FetchByExternalId(CloudregionManager, regionExternalId)
+	if err != nil {
+		return nil, errors.Wrap(err, "SkuResourcesMeta.fetchRegion.FetchByExternalId")
+	}
+
+	r := _region.(*SCloudregion)
+	self.regionCaches[regionExternalId] = r
+	return r, nil
+}
+
+func (self *SSkuResourcesMeta) getSkus(url string, result interface{}) error {
+	objs, err := self.get(self.ElasticCache)
+	if err != nil {
+		return errors.Wrap(err, "SkuResourcesMeta.getSkus")
+	}
+
+	objArray := jsonutils.Marshal(objs)
+	err = objArray.Unmarshal(result)
+	if err != nil {
+		return errors.Wrap(err, "SkuResourcesMeta.Unmarshal")
+	}
+
+	return nil
+}
+
+func (self *SSkuResourcesMeta) SetRegionFilter(region *SCloudregion) {
+	self.region = region
+}
+
+func (self *SSkuResourcesMeta) filterByRegion(items []jsonutils.JSONObject) []jsonutils.JSONObject {
+	if self.region == nil {
+		return items
+	}
+
+	ret := []jsonutils.JSONObject{}
+	for i := range items {
+		item := items[i]
+		provider, _ := item.GetString("provider")
+		regionId, _ := item.GetString("cloudregion_id")
+		if self.region.GetExternalId() != fmt.Sprintf("%s/%s", provider, regionId) {
+			continue
+		}
+
+		ret = append(ret, item)
+	}
+
+	return ret
+}
+
+func (self *SSkuResourcesMeta) get(url string) ([]jsonutils.JSONObject, error) {
+	if self.caches == nil {
+		self.caches = map[string][]jsonutils.JSONObject{}
+	}
+
+	if items, ok := self.caches[url]; !ok || len(items) == 0 {
+		items, err := self._get(url)
+		if err != nil {
+			return nil, errors.Wrap(err, "SkuResourcesMeta.get")
+		}
+
+		self.caches[url] = items
+	}
+
+	items := self.caches[url]
+	return self.filterByRegion(items), nil
+}
+
+func (self *SSkuResourcesMeta) _get(url string) ([]jsonutils.JSONObject, error) {
+	if !strings.HasPrefix(url, "http") {
+		return nil, fmt.Errorf("SkuResourcesMeta.get invalid url %s.expected has prefix 'http'", url)
+	}
+
+	resp, err := http.Get(url)
+	if err != nil {
+		return nil, fmt.Errorf("SkuResourcesMeta.get.Get %s", err)
+	}
+
+	defer resp.Body.Close()
+	content, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("SkuResourcesMeta.get.Read %s", err)
+	}
+
+	contentJson, err := jsonutils.Parse(content)
+	if err != nil {
+		return nil, fmt.Errorf("SkuResourcesMeta.get.Parse %s", err)
+	}
+
+	ret := []jsonutils.JSONObject{}
+	err = contentJson.Unmarshal(&ret)
+	if err != nil {
+		return nil, fmt.Errorf("SkuResourcesMeta.get.Unmarshal %s", err)
+	}
+
+	return ret, nil
+}
+
+// 全量同步elasticcache sku列表.
+func SyncElasticCacheSkus(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) {
+	if isStart {
+		cnt, err := CloudaccountManager.Query().IsTrue("is_public_cloud").CountWithError()
+		if err != nil && err != sql.ErrNoRows {
+			log.Debugf("SyncElasticCacheSkus %s.sync skipped...", err)
+			return
+		} else if cnt == 0 {
+			log.Debugf("SyncElasticCacheSkus no public cloud.sync skipped...")
+			return
+		}
+
+		cnt, err = ElasticcacheSkuManager.Query().Limit(1).CountWithError()
+		if err != nil && err != sql.ErrNoRows {
+			log.Errorf("SyncElasticCacheSkus.QueryElasticcacheSku %s", err)
+			return
+		} else if cnt > 0 {
+			log.Debugf("SyncElasticCacheSkus synced skus, skip...")
+			return
+		}
+	}
+
+	cloudregions := []SCloudregion{}
+	q := CloudregionManager.Query()
+	q = q.In("provider", CloudproviderManager.GetPublicProviderProvidersQuery())
+	err := db.FetchModelObjects(CloudregionManager, q, &cloudregions)
+	if err != nil {
+		log.Errorf("SyncElasticCacheSkus.FetchCloudregions failed: %v", err)
+		return
+	}
+
+	meta, err := fetchSkuResourcesMeta()
+	if err != nil {
+		log.Errorf("SyncElasticCacheSkus.fetchSkuResourcesMeta %s", err)
+		return
+	}
+
+	for i := range cloudregions {
+		region := &cloudregions[i]
+		meta.SetRegionFilter(region)
+		result := ElasticcacheSkuManager.syncDBInstanceSkus(ctx, userCred, region, meta)
+		notes := fmt.Sprintf("syncElasticCacheSkusByRegion %s result: %s", region.Name, result.Result())
+		log.Infof(notes)
+	}
+}
+
+// 同步Region elasticcache sku列表.
+func syncElasticCacheSkusByRegion(ctx context.Context, userCred mcclient.TokenCredential, region *SCloudregion) {
+	meta, err := fetchSkuResourcesMeta()
+	if err != nil {
+		log.Errorf("syncElasticCacheSkusByRegion.fetchSkuResourcesMeta %s", err)
+		return
+	}
+
+	meta.SetRegionFilter(region)
+	result := ElasticcacheSkuManager.syncDBInstanceSkus(ctx, userCred, region, meta)
+	notes := fmt.Sprintf("syncElasticCacheSkusByRegion %s result: %s", region.Name, result.Result())
+	log.Infof(notes)
+}
+
+func fetchSkuResourcesMeta() (*SSkuResourcesMeta, error) {
+	s := auth.GetAdminSession(context.Background(), options.Options.Region, "")
+	meta, err := modules.OfflineCloudmeta.GetSkuSourcesMeta(s)
+	if err != nil {
+		return nil, errors.Wrap(err, "fetchSkuSourceUrls.GetSkuSourcesMeta")
+	}
+
+	ret := &SSkuResourcesMeta{}
+	err = meta.Unmarshal(ret)
+	if err != nil {
+		return nil, errors.Wrap(err, "fetchSkuSourceUrls.Unmarshal")
+	}
+
+	return ret, nil
+}

pkg/compute/models/skustools.go

@@ -18,8 +18,6 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"io/ioutil"
-	"net/http"
 	"strings"
 
 	"yunion.io/x/jsonutils"
@@ -36,10 +34,6 @@ import (
 	"yunion.io/x/onecloud/pkg/mcclient/modules"
 )
 
-const (
-	ELASTICCACHE_SKUS_URL = "http://10.168.222.228:8000/ec.json"
-)
-
 type ServerSkus struct {
 	zone    *SkusZone
 	skus    []jsonutils.JSONObject
@@ -48,14 +42,6 @@ type ServerSkus struct {
 	created int
 }
 
-type ElasticCacheSkus struct {
-	zone    *SkusZone
-	skus    []jsonutils.JSONObject
-	total   int
-	updated int
-	created int
-}
-
 type SkusZone struct {
 	Provider         string
 	RegionId         string
@@ -456,215 +442,3 @@ func diff(origins, compares []string) []string {
 
 	return ret
 }
-
-func doCreateElasticCacheSku(data SElasticcacheSku) error {
-	if err := ElasticcacheSkuManager.TableSpec().Insert(&data); err != nil {
-		log.Debugf("SkusZone doCreate fail: %s", err.Error())
-		return err
-	}
-
-	return nil
-}
-
-func doUpdateElasticCacheSku(odata *SElasticcacheSku, sku jsonutils.JSONObject) error {
-	_, err := db.Update(odata, func() error {
-		if err := sku.Unmarshal(&odata); err != nil {
-			return err
-		}
-
-		return nil
-	})
-
-	if err != nil {
-		log.Debugf("SkusZone doUpdate fail: %s", err.Error())
-		return err
-	}
-
-	return nil
-}
-
-func processElasticCacheSkuData(ndata jsonutils.JSONObject) (jsonutils.JSONObject, error) {
-	// 将external_id 统一替换成 id
-	data, ok := ndata.(*jsonutils.JSONDict)
-	if !ok {
-		log.Debugf("invalid sku dict data: %s", ndata)
-	}
-
-	// 将external_id 统一替换成 id.
-	id, err := ndata.GetString("id")
-	if err != nil {
-		data.Set("external_id", jsonutils.NewString(""))
-	} else {
-		data.Set("external_id", jsonutils.NewString(id))
-		data.Remove("id")
-	}
-
-	provider, _ := ndata.GetString("provider")
-	region, _ := ndata.GetString("region_id")
-	masterZone, _ := ndata.GetString("master_zone_id")
-	slaveZone, _ := ndata.GetString("slave_zone_id")
-
-	if len(masterZone) > 0 {
-		masterZoneId := strings.Join([]string{provider, region, masterZone}, "/")
-		obj, err := db.FetchByExternalId(ZoneManager, masterZoneId)
-		if err == nil {
-			zone := obj.(*SZone)
-			data.Set("cloudregion_id", jsonutils.NewString(zone.CloudregionId))
-			data.Set("zone_id", jsonutils.NewString(zone.Id))
-
-			if len(slaveZone) > 0 {
-				slaveZoneId := strings.Join([]string{provider, region, slaveZone}, "/")
-				obj, err := db.FetchByExternalId(ZoneManager, slaveZoneId)
-				if err == nil {
-					zone := obj.(*SZone)
-					data.Set("slave_zone_id", jsonutils.NewString(zone.Id))
-				} else {
-					return nil, fmt.Errorf("slave zone with external id %s not found", slaveZoneId)
-				}
-			}
-		} else {
-			return nil, fmt.Errorf("master zone with external id %s not found", masterZoneId)
-		}
-	} else {
-		regionId := strings.Join([]string{provider, region}, "/")
-		obj, err := db.FetchByExternalId(CloudregionManager, regionId)
-		if err == nil {
-			region := obj.(*SCloudregion)
-			data.Set("cloudregion_id", jsonutils.NewString(region.Id))
-			data.Set("zone_id", jsonutils.NewString(""))
-		} else {
-			return nil, fmt.Errorf("region with external id %s not found", regionId)
-		}
-	}
-
-	return data, nil
-}
-
-// 全量同步elasticcache sku列表.
-func SyncElasticCacheSkus(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) {
-	skus := fetchExternalElasticCacheSkus()
-	filtedSkus := []jsonutils.JSONObject{}
-	for i := range skus {
-		if data, err := processElasticCacheSkuData(skus[i]); err == nil {
-			filtedSkus = append(filtedSkus, data)
-		} else {
-			log.Debugf("%s", err)
-		}
-	}
-
-	// 更新已经soldout的sku
-	localIds, err := ElasticcacheSkuManager.FetchAllAvailableSkuId()
-	if err != nil {
-		log.Errorf("SyncElasticCacheSkus FetchAllAvailableSkuId %s", err)
-		return
-	}
-
-	updateElasticCacheSkus(localIds, filtedSkus)
-}
-
-// 同步Region elasticcache sku列表.
-func syncElasticCacheSkusByRegion(region *SCloudregion) {
-	skus := fetchExternalElasticCacheSkus()
-	filtedSkus := []jsonutils.JSONObject{}
-	for i := range skus {
-		provider, _ := skus[i].GetString("provider")
-		regionId, _ := skus[i].GetString("region_id")
-		if region.GetExternalId() != fmt.Sprintf("%s/%s", provider, regionId) {
-			continue
-		}
-
-		if data, err := processElasticCacheSkuData(skus[i]); err == nil {
-			filtedSkus = append(filtedSkus, data)
-		} else {
-			log.Debugf("%s", err)
-		}
-	}
-
-	// 更新已经soldout的sku
-	localIds, err := ElasticcacheSkuManager.FetchAllAvailableSkuId()
-	if err != nil {
-		log.Errorf("SyncElasticCacheSkus FetchAllAvailableSkuId %s", err)
-		return
-	}
-
-	updateElasticCacheSkus(localIds, filtedSkus)
-}
-
-func fetchExternalElasticCacheSkus() []jsonutils.JSONObject {
-	resp, err := http.Get(ELASTICCACHE_SKUS_URL)
-	if err != nil {
-		log.Errorf("SyncElasticCacheSkus %s", err)
-		return nil
-	}
-
-	defer resp.Body.Close()
-	content, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		log.Errorf("SyncElasticCacheSkus ReadAll %s", err)
-		return nil
-	}
-
-	elasticcache, err := jsonutils.Parse(content)
-	if err != nil {
-		log.Errorf("SyncElasticCacheSkus Parse %s", err)
-		return nil
-	}
-
-	skus := []jsonutils.JSONObject{}
-	err = elasticcache.Unmarshal(&skus)
-	if err != nil {
-		log.Errorf("SyncElasticCacheSkus Unmarshal %s", err)
-		return nil
-	}
-
-	return skus
-}
-
-func updateElasticCacheSkus(localSkuIds []string, extSkus []jsonutils.JSONObject) {
-	// 本次已被更新的sku id
-	updatedIds := make([]string, 0)
-	for _, sku := range extSkus {
-		name, _ := sku.GetString("name")
-		extId, _ := sku.GetString("external_id")
-		if obj, err := db.FetchByExternalId(ElasticcacheSkuManager, extId); err != nil {
-			if err != sql.ErrNoRows {
-				log.Debugf("SyncToLocalDB external id %s name %s : %s", extId, name, err.Error())
-				return
-			}
-
-			data := SElasticcacheSku{}
-			if e := sku.Unmarshal(&data); e != nil {
-				log.Debugf("sku Unmarshal failed: %s, %s", sku, e.Error())
-				return
-			}
-
-			if err := doCreateElasticCacheSku(data); err != nil {
-				log.Debugf("sku doCreate failed: %s, %s", sku, err)
-				return
-			}
-		} else {
-			odata, ok := obj.(*SElasticcacheSku)
-			if !ok {
-				log.Debugf("SkusZone model assertion error. %s", obj)
-				return
-			}
-
-			if err := doUpdateElasticCacheSku(odata, sku); err != nil {
-				log.Debugf("SkusZone model doUpdate error. %s", err)
-				return
-			}
-
-			updatedIds = append(updatedIds, odata.Id)
-		}
-	}
-
-	// 处理已经下架的sku： 将本次未更新且处于available状态的sku置为soldout状态
-	abandonIds := diff(localSkuIds, updatedIds)
-	log.Debugf("SyncToLocalDB abandon sku %s", abandonIds)
-	err := ElasticcacheSkuManager.MarkAllAsSoldout(abandonIds)
-	if err != nil {
-		return
-	}
-
-	return
-}

pkg/compute/service/service.go

@@ -89,6 +89,7 @@ func StartService() {
 		cron.AddJobEveryFewHour("AutoSyncExtDiskSnapshot", 1, 10, 0, models.DiskManager.AutoSyncExtDiskSnapshot, false)
 		cron.AddJobEveryFewDays("SyncSkus", opts.SyncSkusDay, opts.SyncSkusHour, 0, 0, models.SyncSkus, true)
 		cron.AddJobEveryFewDays("SyncDBInstanceSkus", opts.SyncSkusDay, opts.SyncSkusHour, 0, 0, models.SyncDBInstanceSkus, true)
+		cron.AddJobEveryFewDays("SyncElasticCacheSkus", opts.SyncSkusDay, opts.SyncSkusHour, 0, 0, models.SyncElasticCacheSkus, true)
 		cron.AddJobEveryFewDays("StorageSnapshotsRecycle", 1, 2, 0, 0, models.StorageManager.StorageSnapshotsRecycle, false)
 
 		cron.Start()

pkg/mcclient/modules/managers.go

@@ -158,6 +158,12 @@ func NewYunionmetaManager(keyword, keywordPlural string, columns, adminColumns [
 		Keyword:     keyword, KeywordPlural: keywordPlural}
 }
 
+func NewOfflineCloudmetaManager(keyword, keywordPlural string, columns, adminColumns []string) modulebase.ResourceManager {
+	return modulebase.ResourceManager{
+		BaseManager: *modulebase.NewBaseManager("offlinecloudmeta", "", "", columns, adminColumns),
+		Keyword:     keyword, KeywordPlural: keywordPlural}
+}
+
 func NewAnsibleManager(keyword, keywordPlural string, columns, adminColumns []string) modulebase.ResourceManager {
 	return modulebase.ResourceManager{
 		BaseManager: *modulebase.NewBaseManager("ansible", "", "", columns, adminColumns),

pkg/mcclient/modules/mod_skus.go

@@ -16,8 +16,13 @@ package modules
 
 import (
 	"fmt"
+	"io/ioutil"
+	"net/http"
 	"strings"
 
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/pkg/errors"
+
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/modulebase"
 )
@@ -26,6 +31,10 @@ type SkusManager struct {
 	modulebase.ResourceManager
 }
 
+type OfflineCloudmetaManager struct {
+	modulebase.ResourceManager
+}
+
 type ServerSkusManager struct {
 	modulebase.ResourceManager
 }
@@ -35,9 +44,10 @@ type ElasticcacheSkusManager struct {
 }
 
 var (
-	CloudmetaSkus    SkusManager
-	ServerSkus       ServerSkusManager
-	ElasticcacheSkus ElasticcacheSkusManager
+	CloudmetaSkus    SkusManager             // meta.yunion.io
+	OfflineCloudmeta OfflineCloudmetaManager // aliyun offine sku&rate data
+	ServerSkus       ServerSkusManager       // region service: server sku manager
+	ElasticcacheSkus ElasticcacheSkusManager // region service: elasitc cache sku manager
 )
 
 func init() {
@@ -45,6 +55,10 @@ func init() {
 		[]string{},
 		[]string{})}
 
+	OfflineCloudmeta = OfflineCloudmetaManager{NewOfflineCloudmetaManager("", "",
+		[]string{},
+		[]string{})}
+
 	ServerSkus = ServerSkusManager{NewComputeManager("serversku", "serverskus",
 		[]string{"ID", "Name", "Instance_type_family", "Instance_type_category", "Cpu_core_count",
 			"Memory_size_mb", "Os_name", "Sys_disk_resizable", "Sys_disk_type",
@@ -75,3 +89,29 @@ func (self *SkusManager) GetSkus(s *mcclient.ClientSession, providerId, regionId
 
 	return ret, nil
 }
+
+func (self *OfflineCloudmetaManager) GetSkuSourcesMeta(s *mcclient.ClientSession) (jsonutils.JSONObject, error) {
+	baseUrl, err := s.GetServiceVersionURL(self.ServiceType(), self.EndpointType(), self.GetApiVersion())
+	if err != nil {
+		return nil, err
+	}
+	baseUrl = strings.TrimLeft(baseUrl, "/")
+	metaUrl := strings.Join([]string{baseUrl, "sku.meta"}, "/")
+	resp, err := http.Get(metaUrl)
+	if err != nil {
+		return nil, errors.Wrap(err, "OfflineCloudmetaManager.GetSkuSourcesMeta.Get")
+	}
+
+	defer resp.Body.Close()
+	content, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, errors.Wrap(err, "OfflineCloudmetaManager.GetSkuSourcesMeta.ReadAll")
+	}
+
+	meta, err := jsonutils.Parse(content)
+	if err != nil {
+		return nil, errors.Wrap(err, "OfflineCloudmetaManager.GetSkuSourcesMeta.Parse")
+	}
+
+	return meta, nil
+}