forked from ci4s/pipeline-convert
285 lines
7.2 KiB
Go
285 lines
7.2 KiB
Go
package utils
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
corev1 "k8s.io/api/core/v1"
|
||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||
"k8s.io/client-go/kubernetes"
|
||
"k8s.io/client-go/rest"
|
||
"mime/multipart"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
)
|
||
|
||
const (
|
||
K8SHost = "http://ci4s-gateway-service.argo:8082"
|
||
// K8SHost = "http://172.20.32.98:8082"
|
||
mountPrefix = "argo-workflow"
|
||
CodeOK = 200
|
||
Gataway = "mmp"
|
||
Admin = "fanshuai"
|
||
Password = "h1n2x3j4y5@"
|
||
GetTokenUri = "auth/loginByKey"
|
||
AutoExportType = "auto_export"
|
||
)
|
||
|
||
type TokenRequest struct {
|
||
Name string `json:"username"`
|
||
Password string `json:"key"`
|
||
}
|
||
|
||
type TokenResponse struct {
|
||
Code int `json:"code"`
|
||
Msg interface{} `json:"msg"`
|
||
Data struct {
|
||
AccessToken string `json:"access_token"`
|
||
ExpiresIn int `json:"expires_in"`
|
||
} `json:"data"`
|
||
}
|
||
|
||
type UploadRequest struct {
|
||
Path string `json:"path"`
|
||
UUID string `json:"uuid"`
|
||
}
|
||
|
||
type UploadResponse struct {
|
||
Msg string `json:"msg"`
|
||
Code int `json:"code"`
|
||
Data []struct {
|
||
FileName string `json:"fileName"`
|
||
FileSize string `json:"fileSize"`
|
||
Url string `json:"url"`
|
||
} `json:"data"`
|
||
}
|
||
|
||
type DataDetail struct {
|
||
Id string `json:"id"`
|
||
IdInt int `json:"id_str"`
|
||
Name string `json:"name"`
|
||
Owner string `json:"owner"`
|
||
Identifier string `json:"identifier"`
|
||
}
|
||
|
||
type DataVersionVos struct {
|
||
FileName string `json:"file_name"`
|
||
FileSize string `json:"file_size"`
|
||
Url string `json:"url"`
|
||
}
|
||
|
||
func ParseDataDetail(data string) (*DataDetail, error) {
|
||
var dataDetail DataDetail
|
||
if err := json.Unmarshal([]byte(data), &dataDetail); err != nil {
|
||
return nil, err
|
||
}
|
||
id, err := strconv.Atoi(dataDetail.Id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
dataDetail.IdInt = id
|
||
return &dataDetail, nil
|
||
}
|
||
|
||
func GetToken(url string, name string, password string) (string, error) {
|
||
tokenReq := TokenRequest{
|
||
Name: name,
|
||
Password: password,
|
||
}
|
||
data, err := json.Marshal(tokenReq)
|
||
resp, err := SendPost(&http.Client{Timeout: time.Second * 5}, url, data, "")
|
||
if err != nil {
|
||
fmt.Println("SendPost error:", err)
|
||
return "", err
|
||
}
|
||
var tokenResp TokenResponse
|
||
err = json.Unmarshal(resp, &tokenResp)
|
||
if err != nil {
|
||
fmt.Println("Decode error:", err)
|
||
return "", err
|
||
}
|
||
if tokenResp.Code != CodeOK {
|
||
fmt.Println("get token Code error:", tokenResp.Code)
|
||
return "", fmt.Errorf("get token code error: %d, msg: %s", tokenResp.Code, tokenResp.Msg)
|
||
}
|
||
fmt.Println("get token success, token is:", tokenResp.Data.AccessToken)
|
||
return tokenResp.Data.AccessToken, nil
|
||
}
|
||
|
||
func getDatasetMinioPath(mountPath string) (string, error) {
|
||
podName := os.Getenv("POD_NAME")
|
||
namespace := os.Getenv("POD_NAMESPACE")
|
||
// 使用k8s go client客户端获取pod挂载的pvc
|
||
config, err := rest.InClusterConfig()
|
||
if err != nil {
|
||
fmt.Printf("get config err:%v\n", err)
|
||
return "", err
|
||
}
|
||
|
||
// 创建Kubernetes客户端
|
||
clientset, err := kubernetes.NewForConfig(config)
|
||
if err != nil {
|
||
fmt.Errorf("create kubernetes client err:%v\n", err)
|
||
return "", err
|
||
}
|
||
|
||
fmt.Println("namespace is:", namespace)
|
||
fmt.Println("podName is:", podName)
|
||
pod, err := clientset.CoreV1().Pods(namespace).Get(context.TODO(), podName, metav1.GetOptions{})
|
||
|
||
if err != nil {
|
||
fmt.Errorf("get pod err:%v\n", err)
|
||
return "", err
|
||
}
|
||
var volumeMounts []corev1.VolumeMount
|
||
for _, c := range pod.Spec.Containers {
|
||
if c.Name == "main" {
|
||
volumeMounts = c.VolumeMounts
|
||
}
|
||
}
|
||
|
||
if len(volumeMounts) == 0 {
|
||
return "", fmt.Errorf("no pvc found for pod %s", podName)
|
||
}
|
||
|
||
subPath := ""
|
||
for _, m := range volumeMounts {
|
||
if m.MountPath == mountPath {
|
||
subPath = m.SubPath
|
||
break
|
||
}
|
||
}
|
||
fmt.Println("subpath is ", subPath)
|
||
parts := strings.SplitN(subPath, "argo-workflow/", 2)
|
||
|
||
if len(parts) < 2 {
|
||
fmt.Println("The string does not contain 'argo-workflow/'")
|
||
return "", fmt.Errorf("invalid subPath %s", subPath)
|
||
}
|
||
|
||
// parts[1] 将包含 "argo-workflow/" 后的部分
|
||
fmt.Println("minio path:", parts[1])
|
||
return parts[1], nil
|
||
}
|
||
|
||
func UploadData(url, accessToken string, mountPath string) (*UploadResponse, error) {
|
||
// uuid 为当前的时间戳转换为字符串
|
||
uuid := fmt.Sprintf("%d", time.Now().UnixMilli())
|
||
respBody, err := uploadData(url, uuid, accessToken, mountPath)
|
||
if err != nil {
|
||
fmt.Println("uploadData error:", err)
|
||
return nil, err
|
||
}
|
||
var uploadResp UploadResponse
|
||
err = json.Unmarshal(respBody, &uploadResp)
|
||
if err != nil {
|
||
fmt.Println("Decode error:", err)
|
||
return nil, err
|
||
}
|
||
if uploadResp.Code != CodeOK {
|
||
fmt.Println("upload data Code error:", uploadResp.Code)
|
||
return nil, fmt.Errorf("upload data code error: %d, msg is %s", uploadResp.Code, uploadResp.Msg)
|
||
}
|
||
return &uploadResp, nil
|
||
}
|
||
|
||
func SendPost(client *http.Client, url string, data []byte, token string) ([]byte, error) {
|
||
body := bytes.NewBuffer(data)
|
||
req, err := http.NewRequest(http.MethodPost, url, body)
|
||
|
||
if err != nil {
|
||
fmt.Println("NewRequest error:", err)
|
||
return nil, err
|
||
}
|
||
fmt.Println("url is :", url)
|
||
fmt.Println("data is :", string(data))
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
req.Header.Set("Content-Type", "application/json")
|
||
resp, err := client.Do(req)
|
||
if err != nil {
|
||
fmt.Println("Do error:, url is :", err, url)
|
||
return nil, err
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode != http.StatusOK {
|
||
fmt.Println("StatusCode error:", resp.StatusCode)
|
||
return nil, err
|
||
}
|
||
|
||
respBody, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
fmt.Println("ReadAll error:", err)
|
||
return nil, err
|
||
}
|
||
return respBody, nil
|
||
}
|
||
|
||
// uploadData 上传数据到平台服务， 上传的文件从mountPath中获取， 使用multipart/form-data方式上传
|
||
// 返回值是上传成功后的响应数据
|
||
func uploadData(url string, uuid string, accessToken string, mountPath string) ([]byte, error) {
|
||
body := &bytes.Buffer{}
|
||
writer := multipart.NewWriter(body)
|
||
// mounPath 是一个目录，需要遍历目录下的文件进行上传
|
||
files, err := filepath.Glob(mountPath + "/*")
|
||
if err != nil {
|
||
fmt.Println("获取文件列表失败:", err)
|
||
return nil, err
|
||
}
|
||
for _, file := range files {
|
||
part, err := writer.CreateFormFile("file", filepath.Base(file))
|
||
if err != nil {
|
||
fmt.Println("创建form file失败:", err)
|
||
return nil, err
|
||
}
|
||
f, err := os.Open(file)
|
||
if err != nil {
|
||
fmt.Println("打开文件失败:", err)
|
||
return nil, err
|
||
}
|
||
defer f.Close()
|
||
_, err = io.Copy(part, f)
|
||
if err != nil {
|
||
fmt.Println("写入文件失败:", err)
|
||
return nil, err
|
||
}
|
||
}
|
||
// 写入uuid参数
|
||
err = writer.WriteField("uuid", uuid)
|
||
if err != nil {
|
||
fmt.Println("写入uuid参数失败:", err)
|
||
return nil, err
|
||
}
|
||
writer.Close()
|
||
req, err := http.NewRequest(http.MethodPost, url, body)
|
||
if err != nil {
|
||
fmt.Println("NewRequest error:", err)
|
||
return nil, err
|
||
}
|
||
req.Header.Set("Authorization", "Bearer "+accessToken)
|
||
req.Header.Set("Content-Type", writer.FormDataContentType())
|
||
resp, err := http.DefaultClient.Do(req)
|
||
if err != nil {
|
||
fmt.Println("Do error:", err)
|
||
return nil, err
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
if resp.StatusCode != http.StatusOK {
|
||
fmt.Println("StatusCode error:", resp.StatusCode)
|
||
return nil, err
|
||
}
|
||
// 解析响应数据, 不用ReadAll, 因为已经deprecate
|
||
respBody, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
fmt.Println("ReadAll error:", err)
|
||
return nil, err
|
||
}
|
||
return respBody, err
|
||
}
|