diff --git a/pkg/scheduler/data_manager/sku/sku.go b/pkg/scheduler/data_manager/sku/sku.go index 7cb0addfbb..d2d25a5a3a 100644 --- a/pkg/scheduler/data_manager/sku/sku.go +++ b/pkg/scheduler/data_manager/sku/sku.go @@ -15,31 +15,179 @@ package sku import ( + "context" "fmt" "sync" "time" "yunion.io/x/jsonutils" "yunion.io/x/log" + "yunion.io/x/pkg/errors" "yunion.io/x/pkg/util/wait" "yunion.io/x/sqlchemy" computeapi "yunion.io/x/onecloud/pkg/apis/compute" + "yunion.io/x/onecloud/pkg/cloudcommon/consts" "yunion.io/x/onecloud/pkg/compute/models" + "yunion.io/x/onecloud/pkg/mcclient/auth" + "yunion.io/x/onecloud/pkg/mcclient/informer" + "yunion.io/x/onecloud/pkg/mcclient/modules/compute" ) var ( skuManager *SSkuManager ) -func Start(refreshInterval time.Duration) { +func Start(ctx context.Context, refreshInterval time.Duration) { skuManager = &SSkuManager{ skuMap: newSkuMap(), refreshInterval: refreshInterval, } + skuManager.startWatch(ctx) skuManager.sync() } +func (m *SSkuManager) startWatch(ctx context.Context) { + s := auth.GetAdminSession(ctx, consts.GetRegion()) + informer.NewWatchManagerBySessionBg(s, func(man *informer.SWatchManager) error { + if err := man.For(compute.ServerSkus).AddEventHandler(ctx, newEventHandler(compute.ServerSkus, m)); err != nil { + return errors.Wrapf(err, "watch resource %s", compute.ServerSkus.KeyString()) + } + return nil + }) +} + +func newEventHandler(res informer.IResourceManager, dataMan *SSkuManager) informer.EventHandler { + return &eventHandler{ + res: res, + dataMan: dataMan, + } +} + +type eventHandler struct { + res informer.IResourceManager + dataMan *SSkuManager +} + +func (e eventHandler) keyword() string { + return e.res.GetKeyword() +} + +func (e eventHandler) newServerSkuFromJson(obj *jsonutils.JSONDict) (*models.SServerSku, error) { + sku := &models.SServerSku{} + if err := obj.Unmarshal(sku); err != nil { + return nil, errors.Wrapf(err, "unmarshal server sku by: %s", obj.String()) + } + return sku, nil +} + +func (e eventHandler) newServerSku(obj *jsonutils.JSONDict) (*models.SServerSku, error) { + sku, err := e.newServerSkuFromJson(obj) + if err != nil { + return nil, errors.Wrap(err, "newServerSkuFromJson") + } + if sku.PostpaidStatus == "" { + obj, err := models.ServerSkuManager.FetchById(sku.Id) + if err != nil { + return nil, errors.Wrapf(err, "fetch serversku by id %q", sku.Id) + } + sku = obj.(*models.SServerSku) + } + return sku, nil +} + +func isValidServerSku(sku *models.SServerSku) error { + if sku.PrepaidStatus != computeapi.SkuStatusAvailable && sku.PostpaidStatus != computeapi.SkuStatusAvailable { + return errors.Wrapf(errors.ErrInvalidStatus, "sku: %s, prepaid_status: %q, postpaid_status: %q", sku.Name, sku.PrepaidStatus, sku.PostpaidStatus) + } + if !sku.Enabled.IsTrue() { + return errors.Wrapf(errors.ErrInvalidStatus, "sku: %s, enabled: %q", sku.Name, sku.Enabled) + } + return nil +} + +func newServerSku(sku *models.SServerSku) *ServerSku { + return &ServerSku{ + Id: sku.Id, + Name: sku.Name, + RegionId: sku.CloudregionId, + ZoneId: sku.ZoneId, + Provider: sku.Provider, + } +} + +func (e eventHandler) addServerSku(obj *jsonutils.JSONDict) error { + dbSku, err := e.newServerSku(obj) + if err != nil { + return errors.Wrap(err, "new server sku") + } + if err := isValidServerSku(dbSku); err != nil { + return errors.Wrap(err, "invalid server sku") + } + + sku := newServerSku(dbSku) + log.Infof("add server sku %s", jsonutils.Marshal(sku).String()) + e.dataMan.skuMap.Add(sku.Name, sku) + return nil +} + +func (e eventHandler) OnAdd(obj *jsonutils.JSONDict) { + log.Infof("%s [CREATED]: \n%s", e.keyword(), obj.String()) + if err := e.addServerSku(obj); err != nil { + log.Errorf("add server sku error: %v", err) + return + } +} + +func (e eventHandler) deleteServerSku(sku *ServerSku) error { + e.dataMan.skuMap.Delete(sku.Name, sku) + return nil +} + +func (e eventHandler) updateServerSku(newObj *jsonutils.JSONDict) error { + newSku, err := e.newServerSku(newObj) + if err != nil { + return errors.Wrap(err, "new old server sku") + } + shouldDelete := false + err = isValidServerSku(newSku) + if err != nil { + shouldDelete = true + } + sku := newServerSku(newSku) + if shouldDelete { + log.Infof("delete server sku %s when updating, cause of %v", sku.Name, err) + if err := e.deleteServerSku(sku); err != nil { + return errors.Wrap(err, "delete server sku") + } + } else { + if err := e.addServerSku(newObj); err != nil { + return errors.Wrap(err, "add server sku") + } + } + return nil + +} + +func (e eventHandler) OnUpdate(oldObj, newObj *jsonutils.JSONDict) { + log.Infof("%s [UPDATED]: \n[NEW]: %s\n[OLD]: %s", e.keyword(), newObj.String(), oldObj.String()) + if err := e.updateServerSku(newObj); err != nil { + log.Errorf("update server sku error: %v", err) + } +} + +func (e eventHandler) OnDelete(obj *jsonutils.JSONDict) { + log.Infof("%s [DELETED]: \n%s", e.keyword(), obj.String()) + sku, err := e.newServerSkuFromJson(obj) + if err != nil { + log.Errorf("new server sku error: %v", err) + return + } + if err := e.deleteServerSku(newServerSku(sku)); err != nil { + log.Errorf("delete server sku error: %v", err) + } +} + func SyncOnce(wait bool) error { if skuManager == nil { return fmt.Errorf("sku manager not init") @@ -69,6 +217,7 @@ type ServerSku struct { Name string `json:"name"` RegionId string `json:"cloudregion_id"` ZoneId string `json:"zone_id"` + Provider string `json:"provider"` } type skuList []*ServerSku @@ -126,7 +275,24 @@ func (cache *skuMap) Add(instanceType string, sku *ServerSku) { skus = make([]*ServerSku, 0) } skus = append(skus, sku) - cache.Store(instanceType, skus) + cache.Store(instanceType, skuList(skus)) +} + +func (cache *skuMap) Delete(instanceType string, sku *ServerSku) { + skus := cache.Get(instanceType) + if len(skus) == 0 { + return + } + newSkus := make([]*ServerSku, 0) + for i := range skus { + curSku := skus[i] + if curSku.Id == sku.Id { + log.Infof("delete sku from cache: %s", jsonutils.Marshal(sku)) + continue + } + newSkus = append(newSkus, curSku) + } + cache.Store(instanceType, skuList(newSkus)) } type SSkuManager struct { diff --git a/pkg/scheduler/service/service.go b/pkg/scheduler/service/service.go index 1eae0f2474..f66dae84ad 100644 --- a/pkg/scheduler/service/service.go +++ b/pkg/scheduler/service/service.go @@ -117,7 +117,7 @@ func StartService() error { startSched := func() { stopEverything := make(chan struct{}) ctx := context.Background() - go skuman.Start(utils.ToDuration(o.Options.SkuRefreshInterval)) + go skuman.Start(ctx, utils.ToDuration(o.Options.SkuRefreshInterval)) go schedtag.Start(ctx, utils.ToDuration("30s")) for _, f := range []func(ctx context.Context){