730 lines
24 KiB
Go
730 lines
24 KiB
Go
package mq
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"context"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"fmt"
|
||
"git.hpds.cc/Component/network/frame"
|
||
"git.hpds.cc/pavement/hpds_node"
|
||
"github.com/docker/docker/api/types/registry"
|
||
"github.com/emirpasic/gods/lists/arraylist"
|
||
"github.com/fsnotify/fsnotify"
|
||
"github.com/gammazero/workerpool"
|
||
"github.com/minio/minio-go/v7"
|
||
"github.com/minio/minio-go/v7/pkg/credentials"
|
||
"github.com/shirou/gopsutil/v3/host"
|
||
"go.uber.org/zap"
|
||
"io"
|
||
"mime/multipart"
|
||
"net/http"
|
||
"os"
|
||
"os/exec"
|
||
"path"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"taskExecute/config"
|
||
"taskExecute/pkg/compress"
|
||
"taskExecute/pkg/docker"
|
||
"taskExecute/pkg/download"
|
||
"taskExecute/pkg/utils"
|
||
"taskExecute/proto"
|
||
"time"
|
||
)
|
||
|
||
var (
|
||
wg sync.WaitGroup
|
||
TaskList map[string]docker.ContainerStore
|
||
)
|
||
|
||
func TaskExecuteHandler(data []byte) (frame.Tag, []byte) {
|
||
//fmt.Println("接收数据", string(data))
|
||
cmd := new(InstructionReq)
|
||
err := json.Unmarshal(data, cmd)
|
||
if err != nil {
|
||
return 0x0B, []byte(err.Error())
|
||
}
|
||
switch cmd.Command {
|
||
case TaskExecute:
|
||
//任务执行
|
||
waitWorkerStartFinish(config.WorkPool, cmd.Payload.(map[string]interface{}), ModelTaskExecuteHandler)
|
||
case ModelIssueRepeater:
|
||
//模型下发
|
||
waitWorkerStartFinish(config.WorkPool, cmd.Payload.(map[string]interface{}), ModelIssueRepeaterHandler)
|
||
}
|
||
return frame.Tag(cmd.Command), nil
|
||
}
|
||
|
||
func waitWorkerStartFinish(wp *workerpool.WorkerPool, payload map[string]interface{}, f func(payload map[string]interface{})) {
|
||
startStop := make(chan time.Time, 2)
|
||
wp.Submit(func() {
|
||
startStop <- time.Now()
|
||
f(payload)
|
||
startStop <- time.Now()
|
||
})
|
||
fmt.Println("Task started at:", <-startStop)
|
||
fmt.Println("Task finished at:", <-startStop)
|
||
}
|
||
|
||
// execCommand 执行命令
|
||
func execCommandWait(commandName string, params []string) bool {
|
||
cmd := exec.Command(commandName, params...)
|
||
|
||
//显示运行的命令
|
||
fmt.Println(cmd.Args)
|
||
|
||
stdout, err := cmd.StdoutPipe()
|
||
|
||
if err != nil {
|
||
fmt.Println(err)
|
||
return false
|
||
}
|
||
|
||
_ = cmd.Start()
|
||
|
||
reader := bufio.NewReader(stdout)
|
||
|
||
//实时循环读取输出流中的一行内容
|
||
for {
|
||
wg.Add(1)
|
||
line, err2 := reader.ReadString('\n')
|
||
if err2 != nil || io.EOF == err2 {
|
||
break
|
||
}
|
||
config.Logger.Info("执行命令",
|
||
zap.String("execCommandWait", line))
|
||
}
|
||
|
||
_ = cmd.Wait()
|
||
wg.Done()
|
||
return true
|
||
}
|
||
|
||
func ModelIssueRepeaterHandler(payload map[string]interface{}) {
|
||
hi, _ := host.Info()
|
||
if payload["nodeGuid"].(string) == hi.HostID {
|
||
fileUrl := payload["dockerFile"].(string)
|
||
modelVersion := payload["modelVersion"].(string)
|
||
downFileName := path.Base(fileUrl)
|
||
//判断文件后缀名
|
||
fileType := path.Ext(downFileName)
|
||
fileNameOnly := strings.TrimSuffix(downFileName, fileType)
|
||
dFile := path.Join(config.Cfg.TmpPath, fileNameOnly, downFileName)
|
||
//执行文件下载
|
||
controller := download.ThreadController{
|
||
ThreadCount: download.ThreadCount,
|
||
FileUrl: fileUrl,
|
||
DownloadFolder: dFile,
|
||
DownloadFileName: downFileName,
|
||
Logger: config.Logger,
|
||
}
|
||
controller.Download(download.OneThreadDownloadSize)
|
||
if strings.ToLower(fileType) == ".zip" {
|
||
err := compress.UnzipFromFile(path.Join(config.Cfg.TmpPath, fileNameOnly), dFile)
|
||
if err != nil {
|
||
controller.Logger.With(zap.String("文件解压缩", path.Join(config.Cfg.TmpPath, downFileName))).
|
||
Error("发生错误", zap.Error(err))
|
||
return
|
||
}
|
||
dFile = path.Join(config.Cfg.TmpPath, fileNameOnly, fileNameOnly+".tar")
|
||
}
|
||
//docker 导入并运行
|
||
imgName := fmt.Sprintf("%s:%s", fileNameOnly, modelVersion)
|
||
if strings.ToLower(path.Ext(dFile)) == ".tar" {
|
||
dCli := docker.NewDockerClient()
|
||
err := dCli.ImportImage(imgName, "latest", dFile)
|
||
//err = dCli.LoadImage(dFile)
|
||
if err != nil {
|
||
controller.Logger.With(zap.String("导入docker的文件", dFile)).
|
||
Error("发生错误", zap.Error(err))
|
||
}
|
||
//设置data目录
|
||
dataPath := path.Join(config.Cfg.TmpPath, fileNameOnly, "data")
|
||
_ = os.MkdirAll(dataPath, os.ModePerm)
|
||
vol := make(map[string]string)
|
||
vol[path.Join(dataPath, payload["inPath"].(string))] = payload["inPath"].(string)
|
||
vol[path.Join(dataPath, payload["outPath"].(string))] = payload["outPath"].(string)
|
||
//docker运行
|
||
modelCommand := strings.Split(payload["modelCommand"].(string), " ")
|
||
|
||
dstPort := dCli.CreateContainer(fileNameOnly, imgName, modelCommand, vol, strconv.Itoa(payload["mappedPort"].(int)))
|
||
//保存到本地临时文件
|
||
item := docker.ContainerStore{
|
||
ModelId: payload["modelId"].(int64),
|
||
NodeId: payload["nodeId"].(int64),
|
||
Name: fileNameOnly,
|
||
ImgName: imgName,
|
||
Volumes: vol,
|
||
SrcPort: strconv.Itoa(payload["mappedPort"].(int)),
|
||
DstPort: dstPort,
|
||
Command: modelCommand,
|
||
HttpUrl: payload["httpUrl"].(string),
|
||
}
|
||
docker.ContainerList = append(docker.ContainerList, item)
|
||
docker.SaveStore()
|
||
cli := GetMqClient("task-response", 1)
|
||
ap := cli.EndPoint.(hpds_node.AccessPoint)
|
||
res := new(InstructionReq)
|
||
res.Command = ModelIssueResponse
|
||
res.Payload = item
|
||
pData, _ := json.Marshal(res)
|
||
_ = GenerateAndSendData(ap, pData)
|
||
}
|
||
}
|
||
}
|
||
|
||
func ModelTaskExecuteHandler(payload map[string]interface{}) {
|
||
hi, _ := host.Info()
|
||
currTime := time.Now().Unix()
|
||
//config.Logger.With(zap.Any("收到数据", payload))
|
||
if payload["nodeGuid"] == hi.HostID {
|
||
if len(payload["subDataset"].(string)) > 0 { //如果是订阅数据
|
||
sf := hpds_node.NewStreamFunction(
|
||
payload["subDataset"].(string),
|
||
hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", config.Cfg.Node.Host, config.Cfg.Node.Port)),
|
||
hpds_node.WithObserveDataTags(frame.Tag(payload["subDataTag"].(byte))),
|
||
hpds_node.WithCredential(config.Cfg.Node.Token),
|
||
)
|
||
err := sf.Connect()
|
||
must(config.Logger, err)
|
||
nodeInfo := HpdsMqNode{
|
||
MqType: 2,
|
||
Topic: payload["subDataset"].(string),
|
||
Node: config.Cfg.Node,
|
||
EndPoint: sf,
|
||
}
|
||
_ = sf.SetHandler(func(data []byte) (frame.Tag, []byte) {
|
||
|
||
//查询docker是否已经开启
|
||
issue := new(docker.ContainerStore)
|
||
_ = json.Unmarshal([]byte(payload["issueResult"].(string)), issue)
|
||
dCli := docker.NewDockerClient()
|
||
cList, err := dCli.SearchImage(issue.Name)
|
||
if err != nil {
|
||
|
||
}
|
||
if len(cList) > 0 {
|
||
if len(payload["workflow"].(string)) > 0 {
|
||
//是否设置工作流程
|
||
wf := new(Workflow)
|
||
err = json.Unmarshal([]byte(payload["workflow"].(string)), wf)
|
||
if err != nil {
|
||
|
||
}
|
||
if len(payload["datasetPath"].(string)) > 0 {
|
||
//数据集处理
|
||
opt := &minio.Options{
|
||
Creds: credentials.NewStaticV4(config.Cfg.Minio.AccessKeyId, config.Cfg.Minio.SecretAccessKey, ""),
|
||
Secure: false,
|
||
}
|
||
cli, _ := minio.New(config.Cfg.Minio.Endpoint, opt)
|
||
doneCh := make(chan struct{})
|
||
defer close(doneCh)
|
||
objectCh := cli.ListObjects(context.Background(), config.Cfg.Minio.Bucket, minio.ListObjectsOptions{
|
||
Prefix: payload["datasetPath"].(string),
|
||
Recursive: true,
|
||
})
|
||
for object := range objectCh {
|
||
file, _ := cli.GetObject(context.Background(), config.Cfg.Minio.Bucket, object.Key, minio.GetObjectOptions{})
|
||
imgByte, _ := io.ReadAll(file)
|
||
|
||
f := proto.FileCapture{
|
||
FileName: object.Key,
|
||
File: base64.StdEncoding.EncodeToString(imgByte),
|
||
DatasetName: payload["datasetName"].(string),
|
||
CaptureTime: object.LastModified.Unix(),
|
||
}
|
||
ProcessWorkflow(payload, f, wf)
|
||
}
|
||
|
||
}
|
||
} else {
|
||
f := new(proto.FileCapture)
|
||
err := json.Unmarshal(data, f)
|
||
if err != nil {
|
||
|
||
}
|
||
if len(f.File) > 0 {
|
||
|
||
i := strings.Index(f.File, ",")
|
||
dec := base64.NewDecoder(base64.StdEncoding, strings.NewReader(f.File[i+1:]))
|
||
if len(payload["httpUrl"].(string)) > 0 {
|
||
_ = os.MkdirAll(path.Join(config.Cfg.TmpPath, payload["subDataset"].(string)), os.ModePerm)
|
||
tmpFile, _ := os.Create(path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))
|
||
_, err = io.Copy(tmpFile, dec)
|
||
|
||
reqUrl := fmt.Sprintf("http://localhost:%s/%s", issue.DstPort, issue.HttpUrl)
|
||
response, err := http.Post(reqUrl, "multipart/form-data", dec)
|
||
if err != nil {
|
||
config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
With(zap.String("请求地址", reqUrl)).
|
||
With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
Error("文件提交", zap.Error(err))
|
||
}
|
||
defer func() {
|
||
_ = response.Body.Close()
|
||
config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
Info("模型识别")
|
||
}()
|
||
body, err := io.ReadAll(response.Body)
|
||
if err != nil {
|
||
config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
With(zap.String("请求地址", reqUrl)).
|
||
With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
Error("模型识别", zap.Error(err))
|
||
}
|
||
cli := GetMqClient("task-response", 1)
|
||
ap := cli.EndPoint.(hpds_node.AccessPoint)
|
||
res := new(InstructionReq)
|
||
res.Command = TaskResponse
|
||
res.Payload = body
|
||
pData, _ := json.Marshal(res)
|
||
_ = GenerateAndSendData(ap, pData)
|
||
}
|
||
if len(payload["inPath"].(string)) > 0 {
|
||
outPath := ""
|
||
for k, v := range issue.Volumes {
|
||
if v == payload["outPath"].(string) {
|
||
outPath = k
|
||
break
|
||
}
|
||
}
|
||
//创建一个监控对象
|
||
watch, err := fsnotify.NewWatcher()
|
||
if err != nil {
|
||
config.Logger.Error("创建文件监控", zap.Error(err))
|
||
}
|
||
defer func(watch *fsnotify.Watcher) {
|
||
_ = watch.Close()
|
||
}(watch)
|
||
|
||
err = watch.Add(outPath)
|
||
if err != nil {
|
||
config.Logger.With(zap.String("监控目录", outPath)).
|
||
Error("创建文件监控", zap.Error(err))
|
||
}
|
||
for k, v := range issue.Volumes {
|
||
if v == payload["inPath"].(string) {
|
||
_ = os.MkdirAll(k, os.ModePerm)
|
||
tmpFile, _ := os.Create(path.Join(k, f.FileName))
|
||
_, err = io.Copy(tmpFile, dec)
|
||
break
|
||
}
|
||
}
|
||
list := arraylist.New() // empty
|
||
t1 := time.NewTicker(1 * time.Second)
|
||
go func() {
|
||
for {
|
||
select {
|
||
case ev := <-watch.Events:
|
||
{
|
||
//判断事件发生的类型，如下5种
|
||
// Create 创建
|
||
// Write 写入
|
||
// Remove 删除
|
||
// Rename 重命名
|
||
// Chmod 修改权限
|
||
if ev.Op&fsnotify.Create == fsnotify.Create {
|
||
config.Logger.Info("创建文件", zap.String("文件名", ev.Name))
|
||
list.Add(ev.Name)
|
||
}
|
||
}
|
||
case <-t1.C:
|
||
{
|
||
if list.Size() > 0 {
|
||
returnFileHandleResult(list, payload, issue)
|
||
}
|
||
}
|
||
case err = <-watch.Errors:
|
||
{
|
||
config.Logger.With(zap.String("监控目录", outPath)).
|
||
Error("文件监控", zap.Error(err))
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return frame.Tag(payload["subDataTag"].(byte)), nil
|
||
})
|
||
MqList = append(MqList, nodeInfo)
|
||
} else {
|
||
|
||
//查询docker是否已经开启
|
||
issue := new(docker.ContainerStore)
|
||
cList := make([]registry.SearchResult, 0)
|
||
var err error
|
||
issueResult, ok := payload["issueResult"]
|
||
if ok {
|
||
if len(issueResult.(string)) > 0 {
|
||
_ = json.Unmarshal([]byte(payload["issueResult"].(string)), issue)
|
||
dCli := docker.NewDockerClient()
|
||
cList, err = dCli.SearchImage(issue.Name)
|
||
if err != nil {
|
||
|
||
}
|
||
if len(cList) < 1 {
|
||
//TODO: 启动容器
|
||
}
|
||
}
|
||
} else {
|
||
issue.DstPort = "5000"
|
||
issue.HttpUrl = ""
|
||
}
|
||
//如果是数据集任务
|
||
if _, ok := payload["single"]; ok {
|
||
f := payload["single"].(map[string]interface{})
|
||
if len(payload["workflow"].(string)) > 0 {
|
||
//是否设置工作流程
|
||
wf := new(Workflow)
|
||
err = json.Unmarshal([]byte(payload["workflow"].(string)), wf)
|
||
if err != nil {
|
||
|
||
}
|
||
fn := proto.FileCapture{
|
||
FileId: int64(f["fileId"].(float64)),
|
||
FileName: f["fileName"].(string),
|
||
File: f["file"].(string),
|
||
DatasetName: payload["datasetName"].(string),
|
||
CaptureTime: int64(f["captureTime"].(float64)),
|
||
}
|
||
ProcessWorkflow(payload, fn, wf)
|
||
//if len(payload["datasetPath"].(string)) > 0 {
|
||
// //数据集处理
|
||
// opt := &minio.Options{
|
||
// Creds: credentials.NewStaticV4(config.Cfg.Minio.AccessKeyId, config.Cfg.Minio.SecretAccessKey, ""),
|
||
// Secure: false,
|
||
// }
|
||
// cli, _ := minio.New(config.Cfg.Minio.Endpoint, opt)
|
||
// doneCh := make(chan struct{})
|
||
// defer close(doneCh)
|
||
// objectCh := cli.ListObjects(context.Background(), config.Cfg.Minio.Bucket, minio.ListObjectsOptions{
|
||
// Prefix: payload["datasetPath"].(string),
|
||
// Recursive: true,
|
||
// })
|
||
// for object := range objectCh {
|
||
// file, _ := cli.GetObject(context.Background(), config.Cfg.Minio.Bucket, object.Key, minio.GetObjectOptions{})
|
||
// imgByte, _ := io.ReadAll(file)
|
||
//
|
||
// f := proto.FileCapture{
|
||
// FileName: object.Key,
|
||
// File: base64.StdEncoding.EncodeToString(imgByte),
|
||
// DatasetName: payload["datasetName"].(string),
|
||
// CaptureTime: object.LastModified.Unix(),
|
||
// }
|
||
// ProcessWorkflow(payload, f, wf)
|
||
// }
|
||
//
|
||
//}
|
||
} else {
|
||
reqUrl := fmt.Sprintf("http://127.0.0.1:%s/%s", issue.DstPort, issue.HttpUrl)
|
||
if len(payload["httpUrl"].(string)) > 1 {
|
||
reqUrl = payload["httpUrl"].(string)
|
||
}
|
||
fc := f["file"].(string)
|
||
dec := base64.NewDecoder(base64.StdEncoding, strings.NewReader(fc))
|
||
|
||
buff := new(bytes.Buffer)
|
||
writer := multipart.NewWriter(buff)
|
||
formFile, _ := writer.CreateFormFile("image", f["fileName"].(string))
|
||
_, _ = io.Copy(formFile, dec)
|
||
_ = writer.Close()
|
||
|
||
response, err := http.Post(reqUrl, writer.FormDataContentType(), buff)
|
||
if err != nil {
|
||
config.Logger.With(zap.String("源文件名", f["fileName"].(string))).
|
||
With(zap.String("请求地址", reqUrl)).
|
||
With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f["fileName"].(string)))).
|
||
Error("文件提交", zap.Error(err))
|
||
}
|
||
defer func() {
|
||
_ = response.Body.Close()
|
||
}()
|
||
body, err := io.ReadAll(response.Body)
|
||
if err != nil {
|
||
config.Logger.With(zap.String("源文件名", f["fileName"].(string))).
|
||
With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f["fileName"].(string)))).
|
||
Error("模型识别", zap.Error(err))
|
||
}
|
||
cli := GetMqClient("task-response", 1)
|
||
pay := make(map[string]interface{})
|
||
pay["taskId"] = payload["taskId"]
|
||
pay["fileId"] = f["fileId"]
|
||
pay["taskCode"] = payload["taskCode"]
|
||
pay["nodeId"] = payload["nodeId"]
|
||
pay["modelId"] = payload["modelId"]
|
||
pay["startTime"] = currTime
|
||
pay["finishTime"] = time.Now().Unix()
|
||
pay["subDataset"] = payload["subDataset"]
|
||
pay["datasetArr"] = payload["datasetArr"]
|
||
pay["srcPath"] = f["fileName"]
|
||
pay["body"] = string(body)
|
||
ap := cli.EndPoint.(hpds_node.AccessPoint)
|
||
res := new(InstructionReq)
|
||
res.Command = TaskResponse
|
||
res.Payload = pay
|
||
pData, _ := json.Marshal(res)
|
||
_ = GenerateAndSendData(ap, pData)
|
||
|
||
}
|
||
}
|
||
|
||
//if len(cList) > 0 {
|
||
//if len(payload["workflow"].(string)) > 0 {
|
||
// //是否设置工作流程
|
||
// wf := new(Workflow)
|
||
// err = json.Unmarshal([]byte(payload["workflow"].(string)), wf)
|
||
// if err != nil {
|
||
//
|
||
// }
|
||
// if len(payload["datasetPath"].(string)) > 0 {
|
||
// //数据集处理
|
||
// opt := &minio.Options{
|
||
// Creds: credentials.NewStaticV4(config.Cfg.Minio.AccessKeyId, config.Cfg.Minio.SecretAccessKey, ""),
|
||
// Secure: false,
|
||
// }
|
||
// cli, _ := minio.New(config.Cfg.Minio.Endpoint, opt)
|
||
// doneCh := make(chan struct{})
|
||
// defer close(doneCh)
|
||
// objectCh := cli.ListObjects(context.Background(), config.Cfg.Minio.Bucket, minio.ListObjectsOptions{
|
||
// Prefix: payload["datasetPath"].(string),
|
||
// Recursive: true,
|
||
// })
|
||
// for object := range objectCh {
|
||
// file, _ := cli.GetObject(context.Background(), config.Cfg.Minio.Bucket, object.Key, minio.GetObjectOptions{})
|
||
// imgByte, _ := io.ReadAll(file)
|
||
//
|
||
// f := proto.FileCapture{
|
||
// FileName: object.Key,
|
||
// File: base64.StdEncoding.EncodeToString(imgByte),
|
||
// DatasetName: payload["datasetName"].(string),
|
||
// CaptureTime: object.LastModified.Unix(),
|
||
// }
|
||
// ProcessWorkflow(payload, f, wf)
|
||
// }
|
||
//
|
||
// }
|
||
//} else {
|
||
// if len(payload["datasetPath"].(string)) > 0 {
|
||
// //数据集处理
|
||
// opt := &minio.Options{
|
||
// Creds: credentials.NewStaticV4(config.Cfg.Minio.AccessKeyId, config.Cfg.Minio.SecretAccessKey, ""),
|
||
// Secure: false,
|
||
// }
|
||
// cli, _ := minio.New(config.Cfg.Minio.Endpoint, opt)
|
||
// doneCh := make(chan struct{})
|
||
// defer close(doneCh)
|
||
// objectCh := cli.ListObjects(context.Background(), config.Cfg.Minio.Bucket, minio.ListObjectsOptions{
|
||
// Prefix: payload["datasetPath"].(string),
|
||
// Recursive: true,
|
||
// })
|
||
// for object := range objectCh {
|
||
// file, _ := cli.GetObject(context.Background(), config.Cfg.Minio.Bucket, object.Key, minio.GetObjectOptions{})
|
||
// imgByte, _ := io.ReadAll(file)
|
||
//
|
||
// f := proto.FileCapture{
|
||
// FileName: object.Key,
|
||
// File: base64.StdEncoding.EncodeToString(imgByte),
|
||
// DatasetName: payload["datasetName"].(string),
|
||
// CaptureTime: object.LastModified.Unix(),
|
||
// }
|
||
//
|
||
// if len(payload["httpUrl"].(string)) > 0 {
|
||
// dec := base64.NewDecoder(base64.StdEncoding, strings.NewReader(f.File))
|
||
// reqUrl := fmt.Sprintf("http://localhost:%s/%s", issue.DstPort, issue.HttpUrl)
|
||
// response, err := http.Post(reqUrl, "multipart/form-data", dec)
|
||
// if err != nil {
|
||
// config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
// With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
// Error("文件提交", zap.Error(err))
|
||
// }
|
||
// defer func() {
|
||
// _ = response.Body.Close()
|
||
// config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
// With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
// Info("模型识别")
|
||
// }()
|
||
// body, err := io.ReadAll(response.Body)
|
||
// if err != nil {
|
||
// config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
// With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
// Error("模型识别", zap.Error(err))
|
||
// }
|
||
// cli := GetMqClient("task-response", 1)
|
||
// ap := cli.EndPoint.(hpds_node.AccessPoint)
|
||
// res := new(InstructionReq)
|
||
// res.Command = TaskResponse
|
||
// res.Payload = body
|
||
// pData, _ := json.Marshal(res)
|
||
// _ = GenerateAndSendData(ap, pData)
|
||
// }
|
||
// }
|
||
// }
|
||
//f := new(proto.FileCapture)
|
||
//err := json.Unmarshal(data, f)
|
||
//if err != nil {
|
||
//
|
||
//}
|
||
//if len(f.File) > 0 {
|
||
//
|
||
// i := strings.Index(f.File, ",")
|
||
// dec := base64.NewDecoder(base64.StdEncoding, strings.NewReader(f.File[i+1:]))
|
||
// if len(payload["httpUrl"].(string)) > 0 {
|
||
// _ = os.MkdirAll(path.Join(config.Cfg.TmpPath, payload["subDataset"].(string)), os.ModePerm)
|
||
// tmpFile, _ := os.Create(path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))
|
||
// _, err = io.Copy(tmpFile, dec)
|
||
//
|
||
// reqUrl := fmt.Sprintf("http://localhost:%s/%s", issue.DstPort, issue.HttpUrl)
|
||
// response, err := http.Post(reqUrl, "multipart/form-data", dec)
|
||
// if err != nil {
|
||
// config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
// With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
// Error("文件提交", zap.Error(err))
|
||
// }
|
||
// defer func() {
|
||
// _ = response.Body.Close()
|
||
// config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
// With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
// Info("模型识别")
|
||
// }()
|
||
// body, err := io.ReadAll(response.Body)
|
||
// if err != nil {
|
||
// config.Logger.With(zap.String("源文件名", f.FileName)).
|
||
// With(zap.String("临时文件名", path.Join(config.Cfg.TmpPath, payload["subDataset"].(string), f.FileName))).
|
||
// Error("模型识别", zap.Error(err))
|
||
// }
|
||
// cli := GetMqClient("task-response", 1)
|
||
// ap := cli.EndPoint.(hpds_node.AccessPoint)
|
||
// res := new(InstructionReq)
|
||
// res.Command = TaskResponse
|
||
// res.Payload = body
|
||
// pData, _ := json.Marshal(res)
|
||
// _ = GenerateAndSendData(ap, pData)
|
||
// }
|
||
// if len(payload["inPath"].(string)) > 0 {
|
||
// outPath := ""
|
||
// for k, v := range issue.Volumes {
|
||
// if v == payload["outPath"].(string) {
|
||
// outPath = k
|
||
// break
|
||
// }
|
||
// }
|
||
// //创建一个监控对象
|
||
// watch, err := fsnotify.NewWatcher()
|
||
// if err != nil {
|
||
// config.Logger.Error("创建文件监控", zap.Error(err))
|
||
// }
|
||
// defer func(watch *fsnotify.Watcher) {
|
||
// _ = watch.Close()
|
||
// }(watch)
|
||
//
|
||
// err = watch.Add(outPath)
|
||
// if err != nil {
|
||
// config.Logger.With(zap.String("监控目录", outPath)).
|
||
// Error("创建文件监控", zap.Error(err))
|
||
// }
|
||
// for k, v := range issue.Volumes {
|
||
// if v == payload["inPath"].(string) {
|
||
// _ = os.MkdirAll(k, os.ModePerm)
|
||
// tmpFile, _ := os.Create(path.Join(k, f.FileName))
|
||
// _, err = io.Copy(tmpFile, dec)
|
||
// break
|
||
// }
|
||
// }
|
||
// list := arraylist.New() // empty
|
||
// t1 := time.NewTicker(1 * time.Second)
|
||
// go func() {
|
||
// for {
|
||
// select {
|
||
// case ev := <-watch.Events:
|
||
// {
|
||
// //判断事件发生的类型，如下5种
|
||
// // Create 创建
|
||
// // Write 写入
|
||
// // Remove 删除
|
||
// // Rename 重命名
|
||
// // Chmod 修改权限
|
||
// if ev.Op&fsnotify.Create == fsnotify.Create {
|
||
// config.Logger.Info("创建文件", zap.String("文件名", ev.Name))
|
||
// list.Add(ev.Name)
|
||
// }
|
||
// }
|
||
// case <-t1.C:
|
||
// {
|
||
// if list.Size() > 0 {
|
||
// returnFileHandleResult(list, payload, issue)
|
||
// }
|
||
// }
|
||
// case err = <-watch.Errors:
|
||
// {
|
||
// config.Logger.With(zap.String("监控目录", outPath)).
|
||
// Error("文件监控", zap.Error(err))
|
||
// return
|
||
// }
|
||
// }
|
||
// }
|
||
// }()
|
||
// }
|
||
//}
|
||
}
|
||
//}
|
||
}
|
||
}
|
||
|
||
func returnFileHandleResult(list *arraylist.List, payload map[string]interface{}, issue *docker.ContainerStore) {
|
||
var (
|
||
mu sync.RWMutex
|
||
wgp sync.WaitGroup
|
||
resTime time.Duration
|
||
)
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
startTime := time.Now()
|
||
for i := 0; i < list.Size(); i++ {
|
||
if fn, ok := list.Get(0); ok {
|
||
if utils.PathExists(fn.(string)) {
|
||
wgp.Add(1)
|
||
go func() {
|
||
mr := new(proto.ModelResult)
|
||
src := utils.ReadFile(fn.(string))
|
||
|
||
if src != nil {
|
||
mr.File = base64.StdEncoding.EncodeToString(src)
|
||
mr.TaskCode = utils.GetFileName(fn.(string))
|
||
mr.TaskId = int64(payload["taskId"].(float64))
|
||
mr.FileName = utils.GetFileNameAndExt(fn.(string))
|
||
mr.DatasetName = payload["datasetName"].(string)
|
||
mr.SubDataset = payload["subDataset"].(string)
|
||
mr.FileMd5 = utils.GetFileMd5(src)
|
||
mr.ModelId = int64(payload["modelId"].(float64))
|
||
mr.NodeId = int64(payload["nodeId"].(float64))
|
||
mr.StartTime = startTime.Unix()
|
||
mr.FinishTime = time.Now().Unix()
|
||
cli := GetMqClient("task-response", 1)
|
||
ap := cli.EndPoint.(hpds_node.AccessPoint)
|
||
res := new(InstructionReq)
|
||
res.Command = TaskResponse
|
||
res.Payload = mr
|
||
pData, _ := json.Marshal(res)
|
||
_ = GenerateAndSendData(ap, pData)
|
||
}
|
||
wg.Done()
|
||
}()
|
||
wg.Wait()
|
||
resTime = time.Since(startTime)
|
||
config.Logger.Info("返回任务完成",
|
||
zap.String("文件名", fn.(string)),
|
||
zap.Duration("运行时间", resTime),
|
||
)
|
||
}
|
||
}
|
||
}
|
||
}
|