fix(scheduler): watch serversku (#23328)

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")
@@ -127,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 {

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){