floraachy
/
pcm-coordinator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updated aiScheduler 

Former-commit-id: 9b8d271bb13c6cfb7273ec76084ce8c123fbe3be
This commit is contained in:
tzwang 2024-01-31 11:24:00 +08:00
parent 75543eacae
commit b04a3f1255
10 changed files with 93 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
api/internal/scheduler/schedulers/aiScheduler.go
View File

@ -17,11 +17,10 @@ package schedulers
import (
"errors"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/algorithm/providerPricing"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/entity"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/schedulers/option"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/service/collector"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy/param"
"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/utils"
@ -50,36 +49,24 @@ func (as *AiScheduler) GetNewStructForDb(task *response.TaskInfo, resource strin
}
func (as *AiScheduler) PickOptimalStrategy() (strategy.Strategy, error) {
resources, err := as.findClustersWithResource()
resources, err := as.findClustersWithResources()
if err != nil {
return nil, err
}
if len(resources) == 0 {
return nil, errors.New("no cluster has resources")
}
params := &param.Params{Resources: resources}
if len(resources) < 2 /*|| as.task */ {
var pros []entity.Participant
for _, resource := range resources {
pros = append(pros, entity.Participant{
Participant_id: resource.ParticipantId,
Name: resource.Name,
})
}
strategy := strategy.NewReplicationStrategy(nil)
strategy := strategy.NewReplicationStrategy(&param.ReplicationParams{Params: params /*, Replicas: 1*/})
return strategy, nil
}
task, providerList := as.genTaskAndProviders()
if err != nil {
return nil, nil
}
strategy := strategy.NewPricingStrategy(task, providerList...)
strategy := strategy.NewPricingStrategy(&param.ResourcePricingParams{Params: params})
return strategy, nil
}
func (as *AiScheduler) genTaskAndProviders() (*providerPricing.Task, []*providerPricing.Provider) {
return nil, nil
}
func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) error {
if clusters == nil {
return errors.New("clusters is nil")
@ -97,7 +84,7 @@ func (as *AiScheduler) AssignTask(clusters []*strategy.AssignedCluster) error {
return nil
}
func (as *AiScheduler) findClustersWithResource() ([]*collector.ResourceSpecs, error) {
func (as *AiScheduler) findClustersWithResources() ([]*collector.ResourceSpecs, error) {
var resourceSpecs []*collector.ResourceSpecs
for _, resourceCollector := range *as.ResourceCollector {
spec, err := resourceCollector.GetResourceSpecs()

8
api/internal/scheduler/schedulers/cloudScheduler.go
View File

@ -19,6 +19,7 @@ import (
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/algorithm/providerPricing"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/database"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy/param"
"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/utils"
@ -38,12 +39,9 @@ func NewCloudScheduler() *CloudScheduler {
}
func (cs *CloudScheduler) PickOptimalStrategy() (strategy.Strategy, error) {
task, providerList, err := cs.genTaskAndProviders()
if err != nil {
return nil, nil
}
//获取所有计算中心
//调度算法
strategy := strategy.NewPricingStrategy(task, providerList...)
strategy := strategy.NewPricingStrategy(&param.ResourcePricingParams{})
return strategy, nil
}

4
api/internal/scheduler/strategy/params/params.go → api/internal/scheduler/strategy/param/params.go
View File

@ -1,9 +1,9 @@
package params
package param
import (
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/service/collector"
)
type Params struct {
resources []*collector.ResourceSpecs
Resources []*collector.ResourceSpecs
}

23
api/internal/scheduler/strategy/param/replication.go Normal file
View File

@ -0,0 +1,23 @@
package param
import "gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/entity"
type ReplicationParams struct {
Replicas int32
*Params
}
func (r *ReplicationParams) GetReplicas() int32 {
return r.Replicas
}
func (r *ReplicationParams) GetParticipants() []*entity.Participant {
var participants []*entity.Participant
for _, resource := range r.Resources {
participants = append(participants, &entity.Participant{
Participant_id: resource.ParticipantId,
Name: resource.Name,
})
}
return participants
}

32
api/internal/scheduler/strategy/param/resourcePricing.go Normal file
View File

@ -0,0 +1,32 @@
package param
import (
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/algorithm/providerPricing"
)
type ResourcePricingParams struct {
replicas int32
task *providerPricing.Task
*Params
}
func (r *ResourcePricingParams) GetReplicas() int32 {
return r.replicas
}
func (r *ResourcePricingParams) GetTask() *providerPricing.Task {
return r.task
}
func (r *ResourcePricingParams) GetProviders() []*providerPricing.Provider {
var providerList []*providerPricing.Provider
for _, resource := range r.Resources {
provider := providerPricing.NewProvider(
resource.ParticipantId,
resource.CpuAvail,
resource.MemAvail,
resource.DiskAvail, 0.0, 0.0, 0.0)
providerList = append(providerList, provider)
}
return providerList
}

16
api/internal/scheduler/strategy/params/replicationParams.go
View File

@ -1,16 +0,0 @@
package params
import "gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/entity"
type ReplicationOption struct {
replicas int32
participants []entity.Participant
}
func (o *ReplicationOption) GetReplicas() int32 {
return o.replicas
}
func (o *ReplicationOption) GetParticipants() []entity.Participant {
return o.participants
}

26
api/internal/scheduler/strategy/params/resourcePricingParams.go
View File

@ -1,26 +0,0 @@
package params
import (
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/algorithm/providerPricing"
)
type ResourcePricingOption struct {
replicas int32
task *providerPricing.Task
providers []*providerPricing.Provider
*Params
}
func NewResourcePricingOption(params *Params) *ResourcePricingOption {
return &ResourcePricingOption{
Params: params,
}
}
func (r *ResourcePricingOption) GetReplicas() int32 {
return r.replicas
}
func (r *ResourcePricingOption) GetProviders() []*providerPricing.Provider {
return r.providers
}

6
api/internal/scheduler/strategy/replication.go
View File

@ -3,15 +3,15 @@ package strategy
import (
"github.com/pkg/errors"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/entity"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy/params"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy/param"
)
type ReplicationStrategy struct {
replicas int32
participants []entity.Participant
participants []*entity.Participant
}
func NewReplicationStrategy(params *params.ReplicationOption) *ReplicationStrategy {
func NewReplicationStrategy(params *param.ReplicationParams) *ReplicationStrategy {
return &ReplicationStrategy{replicas: params.GetReplicas(),
participants: params.GetParticipants(),
}

5
api/internal/scheduler/strategy/resourcePricing.go
View File

@ -17,6 +17,7 @@ package strategy
import (
"errors"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/algorithm/providerPricing"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy/param"
)
type PricingStrategy struct {
@ -25,7 +26,9 @@ type PricingStrategy struct {
StrategyList []*providerPricing.Strategy
}
func NewPricingStrategy(task *providerPricing.Task, providers ...*providerPricing.Provider) *PricingStrategy {
func NewPricingStrategy(params *param.ResourcePricingParams) *PricingStrategy {
providers := params.GetProviders()
task := params.GetTask()
var providerList []*providerPricing.Provider
var res [][]int

18
api/internal/scheduler/strategy/test/strategy_test.go
View File

@ -3,7 +3,9 @@ package test
import (
"fmt"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/entity"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/service/collector"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy"
"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/scheduler/strategy/param"
"testing"
)
@ -13,10 +15,23 @@ func TestReplication(t *testing.T) {
{Name: "test2", Participant_id: 2},
{Name: "test3", Participant_id: 3},
}
res := []*collector.ResourceSpecs{
{
ParticipantId: 1,
Name: "test1",
},
{
ParticipantId: 1,
Name: "test2"},
{
ParticipantId: 1,
Name: "test3"},
}
tests := []struct {
name string
replica int32
ps []entity.Participant
res []*collector.ResourceSpecs
}{
{
name: "test1",
@ -32,7 +47,8 @@ func TestReplication(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
repl := strategy.NewReplicationStrategy(nil)
params := &param.Params{Resources: res}
repl := strategy.NewReplicationStrategy(&param.ReplicationParams{Params: params, Replicas: tt.replica})
schedule, err := repl.Schedule()
if err != nil {
return