Merge pull request 'scheduler refactor updated' (#4) from tzwang/pcm-coordinator:master into master

Former-commit-id: f95e482ae3ffa2fae8699ba11e8c609b87c60691

pkg/scheduler/aiScheduler.go

@@ -18,19 +18,21 @@ import (
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/pkg/response"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/models"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/algorithm/providerPricing"
+	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/collector"
+	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/strategies"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/utils"
-	"gorm.io/gorm"
 )
 
-type aiScheduler struct {
+type AiScheduler struct {
 	yamlString string
+	collector  collector.ResourceCollector
 }
 
-func NewAiScheduler(val string) *aiScheduler {
+func NewAiScheduler(val string) *AiScheduler {
-	return &aiScheduler{yamlString: val}
+	return &AiScheduler{yamlString: val}
 }
 
-func (as *aiScheduler) getNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) {
+func (as *AiScheduler) getNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) {
 	ai := models.Ai{
 		ParticipantId: participantId,
 		TaskId:        task.TaskId,
@@ -41,22 +43,16 @@ func (as *aiScheduler) getNewStructForDb(task *response.TaskInfo, resource strin
 	return ai, nil
 }
 
-func (as *aiScheduler) pickOptimalStrategy(task *providerPricing.Task, providers ...*providerPricing.Provider) (*providerPricing.Strategy, error) {
+func (as *AiScheduler) pickOptimalStrategy() (strategies.Strategy, error) {
+	//a, b := as.genTaskAndProviders()
+
 	return nil, nil
 }
 
-func (as *aiScheduler) genTaskAndProviders(task *response.TaskInfo, dbEngin *gorm.DB) (*providerPricing.Task, []*providerPricing.Provider) {
+func (as *AiScheduler) genTaskAndProviders() (*providerPricing.Task, []*providerPricing.Provider) {
-	var proParams []providerParams
+	return nil, nil
-	sqlstr := "SELECT SUM(a.disk_avail) as disk_avail,SUM(a.mem_avail) as mem_avail,SUM(a.cpu_total * a.cpu_usable) as cpu_avail,participant_id from (SELECT * from sc_node_avail_info where created_time in (SELECT MAX(created_time) as time  from sc_node_avail_info where deleted_flag = 0 GROUP BY participant_id,node_name)) a GROUP BY a.participant_id"
+}
-	dbEngin.Raw(sqlstr).Scan(&proParams)
+
-
+func (as *AiScheduler) assignTask() error {
-	var providerList []*providerPricing.Provider
+	return nil
-	for _, p := range proParams {
-		provider := providerPricing.NewProvider(p.Participant_id, p.Cpu_avail, p.Mem_avail, p.Disk_avail, 0.0, 0.0, 0.0)
-		providerList = append(providerList, provider)
-	}
-
-	t := providerPricing.NewTask(0, 1, 2, 75120000, 301214500, 1200, 2, 6, 2000)
-
-	return t, providerList
 }

pkg/scheduler/cloudScheduler.go

@@ -19,9 +19,9 @@ import (
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/pkg/response"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/models"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/algorithm/providerPricing"
+	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/database"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/strategies"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/utils"
-	"gorm.io/gorm"
 	"io"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -29,25 +29,25 @@ import (
 	kyaml "k8s.io/apimachinery/pkg/util/yaml"
 )
 
-type cloudScheduler struct {
+type CloudScheduler struct {
+	storage database.Storage
 }
 
-func NewCloudScheduler() *cloudScheduler {
+func NewCloudScheduler() *CloudScheduler {
-	return &cloudScheduler{}
+	return &CloudScheduler{}
 }
 
-func (cs *cloudScheduler) pickOptimalStrategy(task *providerPricing.Task, providers ...*providerPricing.Provider) (*providerPricing.Strategy, error) {
+func (cs *CloudScheduler) pickOptimalStrategy() (strategies.Strategy, error) {
-
+	task, providerList, err := cs.genTaskAndProviders()
-	//调度算法
-	strategy := strategies.NewPricingStrategy(task, providers...)
-	taskResult, err := strategy.ScheduleWithFullCollaboration()
 	if err != nil {
-		return nil, err
+		return nil, nil
 	}
-	return taskResult.MaxscoreStrategy, nil
+	//调度算法
+	strategy := strategies.NewPricingStrategy(task, providerList...)
+	return strategy, nil
 }
 
-func (cs *cloudScheduler) getNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) {
+func (cs *CloudScheduler) getNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) {
 	cloud := cs.UnMarshalK8sStruct(resource, task.TaskId, task.NsID)
 	cloud.Id = utils.GenSnowflakeID()
 	cloud.NsID = task.NsID
@@ -56,7 +56,7 @@ func (cs *cloudScheduler) getNewStructForDb(task *response.TaskInfo, resource st
 	return cloud, nil
 }
 
-func (cs *cloudScheduler) UnMarshalK8sStruct(yamlString string, taskId int64, nsID string) models.Cloud {
+func (cs *CloudScheduler) UnMarshalK8sStruct(yamlString string, taskId int64, nsID string) models.Cloud {
 	var cloud models.Cloud
 	d := kyaml.NewYAMLOrJSONDecoder(bytes.NewBufferString(yamlString), 4096)
 	var err error
@@ -100,11 +100,11 @@ func (cs *cloudScheduler) UnMarshalK8sStruct(yamlString string, taskId int64, ns
 	return cloud
 }
 
-func (cs *cloudScheduler) genTaskAndProviders(task *response.TaskInfo, dbEngin *gorm.DB) (*providerPricing.Task, []*providerPricing.Provider) {
+func (cs *CloudScheduler) genTaskAndProviders() (*providerPricing.Task, []*providerPricing.Provider, error) {
-	var proParams []providerParams
+	proParams, err := cs.storage.GetProviderParams()
-	sqlstr := "SELECT SUM(a.disk_avail) as disk_avail,SUM(a.mem_avail) as mem_avail,SUM(a.cpu_total * a.cpu_usable) as cpu_avail,participant_id from (SELECT * from sc_node_avail_info where created_time in (SELECT MAX(created_time) as time  from sc_node_avail_info where deleted_flag = 0 GROUP BY participant_id,node_name)) a GROUP BY a.participant_id"
+	if err != nil {
-	dbEngin.Raw(sqlstr).Scan(&proParams)
+		return nil, nil, nil
-
+	}
 	var providerList []*providerPricing.Provider
 	for _, p := range proParams {
 		provider := providerPricing.NewProvider(p.Participant_id, p.Cpu_avail, p.Mem_avail, p.Disk_avail, 0.0, 0.0, 0.0)
@@ -114,5 +114,9 @@ func (cs *cloudScheduler) genTaskAndProviders(task *response.TaskInfo, dbEngin *
 	//replicas := task.Metadata.(map[string]interface{})["spec"].(map[string]interface{})["replicas"].(float64)
 	//t := algorithm.NewTask(0, int(replicas), 2, 75120000, 301214500, 1200, 2, 6, 2000)
 
-	return nil, providerList
+	return nil, providerList, nil
+}
+
+func (cs *CloudScheduler) assignTask() error {
+	return nil
 }

pkg/scheduler/collector/aiCollector.go

@@ -0,0 +1,8 @@
+package collector
+
+type AiCollector struct {
+}
+
+func (a *AiCollector) getResourceSpecs() {
+
+}

pkg/scheduler/collector/collector.go

@@ -0,0 +1,8 @@
+package collector
+
+type ResourceCollector interface {
+	getResourceSpecs()
+}
+
+type ResourceSpecs struct {
+}

pkg/scheduler/common.go

@@ -16,23 +16,15 @@ package scheduler
 
 import (
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/pkg/response"
-	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/algorithm/providerPricing"
+	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/strategies"
-	"gorm.io/gorm"
 	"math/rand"
 	"time"
 )
 
 type scheduleService interface {
 	getNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error)
-	pickOptimalStrategy(task *providerPricing.Task, providers ...*providerPricing.Provider) (*providerPricing.Strategy, error)
+	pickOptimalStrategy() (strategies.Strategy, error)
-	genTaskAndProviders(task *response.TaskInfo, dbEngin *gorm.DB) (*providerPricing.Task, []*providerPricing.Provider)
+	assignTask() error
-}
-
-type providerParams struct {
-	Disk_avail     float64
-	Mem_avail      float64
-	Cpu_avail      float64
-	Participant_id int64
 }
 
 // 求交集

pkg/scheduler/database/aiStorage.go

@@ -0,0 +1,8 @@
+package database
+
+type AiStorage struct {
+}
+
+func (s *AiStorage) getParticipants() {
+
+}

pkg/scheduler/database/cloudStorage.go

@@ -0,0 +1,28 @@
+package database
+
+import "gorm.io/gorm"
+
+type CloudStorage struct {
+	dbEngin *gorm.DB
+}
+
+func NewCloudStorage(dbEngin *gorm.DB) *CloudStorage {
+	return &CloudStorage{dbEngin: dbEngin}
+}
+
+func (c *CloudStorage) GetProviderParams() ([]providerParams, error) {
+	var proParams []providerParams
+	sqlstr := "SELECT SUM(a.disk_avail) as disk_avail,SUM(a.mem_avail) as mem_avail,SUM(a.cpu_total * a.cpu_usable) as cpu_avail,participant_id from (SELECT * from sc_node_avail_info where created_time in (SELECT MAX(created_time) as time  from sc_node_avail_info where deleted_flag = 0 GROUP BY participant_id,node_name)) a GROUP BY a.participant_id"
+	c.dbEngin.Raw(sqlstr).Scan(&proParams)
+	if len(proParams) == 0 {
+		return nil, nil
+	}
+	return proParams, nil
+}
+
+type providerParams struct {
+	Disk_avail     float64
+	Mem_avail      float64
+	Cpu_avail      float64
+	Participant_id int64
+}

pkg/scheduler/database/storage.go

@@ -0,0 +1,5 @@
+package database
+
+type Storage interface {
+	GetProviderParams() ([]providerParams, error)
+}

pkg/scheduler/hpcScheduler.go

@@ -19,19 +19,19 @@ import (
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/constants"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/models"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/algorithm/providerPricing"
+	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/strategies"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/utils"
-	"gorm.io/gorm"
 )
 
-type hpcScheduler struct {
+type HpcScheduler struct {
 	yamlString string
 }
 
-func NewHpcScheduler(val string) *hpcScheduler {
+func NewHpcScheduler(val string) *HpcScheduler {
-	return &hpcScheduler{yamlString: val}
+	return &HpcScheduler{yamlString: val}
 }
 
-func (h *hpcScheduler) getNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) {
+func (h *HpcScheduler) getNewStructForDb(task *response.TaskInfo, resource string, participantId int64) (interface{}, error) {
 	hpc := models.Hpc{}
 	utils.Convert(task.Metadata, &hpc)
 	hpc.Id = utils.GenSnowflakeID()
@@ -42,22 +42,14 @@ func (h *hpcScheduler) getNewStructForDb(task *response.TaskInfo, resource strin
 	return hpc, nil
 }
 
-func (h *hpcScheduler) pickOptimalStrategy(task *providerPricing.Task, providers ...*providerPricing.Provider) (*providerPricing.Strategy, error) {
+func (h *HpcScheduler) pickOptimalStrategy() (strategies.Strategy, error) {
 	return nil, nil
 }
 
-func (h *hpcScheduler) genTaskAndProviders(task *response.TaskInfo, dbEngin *gorm.DB) (*providerPricing.Task, []*providerPricing.Provider) {
+func (h *HpcScheduler) genTaskAndProviders(task *response.TaskInfo) (*providerPricing.Task, []*providerPricing.Provider) {
-	var proParams []providerParams
+	return nil, nil
-	sqlstr := "SELECT SUM(a.disk_avail) as disk_avail,SUM(a.mem_avail) as mem_avail,SUM(a.cpu_total * a.cpu_usable) as cpu_avail,participant_id from (SELECT * from sc_node_avail_info where created_time in (SELECT MAX(created_time) as time  from sc_node_avail_info where deleted_flag = 0 GROUP BY participant_id,node_name)) a GROUP BY a.participant_id"
+}
-	dbEngin.Raw(sqlstr).Scan(&proParams)
+
-
+func (h *HpcScheduler) assignTask() error {
-	var providerList []*providerPricing.Provider
+	return nil
-	for _, p := range proParams {
-		provider := providerPricing.NewProvider(p.Participant_id, p.Cpu_avail, p.Mem_avail, p.Disk_avail, 0.0, 0.0, 0.0)
-		providerList = append(providerList, provider)
-	}
-
-	t := providerPricing.NewTask(0, 1, 2, 75120000, 301214500, 1200, 2, 6, 2000)
-
-	return t, providerList
 }

pkg/scheduler/scheduler.go

@@ -19,7 +19,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/zeromicro/go-zero/core/logx"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/pkg/response"
-	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/algorithm/providerPricing"
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/rpc/client/participantservice"
 	"gorm.io/gorm"
 	"sigs.k8s.io/yaml"
@@ -133,28 +132,16 @@ func (s *scheduler) AssignAndSchedule() error {
 	}
 
 	//生成算法所需参数
-	task, providerList, err := s.obtainParamsForStrategy()
+	//task, providerList, err := s.obtainParamsForStrategy()
-	if err != nil {
+	//if err != nil {
-		return err
+	//	return err
-	}
+	//}
 
 	//集群数量不满足，指定到标签匹配后第一个集群
-	if len(providerList) < 2 {
+	//if len(providerList) < 2 {
-		s.task.ParticipantId = s.participantIds[0]
+	//	s.task.ParticipantId = s.participantIds[0]
-		return nil
+	//	return nil
-	}
+	//}
-
-	//调度算法
-	strategy, err := s.scheduleService.pickOptimalStrategy(task, providerList...)
-	if err != nil {
-		return err
-	}
-
-	//调度结果
-	err = s.assignReplicasToResult(strategy, providerList)
-	if err != nil {
-		return err
-	}
 
 	return nil
 }
@@ -179,33 +166,3 @@ func (s *scheduler) SaveToDb() error {
 	}
 	return nil
 }
-
-func (s *scheduler) obtainParamsForStrategy() (*providerPricing.Task, []*providerPricing.Provider, error) {
-	task, providerList := s.scheduleService.genTaskAndProviders(s.task, s.dbEngin)
-
-	if len(providerList) == 0 {
-		return nil, nil, errors.New("获取集群失败")
-	}
-
-	return task, providerList, nil
-}
-
-func (s *scheduler) assignReplicasToResult(strategy *providerPricing.Strategy, providerList []*providerPricing.Provider) error {
-
-	if len(strategy.Tasksolution) == 0 {
-		return errors.New("调度失败， 未能获取调度结果")
-	}
-
-	for i, e := range strategy.Tasksolution {
-		if e == 0 {
-			continue
-		}
-		s.result[providerList[i].Pid] = string(e)
-	}
-
-	if len(s.result) == 0 {
-		return errors.New("可用集群为空")
-	}
-
-	return nil
-}

pkg/scheduler/strategies/priceBasedStrategy.go

@@ -19,13 +19,13 @@ import (
 	"gitlink.org.cn/jcce-pcm/pcm-coordinator/pkg/scheduler/algorithm/providerPricing"
 )
 
-type pricingStrategy struct {
+type PricingStrategy struct {
 	ProviderList []*providerPricing.Provider
 	Task         *providerPricing.Task
 	StrategyList []*providerPricing.Strategy
 }
 
-func NewPricingStrategy(task *providerPricing.Task, providers ...*providerPricing.Provider) *pricingStrategy {
+func NewPricingStrategy(task *providerPricing.Task, providers ...*providerPricing.Provider) *PricingStrategy {
 	var providerList []*providerPricing.Provider
 	var res [][]int
 
@@ -56,10 +56,10 @@ func NewPricingStrategy(task *providerPricing.Task, providers ...*providerPricin
 		strategyList = append(strategyList, strategy)
 	}
 
-	return &pricingStrategy{ProviderList: providerList, Task: task, StrategyList: strategyList}
+	return &PricingStrategy{ProviderList: providerList, Task: task, StrategyList: strategyList}
 }
 
-func (ps *pricingStrategy) computeMaxScore() (*providerPricing.Task, error) {
+func (ps *PricingStrategy) computeMaxScore() (*providerPricing.Task, error) {
 	maxStrategy := providerPricing.NewStrategy()
 	var maxprofit float64
 
@@ -106,11 +106,7 @@ func (ps *pricingStrategy) computeMaxScore() (*providerPricing.Task, error) {
 	return ps.Task, nil
 }
 
-//type strategyService interface {
+func (ps *PricingStrategy) produceMaxScoreStrategy() (*providerPricing.Strategy, error) {
-//	computeMaxScore() (*providerPricing.Task, error)
-//}
-
-func (ps *pricingStrategy) ScheduleWithFullCollaboration() (*providerPricing.Task, error) {
 	task, err := ps.computeMaxScore()
 	if err != nil {
 		return nil, err
@@ -137,5 +133,31 @@ func (ps *pricingStrategy) ScheduleWithFullCollaboration() (*providerPricing.Tas
 		task.ResourcePerTask = append(task.ResourcePerTask, resourcePerTaskPerProviders)
 	}
 
-	return task, nil
+	return task.MaxscoreStrategy, nil
+}
+
+func (ps *PricingStrategy) Schedule() ([]*AssignedCluster, error) {
+	strategy, err := ps.produceMaxScoreStrategy()
+	if err != nil {
+		return nil, err
+	}
+
+	if len(strategy.Tasksolution) == 0 {
+		return nil, errors.New("调度失败， 未能获取调度结果")
+	}
+
+	var results []*AssignedCluster
+	for i, e := range strategy.Tasksolution {
+		if e == 0 {
+			continue
+		}
+		cluster := &AssignedCluster{ParticipantId: ps.ProviderList[i].Pid, Replicas: int32(e)}
+		results = append(results, cluster)
+	}
+
+	if len(results) == 0 {
+		return nil, errors.New("可用集群为空")
+	}
+
+	return results, nil
 }

pkg/scheduler/strategies/strategy.go

@@ -0,0 +1,14 @@
+package strategies
+
+type Strategy interface {
+	Schedule() ([]*AssignedCluster, error)
+}
+
+type AssignedCluster struct {
+	ParticipantId int64
+	Name          string
+	Replicas      int32
+}
+
+type Options struct {
+}