pcm-coordinator/pkg/scheduler/cloudScheduler.go

/*

 Copyright (c) [2023] [pcm]
 [pcm-coordinator] is licensed under Mulan PSL v2.
 You can use this software according to the terms and conditions of the Mulan PSL v2.
 You may obtain a copy of Mulan PSL v2 at:
          http://license.coscl.org.cn/MulanPSL2
 THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 EITHER EXPaRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details.

*/

package scheduler

import (
	"bytes"
	"encoding/json"
	"errors"
	"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/algo"
	"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"
	syaml "k8s.io/apimachinery/pkg/runtime/serializer/yaml"
	kyaml "k8s.io/apimachinery/pkg/util/yaml"
)

type cloudScheduler struct {
}

func NewCloudScheduler() *cloudScheduler {
	return &cloudScheduler{}
}

func (cs *cloudScheduler) pickOptimalStrategy(task *algo.Task, providers ...*algo.Provider) (*algo.Strategy, error) {
	//参数为空返回 nil
	if len(providers) == 0 || task == nil {
		return nil, errors.New("算法获取参数为空")
	}

	//调度算法
	strategy := algo.NewK8sStrategy(task, providers...)
	taskResult, err := algo.ScheduleWithFullCollaboration(strategy, strategy.ProviderList)
	if err != nil {
		return nil, err
	}
	return taskResult.MaxscoreStrategy, nil
}

func (cs *cloudScheduler) getNewStructForDb(task *response.TaskInfo, participantId int64) (interface{}, error) {
	bytes, err := json.Marshal(task.Metadata)
	if err != nil {
		return nil, err
	}
	cloud := cs.UnMarshalK8sStruct(string(bytes), task.TaskId)
	cloud.Id = utils.GenSnowflakeID()
	cloud.YamlString = string(bytes)
	cloud.ParticipantId = participantId
	return cloud, nil
}

func (cs *cloudScheduler) UnMarshalK8sStruct(yamlString string, taskId int64) models.Cloud {
	var cloud models.Cloud
	d := kyaml.NewYAMLOrJSONDecoder(bytes.NewBufferString(yamlString), 4096)
	var err error
	for {
		var rawObj runtime.RawExtension
		err = d.Decode(&rawObj)
		if err == io.EOF {
			break
		}
		if err != nil {
		}
		obj := &unstructured.Unstructured{}
		syaml.NewDecodingSerializer(unstructured.UnstructuredJSONScheme).Decode(rawObj.Raw, nil, obj)
		if err != nil {
		}

		unstructuredMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
		if err != nil {
		}

		unstructureObj := &unstructured.Unstructured{Object: unstructuredMap}
		cloud = models.Cloud{
			TaskId:     taskId,
			ApiVersion: unstructureObj.GetAPIVersion(),
			Name:       unstructureObj.GetName(),
			Kind:       unstructureObj.GetKind(),
			Namespace:  unstructureObj.GetNamespace(),
			Status:     "Saved",
		}
		// 命名空间为空 设置默认值
		if len(unstructureObj.GetNamespace()) == 0 {
			cloud.Namespace = "default"
		}
	}
	return cloud
}

func (cs *cloudScheduler) genTaskAndProviders(task *response.TaskInfo, dbEngin *gorm.DB) (*algo.Task, []*algo.Provider) {
	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"
	dbEngin.Raw(sqlstr).Scan(&proParams)

	var providerList []*algo.Provider
	for _, p := range proParams {
		provider := algo.NewProvider(p.Participant_id, p.Cpu_avail, p.Mem_avail, p.Disk_avail, 0.0, 0.0, 0.0)
		providerList = append(providerList, provider)
	}

	t := algo.NewTask(0, 1, 2, 75120000, 301214500, 1200, 2, 6, 2000)

	return t, providerList
}