| @@ -144,6 +144,7 @@ func (l *CommitHpcTaskLogic) CommitHpcTask(req *types.CommitHpcTaskReq) (resp *t | |||||
| logx.Info("提交job到指定集群") | logx.Info("提交job到指定集群") | ||||
| resp, err = l.hpcService.HpcExecutorAdapterMap[adapterInfo.Id].SubmitTask(l.ctx, *req) | resp, err = l.hpcService.HpcExecutorAdapterMap[adapterInfo.Id].SubmitTask(l.ctx, *req) | ||||
| if err != nil { | if err != nil { | ||||
| logx.Errorf("提交Hpc到指定集群失败, err: %v", err) | |||||
| return nil, err | return nil, err | ||||
| } | } | ||||
| // 更新任务状态 | // 更新任务状态 | ||||
| @@ -44,6 +44,9 @@ func (l *GetHpcTaskLogLogic) GetHpcTaskLog(req *types.HpcTaskLogReq) (resp inter | |||||
| if tx != nil { | if tx != nil { | ||||
| return nil, fmt.Errorf("数据库查询失败: %v", tx.Error) | return nil, fmt.Errorf("数据库查询失败: %v", tx.Error) | ||||
| } | } | ||||
| if hpcR.ID == 0 { | |||||
| return nil, fmt.Errorf("任务不存在") | |||||
| } | |||||
| var adapterInfo types.AdapterInfo | var adapterInfo types.AdapterInfo | ||||
| l.svcCtx.DbEngin.Raw("SELECT * FROM `t_adapter` where id = ?", hpcR.AdapterId).Scan(&adapterInfo) | l.svcCtx.DbEngin.Raw("SELECT * FROM `t_adapter` where id = ?", hpcR.AdapterId).Scan(&adapterInfo) | ||||
| if adapterInfo.Id == "" { | if adapterInfo.Id == "" { | ||||
| @@ -2,12 +2,15 @@ package hpcservice | |||||
| import ( | import ( | ||||
| "context" | "context" | ||||
| "fmt" | |||||
| "github.com/go-resty/resty/v2" | "github.com/go-resty/resty/v2" | ||||
| "github.com/zeromicro/go-zero/core/logx" | "github.com/zeromicro/go-zero/core/logx" | ||||
| "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/collector" | "gitlink.org.cn/JointCloud/pcm-coordinator/internal/scheduler/service/collector" | ||||
| "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types" | "gitlink.org.cn/JointCloud/pcm-coordinator/internal/types" | ||||
| "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants" | "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/constants" | ||||
| "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result" | "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/repository/result" | ||||
| "gitlink.org.cn/JointCloud/pcm-coordinator/pkg/utils/restyclient" | |||||
| "net/http" | |||||
| ) | ) | ||||
| type ParticipantHpc struct { | type ParticipantHpc struct { | ||||
| @@ -16,6 +19,7 @@ type ParticipantHpc struct { | |||||
| host string | host string | ||||
| userName string | userName string | ||||
| accessToken string | accessToken string | ||||
| *restyclient.RestyClient | |||||
| } | } | ||||
| const ( | const ( | ||||
| @@ -31,6 +35,7 @@ func NewHpc(host string, id int64, platform string) *ParticipantHpc { | |||||
| host: host, | host: host, | ||||
| participantId: id, | participantId: id, | ||||
| platform: platform, | platform: platform, | ||||
| RestyClient: restyclient.InitClient(host, ""), | |||||
| } | } | ||||
| } | } | ||||
| @@ -93,7 +98,9 @@ func (c *ParticipantHpc) SubmitTask(ctx context.Context, req types.CommitHpcTask | |||||
| if err != nil { | if err != nil { | ||||
| return nil, err | return nil, err | ||||
| } | } | ||||
| if resp.Code != http.StatusOK { | |||||
| return nil, fmt.Errorf(resp.Msg) | |||||
| } | |||||
| return &resp, nil | return &resp, nil | ||||
| } | } | ||||
| @@ -113,24 +120,32 @@ func (c *ParticipantHpc) CancelTask(ctx context.Context, jobId string) error { | |||||
| if err != nil { | if err != nil { | ||||
| return err | return err | ||||
| } | } | ||||
| if resp.Code != http.StatusOK { | |||||
| return fmt.Errorf(resp.Msg) | |||||
| } | |||||
| return nil | return nil | ||||
| } | } | ||||
| func (c *ParticipantHpc) GetTaskLogs(ctx context.Context, jobId string) (interface{}, error) { | func (c *ParticipantHpc) GetTaskLogs(ctx context.Context, jobId string) (interface{}, error) { | ||||
| reqUrl := c.host + JobLogUrl | |||||
| logx.WithContext(ctx).Infof("获取超算集群任务日志, url: %s, jobId: %s", JobLogUrl, jobId) | |||||
| if jobId == "" { | |||||
| return nil, fmt.Errorf("jobId is empty") | |||||
| } | |||||
| resp := types.CommonResp{} | resp := types.CommonResp{} | ||||
| logx.WithContext(ctx).Infof("获取超算集群任务日志, url: %s, jobId: %s", reqUrl, jobId) | |||||
| httpClient := resty.New().R() | |||||
| _, err := httpClient.SetHeaders( | |||||
| map[string]string{ | |||||
| _, err := c.Request(JobLogUrl, http.MethodGet, func(req *resty.Request) { | |||||
| req.SetHeaders(map[string]string{ | |||||
| "Content-Type": "application/json", | "Content-Type": "application/json", | ||||
| "traceId": result.TraceIDFromContext(ctx), | "traceId": result.TraceIDFromContext(ctx), | ||||
| }).SetPathParams(map[string]string{ | }).SetPathParams(map[string]string{ | ||||
| "backend": BackendSlurm, | |||||
| "jobId": jobId, | |||||
| }).SetResult(&resp).Get(reqUrl) | |||||
| "backend": BackendSlurm, | |||||
| "jobId": jobId, | |||||
| }).SetResult(&resp) | |||||
| }) | |||||
| if err != nil { | if err != nil { | ||||
| return nil, err | return nil, err | ||||
| } | } | ||||
| if resp.Code != http.StatusOK { | |||||
| return nil, fmt.Errorf(resp.Msg) | |||||
| } | |||||
| return resp, nil | return resp, nil | ||||
| } | } | ||||
| @@ -0,0 +1,105 @@ | |||||
| package restyclient | |||||
| import ( | |||||
| "crypto/tls" | |||||
| "errors" | |||||
| "fmt" | |||||
| "net/http" | |||||
| "time" | |||||
| "github.com/go-resty/resty/v2" | |||||
| ) | |||||
| type RestyClient struct { | |||||
| 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(baseURL, apiKey string) *RestyClient { | |||||
| 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(baseURL, apiKey) | |||||
| HttpClient := NewHttpClient() | |||||
| return &RestyClient{ | |||||
| NoRedirectClient: NoRedirectClient, | |||||
| RestyClient: restyClient, | |||||
| HttpClient: HttpClient, | |||||
| } | |||||
| } | |||||
| func NewRestyClient(baseUrl, apiKey string) *resty.Client { | |||||
| client := resty.New(). | |||||
| SetHeader("user-agent", UserAgent). | |||||
| SetRetryCount(3). | |||||
| SetBaseURL(baseUrl). | |||||
| 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}, | |||||
| }, | |||||
| } | |||||
| } | |||||
| // Request 封装了通用的 HTTP 请求方法 | |||||
| func (c *RestyClient) Request(url string, method string, callback ReqCallback) ([]byte, error) { | |||||
| // 参数校验 | |||||
| if url == "" { | |||||
| return nil, errors.New("URL cannot be empty") | |||||
| } | |||||
| respErr := &RespErr{} | |||||
| req := c.RestyClient.R().SetHeaders(map[string]string{ | |||||
| "Content-Type": "application/json", // 回调仍可覆盖此Header | |||||
| }).SetError(respErr) | |||||
| switch method { | |||||
| case "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS": | |||||
| req.Method = method | |||||
| } | |||||
| req.URL = url | |||||
| if callback != nil { | |||||
| callback(req) | |||||
| } | |||||
| res, err := req.Send() | |||||
| if err != nil { | |||||
| return nil, fmt.Errorf("request to %s %s failed: %w", req.Method, req.URL, err) | |||||
| } | |||||
| // 处理非2xx响应 | |||||
| if !res.IsSuccess() { | |||||
| if respErr.Message != "" { | |||||
| return nil, fmt.Errorf("server error: %s (status %d)", respErr.Message, res.StatusCode()) | |||||
| } | |||||
| return nil, fmt.Errorf("unexpected status %d: %s", res.StatusCode(), res.String()) | |||||
| } | |||||
| return res.Body(), nil | |||||
| } | |||||
| // Post 封装 POST 请求 | |||||
| func (c *RestyClient) Post(path string, body interface{}, result interface{}) error { | |||||
| callback := func(req *resty.Request) { | |||||
| req.SetBody(body).SetResult(result) | |||||
| } | |||||
| _, err := c.Request(path, http.MethodPost, callback) | |||||
| return err | |||||
| } | |||||
| @@ -0,0 +1,34 @@ | |||||
| package restyclient | |||||
| import "github.com/go-resty/resty/v2" | |||||
| type Json map[string]interface{} | |||||
| type TokenResp struct { | |||||
| AccessToken string `json:"access_token"` | |||||
| RefreshToken string `json:"refresh_token"` | |||||
| } | |||||
| type ReqCallback func(*resty.Request) | |||||
| type Error struct { | |||||
| // The general error message | |||||
| // | |||||
| // required: true | |||||
| // example: Unauthorized | |||||
| Error string `json:"error"` | |||||
| // The http error code. | |||||
| // | |||||
| // required: true | |||||
| // example: 401 | |||||
| ErrorCode int `json:"errorCode"` | |||||
| // The http error code. | |||||
| // | |||||
| // required: true | |||||
| // example: you need to provide a valid access token or user credentials to access this api | |||||
| ErrorDescription string `json:"errorDescription"` | |||||
| } | |||||
| type RespErr struct { | |||||
| Code interface{} `json:"code"` | |||||
| Message string `json:"message"` | |||||
| } | |||||