forked from JointCloud/pcm-coordinator
119 lines
3.6 KiB
Go
119 lines
3.6 KiB
Go
package cloud
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"github.com/pkg/errors"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/models/cloud"
|
|
"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"
|
|
"sigs.k8s.io/yaml"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/svc"
|
|
"gitlink.org.cn/JointCloud/pcm-coordinator/api/internal/types"
|
|
|
|
"github.com/zeromicro/go-zero/core/logx"
|
|
)
|
|
|
|
type CommitGeneralTaskLogic struct {
|
|
logx.Logger
|
|
ctx context.Context
|
|
svcCtx *svc.ServiceContext
|
|
}
|
|
|
|
func NewCommitGeneralTaskLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CommitGeneralTaskLogic {
|
|
return &CommitGeneralTaskLogic{
|
|
Logger: logx.WithContext(ctx),
|
|
ctx: ctx,
|
|
svcCtx: svcCtx,
|
|
}
|
|
}
|
|
|
|
func (l *CommitGeneralTaskLogic) CommitGeneralTask(req *types.GeneralTaskReq) error {
|
|
var yamlStr []string
|
|
for _, s := range req.ReqBody {
|
|
j2, err := yaml.YAMLToJSON([]byte(s))
|
|
if err != nil {
|
|
logx.Errorf("Failed to convert yaml to JSON, err: %v", err)
|
|
return err
|
|
}
|
|
yamlStr = append(yamlStr, string(j2))
|
|
}
|
|
result := strings.Join(yamlStr, ",")
|
|
//TODO The namespace is fixed to ns-admin for the time being. Later, the namespace is obtained based on the user
|
|
taskModel := models.Task{
|
|
Status: constants.Saved,
|
|
Name: req.Name,
|
|
CommitTime: time.Now(),
|
|
NsID: "ns-admin",
|
|
YamlString: "[" + result + "]",
|
|
}
|
|
// Save the task data to the database
|
|
tx := l.svcCtx.DbEngin.Create(&taskModel)
|
|
if tx.Error != nil {
|
|
return tx.Error
|
|
}
|
|
|
|
var clusters []*models.CloudModel
|
|
err := l.svcCtx.DbEngin.Raw("SELECT * FROM `t_cluster` where adapter_id = ? and id in ?", req.AdapterId, req.ClusterIds).Scan(&clusters).Error
|
|
if err != nil {
|
|
logx.Errorf("CommitGeneralTask() => sql execution error: %v", err)
|
|
return errors.Errorf("the cluster does not match the drive resources. Check the data")
|
|
}
|
|
taskCloud := cloud.TaskCloudModel{}
|
|
//TODO 执行策略返回集群跟 Replica
|
|
for _, c := range clusters {
|
|
for _, s := range req.ReqBody {
|
|
sStruct := UnMarshalK8sStruct(s)
|
|
unString, _ := sStruct.MarshalJSON()
|
|
taskCloud.TaskId = uint(taskModel.Id)
|
|
taskCloud.AdapterId = c.AdapterId
|
|
taskCloud.ClusterId = c.Id
|
|
taskCloud.ClusterName = c.Name
|
|
taskCloud.Status = "Saved"
|
|
taskCloud.YamlString = string(unString)
|
|
taskCloud.Kind = sStruct.GetKind()
|
|
taskCloud.Namespace = sStruct.GetNamespace()
|
|
tx = l.svcCtx.DbEngin.Create(&taskCloud)
|
|
if tx.Error != nil {
|
|
logx.Errorf("CommitGeneralTask() create taskCloud => sql execution error: %v", err)
|
|
return tx.Error
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func UnMarshalK8sStruct(yamlString string) *unstructured.Unstructured {
|
|
unstructuredObj := &unstructured.Unstructured{}
|
|
d := kyaml.NewYAMLOrJSONDecoder(bytes.NewBufferString(yamlString), 4096)
|
|
var err error
|
|
for {
|
|
var rawObj runtime.RawExtension
|
|
err = d.Decode(&rawObj)
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
obj := &unstructured.Unstructured{}
|
|
syaml.NewDecodingSerializer(unstructured.UnstructuredJSONScheme).Decode(rawObj.Raw, nil, obj)
|
|
unstructuredMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
|
|
if err != nil {
|
|
logx.Errorf("UnMarshalK8sStruct() => Execution failure err:%v", err)
|
|
}
|
|
unstructuredObj = &unstructured.Unstructured{Object: unstructuredMap}
|
|
// 命名空间为空 设置默认值
|
|
if len(unstructuredObj.GetNamespace()) == 0 {
|
|
unstructuredObj.SetNamespace("default")
|
|
}
|
|
}
|
|
return unstructuredObj
|
|
}
|