zhangweiii
/
pcm-coordinator
forked from JointCloud/pcm-coordinator
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

容器接口

This commit is contained in:
zhangwei 2025-07-24 15:18:20 +08:00
parent 1ec0a6709a
commit 91247cf0c6
15 changed files with 618 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
etc/pcm.yaml
View File

@ -94,3 +94,4 @@ JcsMiddleware:
Participant:
AdapterId: "1777144940456666666"
CloudAdapterId: "1770658294298316800"

1
internal/config/config.go
View File

@ -72,4 +72,5 @@ type JcsMiddleware struct {
type Participant struct {
AdapterId string
CloudAdapterId string
}

32
internal/handler/cloud/containercreatehandler.go Normal file
View File

@ -0,0 +1,32 @@
package cloud
import (
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/logic/cloud"
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result"
"io"
"k8s.io/apimachinery/pkg/util/json"
"net/http"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
container "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types/cloud"
)
func ContainerCreateHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req container.CreateParam
body, err := io.ReadAll(r.Body)
if err != nil {
result.ParamErrorResult(r, w, err)
return
}
if err = json.Unmarshal(body, &req); err != nil {
result.ParamErrorResult(r, w, err)
return
}
l := cloud.NewContainerCreateLogic(r.Context(), svcCtx)
resp, err := l.ContainerCreate(&req)
result.HttpResult(r, w, resp, err)
}
}

32
internal/handler/cloud/containerdeletehandler.go Normal file
View File

@ -0,0 +1,32 @@
package cloud
import (
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/logic/cloud"
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result"
"io"
"k8s.io/apimachinery/pkg/util/json"
"net/http"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
container "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types/cloud"
)
func ContainerDeleteHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req container.DeleteParam
body, err := io.ReadAll(r.Body)
if err != nil {
result.ParamErrorResult(r, w, err)
return
}
if err = json.Unmarshal(body, &req); err != nil {
result.ParamErrorResult(r, w, err)
return
}
l := cloud.NewContainerDeleteLogic(r.Context(), svcCtx)
resp, err := l.ContainerDelete(&req)
result.HttpResult(r, w, resp, err)
}
}

32
internal/handler/cloud/containergethandler.go Normal file
View File

@ -0,0 +1,32 @@
package cloud
import (
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/logic/cloud"
"gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result"
"io"
"k8s.io/apimachinery/pkg/util/json"
"net/http"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
container "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types/cloud"
)
func ContainerGetHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req container.GetParam
body, err := io.ReadAll(r.Body)
if err != nil {
result.ParamErrorResult(r, w, err)
return
}
if err = json.Unmarshal(body, &req); err != nil {
result.ParamErrorResult(r, w, err)
return
}
l := cloud.NewContainerGetLogic(r.Context(), svcCtx)
resp, err := l.ContainerGet(&req)
result.HttpResult(r, w, resp, err)
}
}

18
internal/handler/routes.go
View File

@ -376,6 +376,24 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
Path: "/task/list",
Handler: cloud.CloudListHandler(serverCtx),
},
{
// 创建容器
Method: http.MethodPost,
Path: "/cloud/container/create",
Handler: cloud.ContainerCreateHandler(serverCtx),
},
{
// 删除容器
Method: http.MethodDelete,
Path: "/cloud/container/delete",
Handler: cloud.ContainerDeleteHandler(serverCtx),
},
{
// 获取容器
Method: http.MethodGet,
Path: "/cloud/container/get",
Handler: cloud.ContainerGetHandler(serverCtx),
},
},
rest.WithPrefix("/pcm/v1"),
)

64
internal/logic/cloud/containercreatelogic.go Normal file
View File

@ -0,0 +1,64 @@
/*
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 cloud
import (
"context"
"errors"
"github.com/zeromicro/go-zero/core/logx"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/participant/cloud"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
container "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types/cloud"
"net/http"
)
type ContainerCreateLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewContainerCreateLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ContainerCreateLogic {
return &ContainerCreateLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *ContainerCreateLogic) ContainerCreate(req *container.CreateParam) (resp interface{}, err error) {
param := &cloud.CreateParam{
Name: req.Name,
Port: req.Port,
Cpu: req.Cpu,
Memory: req.Memory,
Image: req.Image,
Args: req.Args,
MountPath: req.MountPath,
Envs: req.Envs,
NodePort: req.NodePort,
ContainerGroupName: req.ContainerGroupName,
CreateParameter: req.ContainerCreateParameter,
}
create, err := l.svcCtx.Cloud.ContainerCreate(req.ClusterId, param)
if err != nil {
return nil, err
}
if create.Code != http.StatusOK {
return nil, errors.New(create.Message)
}
resp = create.Data
return
}

54
internal/logic/cloud/containerdeletelogic.go Normal file
View File

@ -0,0 +1,54 @@
/*
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 cloud
import (
"context"
"errors"
"github.com/zeromicro/go-zero/core/logx"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/participant/cloud"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
container "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types/cloud"
"net/http"
)
type ContainerDeleteLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewContainerDeleteLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ContainerDeleteLogic {
return &ContainerDeleteLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *ContainerDeleteLogic) ContainerDelete(req *container.DeleteParam) (resp interface{}, err error) {
param := &cloud.DeleteParam{
Name: req.Name,
}
create, err := l.svcCtx.Cloud.ContainerDelete(req.ClusterId, param)
if err != nil {
return nil, err
}
if create.Code != http.StatusOK {
return nil, errors.New(create.Message)
}
resp = create.Data
return
}

54
internal/logic/cloud/containergetlogic.go Normal file
View File

@ -0,0 +1,54 @@
/*
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 cloud
import (
"context"
"errors"
"github.com/zeromicro/go-zero/core/logx"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/participant/cloud"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/svc"
container "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types/cloud"
"net/http"
)
type ContainerGetLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewContainerGetLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ContainerGetLogic {
return &ContainerGetLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *ContainerGetLogic) ContainerGet(req *container.GetParam) (resp interface{}, err error) {
param := &cloud.GetParam{
Name: req.Name,
}
get, err := l.svcCtx.Cloud.ContainerGet(req.ClusterId, param)
if err != nil {
return nil, err
}
if get.Code != http.StatusOK {
return nil, errors.New(get.Message)
}
resp = get.Data
return
}

81
internal/participant/cloud/client.go Normal file
View File

@ -0,0 +1,81 @@
package cloud
import (
"crypto/tls"
"errors"
"fmt"
"net/http"
"time"
"github.com/go-resty/resty/v2"
)
type ReqCallback func(req *resty.Request)
var (
NoRedirectClient *resty.Client
RestyClient *resty.Client
HttpClient *http.Client
)
var UserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
var DefaultTimeout = time.Second * 300
func InitClient() {
NoRedirectClient = resty.New().SetRedirectPolicy(
resty.RedirectPolicyFunc(func(req *http.Request, via []*http.Request) error {
return http.ErrUseLastResponse
}),
).SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true})
NoRedirectClient.SetHeader("user-agent", UserAgent)
RestyClient = NewRestyClient()
HttpClient = NewHttpClient()
}
func NewRestyClient() *resty.Client {
client := resty.New().
SetHeader("user-agent", UserAgent).
SetRetryCount(3).
SetRetryResetReaders(true).
SetTimeout(DefaultTimeout).
SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true})
return client
}
func NewHttpClient() *http.Client {
return &http.Client{
Timeout: time.Hour * 48,
Transport: &http.Transport{
Proxy: http.ProxyFromEnvironment,
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
},
}
}
func Request(url string, method string, callback ReqCallback) ([]byte, error) {
respErr := &RespErr{}
req := RestyClient.R().
SetHeaders(map[string]string{
"Content-Type": "application/json",
}).
SetError(&respErr)
if callback != nil {
callback(req)
}
res, err := req.Execute(method, url)
if err != nil {
return nil, err
}
if respErr.Message != "" {
return nil, errors.New(respErr.Message)
}
if res.StatusCode() != http.StatusOK && res.StatusCode() != http.StatusCreated {
return nil, errors.New(fmt.Sprintf("msg: %s, status: %d", res.String(), res.StatusCode()))
}
return res.Body(), nil
}

105
internal/participant/cloud/cloud.go Normal file
View File

@ -0,0 +1,105 @@
package cloud
import (
"fmt"
"github.com/go-resty/resty/v2"
"github.com/pkg/errors"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/database"
"net/http"
"sync"
)
const CreateContainer = "/cloud/container/create"
const DeleteContainer = "/cloud/container/delete"
const GetContainer = "/cloud/container/get"
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 *DeleteParam) (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+DeleteContainer, http.MethodDelete, 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.MethodGet, func(req *resty.Request) {
req.SetQueryParams(map[string]string{
"pfId": platformId,
"name": param.Name,
}).SetBody(param).SetError(&respErr).SetResult(&resp)
})
if err != nil {
return nil, err
}
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
}

38
internal/participant/cloud/model.go Normal file
View File

@ -0,0 +1,38 @@
package cloud
type RespErr struct {
Code int32 `json:"code"`
Message string `json:"message"`
}
type Resp struct {
Code int32 `json:"code"`
Message string `json:"msg"`
Data interface{} `json:"data"`
}
type CreateParam struct {
ContainerGroupName string `json:"containerGroupName"`
Name string `json:"name"`
Image string `json:"image"`
Cpu string `json:"cpu,omitempty"`
Memory string `json:"memory,omitempty"`
Port int32 `json:"port,omitempty"`
NodePort int32 `json:"nodePort,omitempty"`
MountPath string `json:"mountPath,omitempty"`
Args []string `json:"args,omitempty"`
Envs []struct {
Name string `json:"name,omitempty"`
Value string `json:"value,omitempty"`
} `json:"envs,omitempty"`
CreateParameter interface{} `json:"createParameter,omitempty"`
}
type DeleteParam struct {
Name string `json:"name,omitempty"`
DeleteParameter interface{} `json:"deleteParameter,omitempty"`
}
type GetParam struct {
Name string `json:"name,omitempty"`
GetParameter interface{} `json:"getParameter,omitempty"`
}

19
internal/scheduler/database/cloudStorage.go
View File

@ -1,22 +1,24 @@
package database
import (
"github.com/zeromicro/go-zero/core/logx"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/entity"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/types"
"gorm.io/gorm"
)
type CloudStorage struct {
dbEngin *gorm.DB
DbEngin *gorm.DB
}
func NewCloudStorage(dbEngin *gorm.DB) *CloudStorage {
return &CloudStorage{dbEngin: dbEngin}
return &CloudStorage{DbEngin: dbEngin}
}
func (c *CloudStorage) GetProviderParams() ([]entity.ProviderParams, error) {
var proParams []entity.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)
c.DbEngin.Raw(sqlstr).Scan(&proParams)
if len(proParams) == 0 {
return nil, nil
}
@ -26,9 +28,18 @@ func (c *CloudStorage) GetProviderParams() ([]entity.ProviderParams, error) {
func (c *CloudStorage) FindAvailableParticipants() ([]entity.Participant, error) {
var parts []entity.Participant
sqlstr := "select id as participant_id, name as name from sc_participant_phy_info"
c.dbEngin.Raw(sqlstr).Scan(&parts)
c.DbEngin.Raw(sqlstr).Scan(&parts)
if len(parts) == 0 {
return nil, nil
}
return parts, nil
}
func (c *CloudStorage) GetClustersByAdapterId(id string) (*types.ClusterListResp, error) {
var resp types.ClusterListResp
tx := c.DbEngin.Raw("select * from t_cluster where `deleted_at` IS NULL and `adapter_id` = ? ORDER BY create_time Desc", id).Scan(&resp.List)
if tx.Error != nil {
logx.Errorf(tx.Error.Error())
return nil, tx.Error
}
return &resp, nil
}

9
internal/svc/servicecontext.go
View File

@ -24,6 +24,7 @@ import (
"gitlink.org.cn/JointCloud/pcm-ac/hpcacclient"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/config"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/participant"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/participant/cloud"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/database"
"gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service"
@ -58,6 +59,7 @@ type ServiceContext struct {
HttpClient *resty.Client
Scheduler *scheduler.Scheduler
Ai *participant.Ai
Cloud *cloud.Cloud
}
func NewServiceContext(c config.Config) *ServiceContext {
@ -111,6 +113,7 @@ func NewServiceContext(c config.Config) *ServiceContext {
// scheduler
storage := &database.AiStorage{DbEngin: dbEngin}
hpcStorage := &database.HpcStorage{DbEngin: dbEngin}
cloudStorage := &database.CloudStorage{DbEngin: dbEngin}
cache := make(map[string]interface{}, 0)
aiService, err := service.NewAiService(&c, storage, cache)
if err != nil {
@ -128,6 +131,11 @@ func NewServiceContext(c config.Config) *ServiceContext {
logx.Error(err.Error())
panic(err)
}
cloud, err := cloud.New(cloudStorage, c.Participant.CloudAdapterId)
if err != nil {
logx.Error(err.Error())
panic(err)
}
return &ServiceContext{
DbEngin: dbEngin,
Cron: cron.New(cron.WithSeconds()),
@ -144,5 +152,6 @@ func NewServiceContext(c config.Config) *ServiceContext {
HttpClient: httpClient,
Scheduler: scheduler,
Ai: ai,
Cloud: cloud,
}
}

80
internal/types/cloud/container.go Normal file
View File

@ -0,0 +1,80 @@
package cloud
type ContainerCreateParameter interface {
ContainerCreateParam()
}
type ContainerGetParameter interface {
ContainerGetParam()
}
type ContainerDeleteParameter interface {
ContainerDeleteParam()
}
type CreateParam struct {
ClusterId string `json:"clusterId,omitempty"`
ContainerGroupName string `json:"containerGroupName"`
Name string `json:"name"`
Image string `json:"image"`
Cpu string `json:"cpu,omitempty"`
Memory string `json:"memory,omitempty"`
Port int32 `json:"port,omitempty"`
NodePort int32 `json:"nodePort,omitempty"`
MountPath string `json:"mountPath,omitempty"`
Args []string `json:"args,omitempty"`
Envs []struct {
Name string `json:"name,omitempty"`
Value string `json:"value,omitempty"`
} `json:"envs,omitempty"`
ContainerCreateParameter ContainerCreateParameter `json:"containerCreateParameter,omitempty"`
}
type K8sCreateParam struct {
}
type EciCreateParam struct {
}
func (k K8sCreateParam) ContainerCreateParam() {
}
func (e EciCreateParam) ContainerCreateParam() {
}
// 删除容器参数
type DeleteParam struct {
ClusterId string `json:"clusterId,omitempty"`
Name string `json:"name,omitempty"`
ContainerDeleteParameter ContainerDeleteParameter `json:"containerDeleteParameter,omitempty"`
}
func (k K8sDeleteParam) ContainerDeleteParameter() {
}
func (e EciDeleteParam) ContainerDeleteParameter() {
}
type K8sDeleteParam struct {
}
type EciDeleteParam struct {
RegionId string `json:"regionId,omitempty"`
ContainerGroupId string `json:"containerGroupId,omitempty"`
}
// 获取容器信息
type GetParam struct {
ClusterId string `json:"clusterId,omitempty"`
Name string `json:"name,omitempty"`
GetParameter ContainerGetParameter `json:"getParameter,omitempty"`
}
func (g K8sGetParam) ContainerGetParam() {
}
func (g EciGetParam) ContainerGetParam() {
}
type K8sGetParam struct {
}
type EciGetParam struct {
RegionId string `json:"regionId,omitempty"`
ContainerGroupName string `json:"containerGroupName,omitempty"`
}