pcm-coordinator/internal/participant/cloud/cloud.go

package cloud

import (
	"fmt"
	"github.com/go-resty/resty/v2"
	"github.com/pkg/errors"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/database"
	"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types/cloud"
	"k8s.io/apimachinery/pkg/util/json"
	"net/http"
	"sync"
)

const CreateContainer = "/cloud/container/create"
const DeleteContainer = "/cloud/container/delete"
const GetContainer = "/cloud/container/get"
const LogContainer = "/cloud/container/log"
const MetricsContainer = "/cloud/container/metrics"
const ModelBind = "/cloud/model/bind"

type Cloud struct {
	store  *database.CloudStorage
	idAddr sync.Map
}

func New(store *database.CloudStorage, adapterId string) (*Cloud, error) {
	if store == nil {
		return nil, errors.New("store cannot be nil")
	}

	a := &Cloud{
		store: store,
	}

	css, err := store.GetClustersByAdapterId(adapterId)
	if err != nil {
		return nil, fmt.Errorf("failed to get clusters: %w", err)
	}

	for _, info := range css.List {
		a.idAddr.Store(info.Id, info.Server)
	}

	InitClient()
	return a, nil
}

func (c *Cloud) ContainerCreate(platformId string, param *CreateParam) (resp *Resp, err error) {
	addr, ok := c.GetServerAddrById(platformId)
	if !ok {
		return nil, fmt.Errorf("clusterId not found: %s", platformId)
	}
	respErr := &RespErr{}
	_, err = Request(addr+CreateContainer, http.MethodPost, func(req *resty.Request) {
		req.SetQueryParams(map[string]string{
			"pfId": platformId,
		}).SetBody(param).SetError(&respErr).SetResult(&resp)
	})
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func (c *Cloud) ContainerDelete(platformId string, param *cloud.DeleteParam) (resp *Resp, err error) {
	marshal, err := json.Marshal(param)
	if err != nil {
		return nil, err
	}
	println(string(marshal))
	addr, ok := c.GetServerAddrById(platformId)
	if !ok {
		return nil, fmt.Errorf("clusterId not found: %s", platformId)
	}
	respErr := &RespErr{}
	_, err = Request(addr+DeleteContainer, http.MethodPost, func(req *resty.Request) {
		req.SetQueryParams(map[string]string{
			"pfId": platformId,
		}).SetBody(param).SetError(&respErr).SetResult(&resp)
	})
	if err != nil {
		return nil, err
	}
	return
}

func (c *Cloud) ContainerGet(platformId string, param *GetParam) (resp *Resp, err error) {
	addr, ok := c.GetServerAddrById(platformId)
	if !ok {
		return nil, fmt.Errorf("clusterId not found: %s", platformId)
	}
	respErr := &RespErr{}
	_, err = Request(addr+GetContainer, http.MethodPost, func(req *resty.Request) {
		req.SetQueryParams(map[string]string{
			"pfId": platformId,
		}).SetBody(param).SetError(&respErr).SetResult(&resp)
	})
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func (c *Cloud) BindModel(platformId string, filePath string, fileName string) (resp *Resp, err error) {
	addr, ok := c.GetServerAddrById(platformId)
	if !ok {
		return nil, fmt.Errorf("clusterId not found: %s", platformId)
	}
	respErr := &RespErr{}
	_, err = Request(addr+ModelBind, http.MethodPut, func(req *resty.Request) {
		req.SetQueryParams(map[string]string{
			"filePath": filePath,
			"fileName": fileName,
		}).SetError(&respErr).SetResult(&resp)
	})
	if err != nil {
		return nil, err
	}
	marshal, err := json.Marshal(resp.Data)
	if err != nil {
		return nil, err
	}
	fmt.Printf(string(marshal))
	return resp, nil
}

func (c *Cloud) ContainerLog(platformId string, param *cloud.LogParam) (resp *Resp, err error) {
	addr, ok := c.GetServerAddrById(platformId)
	if !ok {
		return nil, fmt.Errorf("clusterId not found: %s", platformId)
	}
	bytes, err := json.Marshal(param)
	if err != nil {
		return nil, err
	}
	fmt.Printf(string(bytes))
	respErr := &RespErr{}
	asd, err := Request(addr+LogContainer, http.MethodPost, func(req *resty.Request) {
		req.SetQueryParams(map[string]string{
			"pfId": platformId,
		}).SetBody(param).SetError(&respErr).SetResult(&resp)
	})
	if err != nil {
		return nil, err
	}
	json.Marshal(asd)
	marshal, err := json.Marshal(resp.Data)
	if err != nil {
		return nil, err
	}
	fmt.Printf(string(marshal))
	return resp, nil
}

func (c *Cloud) GetServerAddrById(id string) (string, bool) {
	val, ok := c.idAddr.Load(id)
	if !ok {
		return "", false
	}
	addr, ok := val.(string)
	if !ok {
		return "", false
	}
	return addr, true
}

func (c *Cloud) ContainerMetrics(platformId string, param *cloud.MetricsParam) (resp *Resp, err error) {
	addr, ok := c.GetServerAddrById(platformId)
	if !ok {
		return nil, fmt.Errorf("clusterId not found: %s", platformId)
	}
	bytes, err := json.Marshal(param)
	if err != nil {
		return nil, err
	}
	fmt.Printf(string(bytes))
	respErr := &RespErr{}
	asd, err := Request(addr+MetricsContainer, http.MethodPost, func(req *resty.Request) {
		req.SetQueryParams(map[string]string{
			"pfId": platformId,
		}).SetBody(param).SetError(&respErr).SetResult(&resp)
	})
	if err != nil {
		return nil, err
	}
	json.Marshal(asd)
	marshal, err := json.Marshal(resp.Data)
	if err != nil {
		return nil, err
	}
	fmt.Printf(string(marshal))
	return resp, nil
}