From 685b46c1e5ac12e527f2b2b2f13e2c52b8382477 Mon Sep 17 00:00:00 2001 From: wangjian Date: Mon, 24 Apr 2023 15:14:04 +0800 Subject: [PATCH] =?UTF-8?q?1=E3=80=81=E5=A2=9E=E5=8A=A0=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E5=A4=84=E7=90=86=E8=BF=9B=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/server.go | 4 +- config/config-dev.yaml | 16 +- config/config.go | 18 +- config/config.yaml | 13 +- go.mod | 37 ++- internal/balance/hash.go | 14 +- internal/balance/index.go | 2 +- internal/balance/randomBalance.go | 11 +- internal/balance/roundRobinBalance.go | 11 +- internal/balance/weightRoundRobinBalance.go | 14 +- internal/minio/index.go | 22 +- internal/proto/proto.go | 20 ++ model/TaskLog.go | 14 + model/dataset.go | 15 ++ model/file.go | 15 ++ model/index.go | 1 + model/task.go | 60 +++++ model/taskResult.go | 3 +- mq/index.go | 387 ++++++++++++++++++++++------ mq/instruction.go | 27 ++ test/task_test.go | 11 + 21 files changed, 575 insertions(+), 140 deletions(-) create mode 100644 internal/proto/proto.go create mode 100644 model/TaskLog.go create mode 100644 model/dataset.go create mode 100644 model/file.go create mode 100644 test/task_test.go diff --git a/cmd/server.go b/cmd/server.go index f9920ad..b351e3a 100644 --- a/cmd/server.go +++ b/cmd/server.go @@ -36,7 +36,7 @@ func NewStartCmd() *cobra.Command { Short: "Start hpds_web application", Run: func(cmd *cobra.Command, args []string) { var ( - cfg *config.WebConfig + cfg *config.ControlCenterConfig err error ) ctx, cancel := context.WithCancel(context.Background()) @@ -80,7 +80,7 @@ func NewStartCmd() *cobra.Command { consulCfg, err := discover.NewConsulConfig(fmt.Sprintf("%s:%d", cfg.Consul.Host, cfg.Consul.Port), cfg.Name, cfg.Name, cfg.Consul.Host, cfg.Consul.Port, tags, 300, 300, 300) must(err) - + config.Cfg = cfg //连接数据库 model.New(cfg.Db.DriveName, cfg.Db.Conn, cfg.Mode == "dev") diff --git a/config/config-dev.yaml b/config/config-dev.yaml index b356215..1f16fbc 100644 --- a/config/config-dev.yaml +++ b/config/config-dev.yaml @@ -27,6 +27,12 @@ cache: port: 6379 db: 0 pool_size: 10 +minio: + protocol: http + endpoint: 127.0.0.1:9000 + accessKeyId: root + secretAccessKey: OIxv7QptYBO3 + bucket: jky-data node: host: 127.0.0.1 port: 27188 @@ -36,8 +42,14 @@ functions: dataTag: 12 mqType: 2 - name: task-response - dataTag: 14 - mqType: 1 + dataTag: 24 + mqType: 2 - name: task-execute dataTag: 16 + mqType: 1 + - name: task-execute-log + dataTag: 26 + mqType: 2 + - name: task-log + dataTag: 28 mqType: 1 \ No newline at end of file diff --git a/config/config.go b/config/config.go index 95b825b..0e25aa8 100644 --- a/config/config.go +++ b/config/config.go @@ -10,7 +10,11 @@ import ( yaml "gopkg.in/yaml.v3" ) -type WebConfig struct { +var ( + Cfg *ControlCenterConfig +) + +type ControlCenterConfig struct { Name string `yaml:"name,omitempty"` Host string `yaml:"host,omitempty"` Port int `yaml:"port,omitempty"` @@ -77,7 +81,7 @@ type FuncConfig struct { MqType uint `yaml:"mqType"` //消息类型， 发布，1；订阅；2 } -func ParseConfigByFile(path string) (cfg *WebConfig, err error) { +func ParseConfigByFile(path string) (cfg *ControlCenterConfig, err error) { buffer, err := os.ReadFile(path) if err != nil { return nil, err @@ -85,10 +89,10 @@ func ParseConfigByFile(path string) (cfg *WebConfig, err error) { return load(buffer) } -func load(buf []byte) (cfg *WebConfig, err error) { +func load(buf []byte) (cfg *ControlCenterConfig, err error) { cViper := viper.New() cViper.SetConfigType("yaml") - cfg = new(WebConfig) + cfg = new(ControlCenterConfig) cfg.Funcs = make([]FuncConfig, 0) //cViper.ReadConfig(bytes.NewBuffer(buf)) err = yaml.Unmarshal(buf, cfg) @@ -99,7 +103,7 @@ func load(buf []byte) (cfg *WebConfig, err error) { return } -func UpdateLocalConfig(cfg *WebConfig, fn string) error { +func UpdateLocalConfig(cfg *ControlCenterConfig, fn string) error { data, err := yaml.Marshal(cfg) if err != nil { return err @@ -108,7 +112,7 @@ func UpdateLocalConfig(cfg *WebConfig, fn string) error { return err } -func UpdateRemoteConfig(cfg *WebConfig) error { +func UpdateRemoteConfig(cfg *ControlCenterConfig) error { consulClient, err := consulapi.NewClient(&consulapi.Config{Address: fmt.Sprintf("%s:%d", cfg.Consul.Host, cfg.Consul.Port)}) if err != nil { return err @@ -124,7 +128,7 @@ func UpdateRemoteConfig(cfg *WebConfig) error { return nil } -func GetRemoteConfig(remoteAddr, path string) (cfg *WebConfig, err error) { +func GetRemoteConfig(remoteAddr, path string) (cfg *ControlCenterConfig, err error) { consulClient, err := consulapi.NewClient(&consulapi.Config{Address: remoteAddr}) if err != nil { return nil, err diff --git a/config/config.yaml b/config/config.yaml index 7ef3888..1f16fbc 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -32,17 +32,24 @@ minio: endpoint: 127.0.0.1:9000 accessKeyId: root secretAccessKey: OIxv7QptYBO3 + bucket: jky-data node: - host: 114.55.236.153 - port: 9188 + host: 127.0.0.1 + port: 27188 token: 06d36c6f5705507dae778fdce90d0767 functions: - name: task-request dataTag: 12 mqType: 2 - name: task-response - dataTag: 14 + dataTag: 24 mqType: 2 - name: task-execute dataTag: 16 + mqType: 1 + - name: task-execute-log + dataTag: 26 + mqType: 2 + - name: task-log + dataTag: 28 mqType: 1 \ No newline at end of file diff --git a/go.mod b/go.mod index 0401740..363ccf6 100644 --- a/go.mod +++ b/go.mod @@ -4,10 +4,12 @@ go 1.19 require ( git.hpds.cc/Component/logging v0.0.0-20230106105738-e378e873921b - git.hpds.cc/pavement/hpds_node v0.0.0-20230402152619-41414aafa930 + git.hpds.cc/Component/network v0.0.0-20230405135741-a4ea724bab76 + git.hpds.cc/pavement/hpds_node v0.0.0-20230405153516-9403c4d01e12 github.com/go-sql-driver/mysql v1.7.0 github.com/hashicorp/consul/api v1.20.0 github.com/minio/minio-go v6.0.14+incompatible + github.com/minio/minio-go/v7 v7.0.52 github.com/spf13/cobra v1.6.1 github.com/spf13/viper v1.15.0 go.uber.org/zap v1.23.0 @@ -22,10 +24,10 @@ require ( cloud.google.com/go/firestore v1.9.0 // indirect cloud.google.com/go/longrunning v0.3.0 // indirect git.hpds.cc/Component/mq_coder v0.0.0-20221010064749-174ae7ae3340 // indirect - git.hpds.cc/Component/network v0.0.0-20221012021659-2433c68452d5 // indirect github.com/armon/go-metrics v0.4.0 // indirect github.com/coreos/go-semver v0.3.0 // indirect github.com/coreos/go-systemd/v22 v22.3.2 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect github.com/fatih/color v1.13.0 // indirect github.com/fsnotify/fsnotify v1.6.0 // indirect github.com/go-ini/ini v1.67.0 // indirect @@ -37,6 +39,8 @@ require ( github.com/golang/protobuf v1.5.2 // indirect github.com/golang/snappy v0.0.4 // indirect github.com/google/go-cmp v0.5.9 // indirect + github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 // indirect + github.com/google/uuid v1.3.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.2.1 // indirect github.com/googleapis/gax-go/v2 v2.7.0 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect @@ -48,22 +52,28 @@ require ( github.com/hashicorp/serf v0.10.1 // indirect github.com/inconshreveable/mousetrap v1.0.1 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/lucas-clemente/quic-go v0.29.1 // indirect + github.com/klauspost/compress v1.16.0 // indirect + github.com/klauspost/cpuid/v2 v2.2.4 // indirect github.com/magiconair/properties v1.8.7 // indirect - github.com/marten-seemann/qtls-go1-18 v0.1.2 // indirect - github.com/marten-seemann/qtls-go1-19 v0.1.0 // indirect github.com/matoous/go-nanoid/v2 v2.0.0 // indirect github.com/mattn/go-colorable v0.1.12 // indirect github.com/mattn/go-isatty v0.0.14 // indirect + github.com/minio/md5-simd v1.1.2 // indirect + github.com/minio/sha256-simd v1.0.0 // indirect github.com/mitchellh/go-homedir v1.1.0 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect - github.com/nxadm/tail v1.4.8 // indirect github.com/onsi/ginkgo v1.16.4 // indirect + github.com/onsi/ginkgo/v2 v2.2.0 // indirect github.com/pelletier/go-toml/v2 v2.0.6 // indirect github.com/pkg/errors v0.9.1 // indirect + github.com/quic-go/qtls-go1-19 v0.2.1 // indirect + github.com/quic-go/qtls-go1-20 v0.1.1 // indirect + github.com/quic-go/quic-go v0.33.0 // indirect + github.com/rs/xid v1.4.0 // indirect github.com/sagikazarmark/crypt v0.9.0 // indirect + github.com/sirupsen/logrus v1.9.0 // indirect github.com/spf13/afero v1.9.3 // indirect github.com/spf13/cast v1.5.0 // indirect github.com/spf13/jwalterweatherman v1.1.0 // indirect @@ -77,16 +87,16 @@ require ( go.opencensus.io v0.24.0 // indirect go.uber.org/atomic v1.9.0 // indirect go.uber.org/multierr v1.8.0 // indirect - golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e // indirect - golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect - golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect - golang.org/x/net v0.4.0 // indirect + golang.org/x/crypto v0.6.0 // indirect + golang.org/x/exp v0.0.0-20221205204356-47842c84f3db // indirect + golang.org/x/mod v0.6.0 // indirect + golang.org/x/net v0.7.0 // indirect golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783 // indirect golang.org/x/sync v0.1.0 // indirect - golang.org/x/sys v0.3.0 // indirect - golang.org/x/text v0.5.0 // indirect + golang.org/x/sys v0.5.0 // indirect + golang.org/x/text v0.7.0 // indirect golang.org/x/time v0.1.0 // indirect - golang.org/x/tools v0.1.12 // indirect + golang.org/x/tools v0.2.0 // indirect golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect google.golang.org/api v0.107.0 // indirect google.golang.org/appengine v1.6.7 // indirect @@ -95,6 +105,5 @@ require ( google.golang.org/protobuf v1.28.1 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect - gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect xorm.io/builder v0.3.11-0.20220531020008-1bd24a7dc978 // indirect ) diff --git a/internal/balance/hash.go b/internal/balance/hash.go index ef111ff..e06e8ed 100644 --- a/internal/balance/hash.go +++ b/internal/balance/hash.go @@ -30,16 +30,16 @@ func (s UInt32Slice) Swap(i, j int) { type ConsistentHashBalance struct { mux sync.RWMutex hash Hash - replicas int //复制因子 - keys UInt32Slice //已排序的节点hash切片 - hashMap map[uint32]int64 //节点哈希和key的map, 键是hash值，值是节点key + replicas int //复制因子 + keys UInt32Slice //已排序的节点hash切片 + hashMap map[uint32]*model.NodeLastStateItem //节点哈希和key的map, 键是hash值，值是节点key } func NewConsistentHashBalance(replicas int, fn Hash) *ConsistentHashBalance { m := &ConsistentHashBalance{ replicas: replicas, hash: fn, - hashMap: make(map[uint32]int64), + hashMap: make(map[uint32]*model.NodeLastStateItem), } if m.hash == nil { //最多32位，保证是一个2^32-1环 @@ -62,7 +62,7 @@ func (c *ConsistentHashBalance) Add(params model.NodeLastStateItem) error { for i := 0; i < c.replicas; i++ { hash := c.hash([]byte(strconv.Itoa(i) + fmt.Sprintf("%d", params.NodeId))) c.keys = append(c.keys, hash) - c.hashMap[hash] = params.NodeId + c.hashMap[hash] = ¶ms } // 对所有虚拟节点的哈希值进行排序，方便之后进行二分查找 @@ -71,9 +71,9 @@ func (c *ConsistentHashBalance) Add(params model.NodeLastStateItem) error { } // Get 方法根据给定的对象获取最靠近它的那个节点 -func (c *ConsistentHashBalance) Get(key int64) (int64, error) { +func (c *ConsistentHashBalance) Get(key int64) (*model.NodeLastStateItem, error) { if c.IsEmpty() { - return 0, errors.New("node is empty") + return nil, errors.New("node is empty") } hash := c.hash([]byte(fmt.Sprintf("%d", key))) diff --git a/internal/balance/index.go b/internal/balance/index.go index bfeb1e2..4843e94 100644 --- a/internal/balance/index.go +++ b/internal/balance/index.go @@ -4,5 +4,5 @@ import "hpds_control_center/model" type LoadBalance interface { Add(model.NodeLastStateItem) error - Get(int64) (int64, error) + Get(int64) (*model.NodeLastStateItem, error) } diff --git a/internal/balance/randomBalance.go b/internal/balance/randomBalance.go index 982baba..d93a6b8 100644 --- a/internal/balance/randomBalance.go +++ b/internal/balance/randomBalance.go @@ -9,24 +9,23 @@ import ( type RandomBalance struct { curIndex int - rss []int64 + rss []*model.NodeLastStateItem } func (r *RandomBalance) Add(params model.NodeLastStateItem) error { - nodeId := params.NodeId - r.rss = append(r.rss, nodeId) + r.rss = append(r.rss, ¶ms) return nil } -func (r *RandomBalance) Next() int64 { +func (r *RandomBalance) Next() *model.NodeLastStateItem { if len(r.rss) == 0 { - return 0 + return nil } r.curIndex = rand.Intn(len(r.rss)) return r.rss[r.curIndex] } -func (r *RandomBalance) Get(int64) (int64, error) { +func (r *RandomBalance) Get(int64) (*model.NodeLastStateItem, error) { return r.Next(), nil } diff --git a/internal/balance/roundRobinBalance.go b/internal/balance/roundRobinBalance.go index b9a63d8..d17f007 100644 --- a/internal/balance/roundRobinBalance.go +++ b/internal/balance/roundRobinBalance.go @@ -7,18 +7,17 @@ import ( // RoundRobinBalance 轮询负载均衡 type RoundRobinBalance struct { curIndex int - rss []int64 + rss []*model.NodeLastStateItem } func (r *RoundRobinBalance) Add(params model.NodeLastStateItem) error { - nodeId := params.NodeId - r.rss = append(r.rss, nodeId) + r.rss = append(r.rss, ¶ms) return nil } -func (r *RoundRobinBalance) Next() int64 { +func (r *RoundRobinBalance) Next() *model.NodeLastStateItem { if len(r.rss) == 0 { - return 0 + return nil } lens := len(r.rss) if r.curIndex >= lens { @@ -30,6 +29,6 @@ func (r *RoundRobinBalance) Next() int64 { return curNode } -func (r *RoundRobinBalance) Get(int64) (int64, error) { +func (r *RoundRobinBalance) Get(int64) (*model.NodeLastStateItem, error) { return r.Next(), nil } diff --git a/internal/balance/weightRoundRobinBalance.go b/internal/balance/weightRoundRobinBalance.go index 9831dec..c504caa 100644 --- a/internal/balance/weightRoundRobinBalance.go +++ b/internal/balance/weightRoundRobinBalance.go @@ -8,11 +8,11 @@ import ( type WeightRoundRobinBalance struct { curIndex int rss []*WeightNode - rsw []int + rsw []*model.NodeLastStateItem } type WeightNode struct { - node model.NodeLastStateItem + node *model.NodeLastStateItem Weight int //初始化时对节点约定的权重 currentWeight int //节点临时权重，每轮都会变化 effectiveWeight int //有效权重, 默认与weight相同 , totalWeight = sum(effectiveWeight) //出现故障就-1 @@ -33,7 +33,7 @@ func (r *WeightRoundRobinBalance) Add(params model.NodeLastStateItem) error { weightTask = 100 } node := &WeightNode{ - node: params, + node: ¶ms, Weight: int(400 - weightCpu - weightMem - weightSwap - weightTask), } node.effectiveWeight = node.Weight @@ -41,7 +41,7 @@ func (r *WeightRoundRobinBalance) Add(params model.NodeLastStateItem) error { return nil } -func (r *WeightRoundRobinBalance) Next() int64 { +func (r *WeightRoundRobinBalance) Next() *model.NodeLastStateItem { var best *WeightNode total := 0 for i := 0; i < len(r.rss); i++ { @@ -62,14 +62,14 @@ func (r *WeightRoundRobinBalance) Next() int64 { } if best == nil { - return 0 + return nil } //5 变更临时权重为 临时权重-有效权重之和 best.currentWeight -= total - return best.node.NodeId + return best.node } -func (r *WeightRoundRobinBalance) Get(int64) (int64, error) { +func (r *WeightRoundRobinBalance) Get(int64) (*model.NodeLastStateItem, error) { return r.Next(), nil } diff --git a/internal/minio/index.go b/internal/minio/index.go index e6b2be4..b407b6f 100644 --- a/internal/minio/index.go +++ b/internal/minio/index.go @@ -1,8 +1,11 @@ package minio import ( + "context" "git.hpds.cc/Component/logging" - "github.com/minio/minio-go" + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" + "io" ) type MinClient struct { @@ -11,7 +14,11 @@ type MinClient struct { } func NewClient(ak, sak, ep string, useSSL bool, logger *logging.Logger) *MinClient { - client, err := minio.New(ep, ak, sak, useSSL) + opt := &minio.Options{ + Creds: credentials.NewStaticV4(ak, sak, ""), + Secure: useSSL, + } + client, err := minio.New(ep, opt) if err != nil { return nil } @@ -22,9 +29,18 @@ func NewClient(ak, sak, ep string, useSSL bool, logger *logging.Logger) *MinClie } func (cli *MinClient) UploadObject(fn, dst, bucket string) error { - _, err := cli.Client.FPutObject(bucket, dst, fn, minio.PutObjectOptions{}) + _, err := cli.Client.FPutObject(context.Background(), bucket, dst, fn, minio.PutObjectOptions{}) if err != nil { return err } return nil } + +func (cli *MinClient) GetObject(dstUrl, bucket string) ([]byte, error) { + f, err := cli.Client.GetObject(context.Background(), bucket, dstUrl, minio.GetObjectOptions{}) + if err != nil { + return nil, err + } + imgByte, _ := io.ReadAll(f) + return imgByte, nil +} diff --git a/internal/proto/proto.go b/internal/proto/proto.go new file mode 100644 index 0000000..662bf91 --- /dev/null +++ b/internal/proto/proto.go @@ -0,0 +1,20 @@ +package proto + +type TaskLogPayload struct { + PayloadType int `json:"payloadType"` + TaskId int64 `json:"taskId"` + TaskCode string `json:"taskCode"` + NodeId int64 `json:"nodeId"` + NodeGuid string `json:"nodeGuid"` + TaskContent string `json:"taskContent"` + Status int `json:"status"` //1：执行成功;2:执行失败 + EventTime int64 `json:"eventTime"` +} +type TaskLogProgress struct { + PayloadType int `json:"payloadType"` + TaskId int64 `json:"taskId"` + TotalCount int64 `json:"totalCount"` + CompletedCount int64 `json:"completedCount"` + FailingCount int64 `json:"failingCount"` + UnfinishedCount int64 `json:"unfinishedCount"` +} diff --git a/model/TaskLog.go b/model/TaskLog.go new file mode 100644 index 0000000..fab94c7 --- /dev/null +++ b/model/TaskLog.go @@ -0,0 +1,14 @@ +package model + +type TaskLog struct { + TaskLogId int64 `xorm:"not null pk autoincr BIGINT(11)" json:"taskLogId"` + TaskId int64 `xorm:"INT(11) index" json:"taskId"` + NodeId int64 `xorm:"INT(11) index" json:"nodeId"` + Content string `xorm:"LANGTEXT" json:"content"` + CreateAt int64 `xorm:"created" json:"createAt"` + UpdateAt int64 `xorm:"updated" json:"updateAt"` +} + +func InsertLog(taskLog *TaskLog) { + _, _ = DB.Insert(taskLog) +} diff --git a/model/dataset.go b/model/dataset.go new file mode 100644 index 0000000..b032cb2 --- /dev/null +++ b/model/dataset.go @@ -0,0 +1,15 @@ +package model + +type Dataset struct { + DatasetId int64 `xorm:"not null pk autoincr INT(11)" json:"datasetId"` + DatasetName string `xorm:"varchar(200) not null" json:"datasetName"` + DatasetDesc string `xorm:"varchar(200)" json:"datasetDesc"` + StoreName string `xorm:"varchar(200)" json:"storeName"` + CategoryId int `xorm:"not null SMALLINT default 1" json:"categoryId"` //业务分类, 1:道路 2:桥梁 3:隧道 4:边坡 + ProjectId int64 `xorm:"INT(11) not null default 0 index" json:"projectId"` + OwnerId int64 `xorm:"INT(11) not null default 0 index" json:"ownerId"` + Creator int64 `xorm:"INT(11) not null default 0 index" json:"creator"` + Status int `xorm:"SMALLINT not null default 1" json:"status"` + CreateAt int64 `xorm:"created" json:"createAt"` + UpdateAt int64 `xorm:"updated" json:"updateAt"` +} diff --git a/model/file.go b/model/file.go new file mode 100644 index 0000000..c6034bc --- /dev/null +++ b/model/file.go @@ -0,0 +1,15 @@ +package model + +type FileManager struct { + FileId int64 `xorm:"not null pk autoincr INT(11)" json:"fileId"` //文件编号 + FileName string `xorm:"VARCHAR(200)" json:"fileName"` //文件名 + AccessUrl string `xorm:"VARCHAR(400)" json:"url"` //访问路径 + Scene string `xorm:"VARCHAR(40)" json:"scene"` //应用场景，0 : 其他, 1: 道路; 2: 桥梁; 3:隧道; 4: 边坡 + DataType int `xorm:"TINYINT index default 0" json:"dataType"` //数据类型，1:数据集;2:病害库;3:应用;0:其他 + DatasetId int64 `xorm:"INT(11) index default 0" json:"datasetId"` //数据集 + FileSize int64 `xorm:"BIGINT" json:"fileSize"` //文件大小 + FileMd5 string `xorm:"VARCHAR(64)" json:"fileMd5"` //文件MD5 + Creator int64 `xorm:"INT(11) index" json:"creator"` //上传人 + CreateAt int64 `xorm:"created" json:"createAt"` //上传时间 + UpdateAt int64 `xorm:"updated" json:"updateAt"` //更新时间 +} diff --git a/model/index.go b/model/index.go index 4e5c9dc..d20c6c3 100644 --- a/model/index.go +++ b/model/index.go @@ -24,6 +24,7 @@ func New(driveName, dsn string, showSql bool) { &Node{}, &NodeState{}, &Task{}, + &TaskLog{}, &TaskResult{}, ) if err != nil { diff --git a/model/task.go b/model/task.go index 11b7860..50d7950 100644 --- a/model/task.go +++ b/model/task.go @@ -1,5 +1,10 @@ package model +import ( + "hpds_control_center/internal/proto" + "time" +) + type Task struct { TaskId int64 `xorm:"not null pk autoincr INT(11)" json:"taskId"` ModelId int64 `xorm:"INT(11) index" json:"modelId"` @@ -11,6 +16,10 @@ type Task struct { AppointmentTime string `xorm:"VARCHAR(30)" json:"appointmentTime"` StartTime int64 `xorm:"BIGINT" json:"startTime"` FinishTime int64 `xorm:"BIGINT" json:"finishTime"` + TotalCount int64 `xorm:"INT" json:"totalCount"` + FailingCount int64 `xorm:"INT" json:"failingCount"` + CompletedCount int64 `xorm:"INT" json:"completedCount"` + UnfinishedCount int64 `xorm:"INT" json:"unfinishedCount"` Status int `xorm:"not null SMALLINT default 0" json:"status"` // 1:等待执行; 2:执行中; 3:执行完成; 4:任务分配失败; 5:任务执行失败 CreateAt int64 `xorm:"created" json:"createAt"` UpdateAt int64 `xorm:"updated" json:"updateAt"` @@ -22,3 +31,54 @@ func UpdateTaskExecuteNode(id, nodeId int64) { item.NodeId = nodeId _, _ = DB.ID(id).Cols("node_id").Update(item) } + +func UpdateTaskProgress(taskProgress *proto.TaskLogProgress) { + task := new(Task) + h, err := DB.ID(taskProgress.TaskId).Get(task) + if err != nil || !h { + return + } + if taskProgress.CompletedCount > task.CompletedCount { + task.CompletedCount = taskProgress.CompletedCount + } + if taskProgress.FailingCount > task.FailingCount { + task.FailingCount = taskProgress.FailingCount + } + if taskProgress.TotalCount != task.TotalCount && taskProgress.TotalCount > 0 { + task.TotalCount = taskProgress.TotalCount + } + if taskProgress.UnfinishedCount != task.UnfinishedCount && taskProgress.UnfinishedCount >= 0 { + task.UnfinishedCount = taskProgress.UnfinishedCount + } + if task.CompletedCount+task.FailingCount >= task.TotalCount { + task.Status = 3 + } else { + task.Status = 2 + } + task.UpdateAt = time.Now().Unix() + _, err = DB.ID(taskProgress.TaskId).Cols("completed_count", "failing_count", "total_count", "unfinished_count", "update_at").Update(task) + if err != nil { + return + } +} + +func UpdateTaskProgressByLog(res *TaskResult) float64 { + item := new(Task) + h, err := DB.ID(res.TaskId).Get(item) + if err != nil || !h { + return -1 + } + item.CompletedCount += 1 + item.UnfinishedCount -= 1 + if item.CompletedCount <= item.TotalCount { + item.FinishTime = time.Now().Unix() + item.UnfinishedCount = 0 + item.Status = 3 + } + item.UpdateAt = time.Now().Unix() + _, _ = DB.ID(res.TaskId).Cols("completed_count", "total_count", "unfinished_count", "update_at").Update(item) + if item.TotalCount > 0 { + return float64(item.UnfinishedCount) / float64(item.TotalCount) + } + return -1 +} diff --git a/model/taskResult.go b/model/taskResult.go index b055467..91dd03f 100644 --- a/model/taskResult.go +++ b/model/taskResult.go @@ -3,6 +3,7 @@ package model type TaskResult struct { ResultId int64 `xorm:"not null pk autoincr INT(11)" json:"resultId"` TaskId int64 `xorm:"INT(11) index" json:"taskId"` + FileId int64 `xorm:"INT(11) index" json:"fileId"` TaskCode string `xorm:"varchar(200)" json:"taskCode"` ModelId int64 `xorm:"INT(11)" json:"modelId"` NodeId int64 `xorm:"INT(11)" json:"nodeId"` @@ -11,5 +12,5 @@ type TaskResult struct { SubDataset string `xorm:"varchar(200)" json:"subDataset"` DatasetId int64 `xorm:"INT(11)" json:"datasetId"` SrcPath string `xorm:"varchar(500)" json:"srcPath"` - Result string `xorm:"TEXT" json:"result"` + Result string `xorm:"LONGTEXT" json:"result"` } diff --git a/mq/index.go b/mq/index.go index 041e477..ce05661 100644 --- a/mq/index.go +++ b/mq/index.go @@ -1,20 +1,31 @@ package mq import ( + "encoding/base64" "encoding/json" "fmt" "git.hpds.cc/Component/logging" + "git.hpds.cc/Component/network/frame" + "github.com/google/uuid" "go.uber.org/zap" "hpds_control_center/config" "hpds_control_center/internal/balance" + "hpds_control_center/internal/minio" + "hpds_control_center/internal/proto" "hpds_control_center/model" "os" + "strconv" + "strings" + "sync" "time" "git.hpds.cc/pavement/hpds_node" ) -var MqList []HpdsMqNode +var ( + MqList []HpdsMqNode + TaskList = make(map[int64]*TaskItem) +) type HpdsMqNode struct { MqType uint @@ -24,6 +35,15 @@ type HpdsMqNode struct { Logger *logging.Logger } +type TaskItem struct { + TaskId int64 + TotalCount int64 + CompletedCount int64 + FailingCount int64 + UnfinishedCount int64 + LastSendTime int64 +} + func must(logger *logging.Logger, err error) { if err != nil { if logger != nil { @@ -43,7 +63,7 @@ func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *loggin sf := hpds_node.NewStreamFunction( v.Name, hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", node.Host, node.Port)), - hpds_node.WithObserveDataTags(v.DataTag), + hpds_node.WithObserveDataTags(frame.Tag(v.DataTag)), hpds_node.WithCredential(node.Token), ) err = sf.Connect() @@ -57,6 +77,10 @@ func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *loggin switch v.Name { case "task-request": _ = sf.SetHandler(TaskRequestHandler) + case "task-response": + _ = sf.SetHandler(TaskResponseHandler) + case "task-execute-log": + _ = sf.SetHandler(TaskExecuteLogHandler) default: } @@ -75,7 +99,7 @@ func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *loggin EndPoint: ap, } must(logger, err) - ap.SetDataTag(v.DataTag) + ap.SetDataTag(frame.Tag(v.DataTag)) mqList = append(mqList, nodeInfo) } @@ -101,8 +125,7 @@ func GenerateAndSendData(stream hpds_node.AccessPoint, data []byte) error { return nil } -func TaskRequestHandler(data []byte) (byte, []byte) { - fmt.Println("接收数据", string(data)) +func TaskRequestHandler(data []byte) (frame.Tag, []byte) { cmd := new(InstructionReq) err := json.Unmarshal(data, cmd) if err != nil { @@ -111,70 +134,144 @@ func TaskRequestHandler(data []byte) (byte, []byte) { switch cmd.Command { case TaskAdd: payload := cmd.Payload.(map[string]interface{}) - if payload["nodeId"].(float64) == 0 { - //根据业务属性进行分配节点 - m := model.GetModelById(int64(payload["modelId"].(float64))) - var nodeList []model.Node - //todo 需要增加模型下发记录 - if m.IsLightWeight { - nodeList = model.GetLightWeight(m.ModelId) - } else { - nodeList = model.GetAllNode(m.ModelId) - } - if nodeList != nil { - if len(nodeList) > 1 { - //这里采用加权算法，权重采用CPU占用+mem使用+任务执行状态 - list := model.GetNodeState(nodeList) - lb := balance.LoadBalanceFactory(balance.LbWeightRoundRobin) - for _, v := range list { - _ = lb.Add(v) - } - nodeId, _ := lb.Get(0) - - payload["nodeId"] = nodeId - cmd := &InstructionReq{ - Command: TaskExecute, - Payload: payload, - } - - pData, _ := json.Marshal(cmd) - cli := GetMqClient("task-execute", 1) - if cli != nil { - _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) - } - model.UpdateTaskExecuteNode(int64(payload["taskId"].(float64)), nodeId) + if len(payload["subDataset"].(string)) > 0 { + if payload["nodeId"].(float64) == 0 { + //根据业务属性进行分配节点 + m := model.GetModelById(int64(payload["modelId"].(float64))) + var nodeList []model.Node + //todo 需要增加模型下发记录 + if m.IsLightWeight { + nodeList = model.GetLightWeight(m.ModelId) } else { - payload["nodeId"] = nodeList[0].NodeId - issue := new(model.IssueModel) - h, _ := model.DB.Where("model_id=? and node_id =?", int64(payload["modelId"].(float64)), nodeList[0].NodeId).Get(issue) - if !h { - - } - payload["issueResult"] = issue.IssueResult - cmd := &InstructionReq{ - Command: TaskExecute, - Payload: payload, - } - pData, _ := json.Marshal(cmd) - cli := GetMqClient("task-execute", 1) - if cli != nil { - _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) - } - model.UpdateTaskExecuteNode(int64(payload["taskId"].(float64)), nodeList[0].NodeId) + nodeList = model.GetAllNode(m.ModelId) } + if nodeList != nil { + if len(nodeList) > 1 { + //这里采用加权算法，权重采用CPU占用+mem使用+任务执行状态 + list := model.GetNodeState(nodeList) + lb := balance.LoadBalanceFactory(balance.LbWeightRoundRobin) + for _, v := range list { + _ = lb.Add(v) + } + nodeId, _ := lb.Get(0) + if nodeId == nil { + //todo 需要增加未能获取的处理 + } + payload["nodeId"] = nodeId.NodeId + payload["nodeGuid"] = nodeId.NodeGuid + + cmd := &InstructionReq{ + Command: TaskExecute, + Payload: payload, + } + + pData, _ := json.Marshal(cmd) + cli := GetMqClient("task-execute", 1) + if cli != nil { + _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) + } + model.UpdateTaskExecuteNode(int64(payload["taskId"].(float64)), nodeId.NodeId) + } else { + payload["nodeId"] = nodeList[0].NodeId + issue := new(model.IssueModel) + h, _ := model.DB.Where("model_id=? and node_id =?", int64(payload["modelId"].(float64)), nodeList[0].NodeId).Get(issue) + if !h { + + } + payload["issueResult"] = issue.IssueResult + cmd := &InstructionReq{ + Command: TaskExecute, + Payload: payload, + } + pData, _ := json.Marshal(cmd) + cli := GetMqClient("task-execute", 1) + if cli != nil { + _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) + } + model.UpdateTaskExecuteNode(int64(payload["taskId"].(float64)), nodeList[0].NodeId) + } + } else { + + } + } else { - + cmd := &InstructionReq{ + Command: TaskExecute, + Payload: payload, + } + pData, _ := json.Marshal(cmd) + cli := GetMqClient("task-execute", 1) + if cli != nil { + _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) + } } - } else { - cmd := &InstructionReq{ - Command: TaskExecute, - Payload: payload, - } - pData, _ := json.Marshal(cmd) - cli := GetMqClient("task-execute", 1) - if cli != nil { - _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) + if len(payload["datasetArr"].(string)) > 0 { + GoroutinueChan := make(chan bool, 5) + datasetArr := strings.Split(payload["datasetArr"].(string), ",") + //for _, val := range datasetArr { + // dId, err := strconv.ParseInt(val, 10, 64) + // if err != nil { + // continue + // } + // dt := new(model.Dataset) + // _, _ = model.DB.ID(dId).Get(dt) + fileList := make([]model.FileManager, 0) + _ = model.DB.In("dataset_id", datasetArr). + Find(&fileList) + item := &TaskItem{ + TaskId: int64(payload["taskId"].(float64)), + TotalCount: int64(len(fileList)), + CompletedCount: 0, + FailingCount: 0, + UnfinishedCount: int64(len(fileList)), + LastSendTime: time.Now().Unix(), + } + TaskList[int64(payload["taskId"].(float64))] = item + //获取任务总数，并入库 + taskProgress := &proto.TaskLogProgress{ + PayloadType: 1, + TaskId: int64(payload["taskId"].(float64)), + TotalCount: int64(len(fileList)), + CompletedCount: 0, + FailingCount: 0, + UnfinishedCount: int64(len(fileList)), + } + model.UpdateTaskProgress(taskProgress) + //taskProgressCmd := &InstructionReq{ + // Command: TaskLog, + // Payload: taskProgress, + //} + //deliver("task-log", 1, taskProgressCmd) + + //数据集处理 + minioCli := minio.NewClient(config.Cfg.Minio.AccessKeyId, config.Cfg.Minio.SecretAccessKey, config.Cfg.Minio.Endpoint, false, logging.L()) + for _, v := range fileList { + GoroutinueChan <- true + go func(fa model.FileManager) { + dstPath := strings.Replace(fa.AccessUrl, fmt.Sprintf("%s://%s/", config.Cfg.Minio.Protocol, config.Cfg.Minio.Endpoint), "", 1) + + dstPath = strings.Replace(dstPath, config.Cfg.Minio.Bucket, "", 1) + imgByte, _ := minioCli.GetObject(dstPath, config.Cfg.Minio.Bucket) + f := FileCapture{ + FileId: fa.FileId, + FileName: fa.FileName, + File: base64.StdEncoding.EncodeToString(imgByte), + DatasetName: payload["datasetName"].(string), + CaptureTime: fa.CreateAt, + } + payload["single"] = f + taskCode, _ := uuid.NewUUID() + payload["taskCode"] = taskCode.String() + cmd := &InstructionReq{ + Command: TaskExecute, + Payload: payload, + } + deliver("task-execute", 1, cmd) + <-GoroutinueChan + }(v) + } + //} } } @@ -209,25 +306,153 @@ func TaskRequestHandler(data []byte) (byte, []byte) { item.UpdateAt = time.Now().Unix() _, _ = model.DB.Insert(item) } - case TaskResponse: - payload := cmd.Payload.(map[string]interface{}) - item := new(model.TaskResult) - item.TaskId = int64(payload["taskId"].(float64)) - item.TaskCode = payload["taskCode"].(string) - item.NodeId = int64(payload["nodeId"].(float64)) - item.ModelId = int64(payload["modelId"].(float64)) - item.StartTime = int64(payload["startTime"].(float64)) - item.FinishTime = int64(payload["finishTime"].(float64)) - item.SubDataset = payload["subDataset"].(string) - item.DatasetId = int64(payload["datasetArr"].(float64)) - item.SrcPath = payload["srcPath"].(string) - item.Result = payload["body"].(string) - _, _ = model.DB.Insert(item) - //fn := payload["fileName"].(string) - //dec := base64.NewDecoder(base64.StdEncoding, strings.NewReader(payload["file"].(string))) + //case TaskResponse: + // payload := cmd.Payload.(map[string]interface{}) + // item := new(model.TaskResult) + // item.TaskId = int64(payload["taskId"].(float64)) + // item.TaskCode = payload["taskCode"].(string) + // item.NodeId = int64(payload["nodeId"].(float64)) + // item.ModelId = int64(payload["modelId"].(float64)) + // item.StartTime = int64(payload["startTime"].(float64)) + // item.FinishTime = int64(payload["finishTime"].(float64)) + // item.SubDataset = payload["subDataset"].(string) + // item.DatasetId = int64(payload["datasetArr"].(float64)) + // item.SrcPath = payload["srcPath"].(string) + // item.Result = payload["body"].(string) + // _, _ = model.DB.Insert(item) + // //fn := payload["fileName"].(string) + // //dec := base64.NewDecoder(base64.StdEncoding, strings.NewReader(payload["file"].(string))) default: } - return byte(cmd.Command), nil + return frame.Tag(cmd.Command), nil +} + +func TaskResponseHandler(data []byte) (frame.Tag, []byte) { + cmd := new(InstructionReq) + err := json.Unmarshal(data, cmd) + if err != nil { + return 0x0B, []byte(err.Error()) + } + switch cmd.Command { + case TaskResponse: + payload := cmd.Payload.(map[string]interface{}) + item := new(model.TaskResult) + item.TaskId = int64(payload["taskId"].(float64)) + if _, ok := payload["taskCode"]; ok && payload["taskCode"] != nil { + item.TaskCode = payload["taskCode"].(string) + } + if _, ok := payload["fileId"]; ok { + item.FileId = int64(payload["fileId"].(float64)) + } + item.NodeId = int64(payload["nodeId"].(float64)) + item.ModelId = int64(payload["modelId"].(float64)) + item.StartTime = int64(payload["startTime"].(float64)) + item.FinishTime = int64(payload["finishTime"].(float64)) + if _, ok := payload["subDataset"]; ok { + item.SubDataset = payload["subDataset"].(string) + } + item.DatasetId, _ = strconv.ParseInt(payload["datasetArr"].(string), 10, 64) + if _, ok := payload["srcPath"]; ok && payload["srcPath"] != nil { + item.SrcPath = payload["srcPath"].(string) + } + item.Result = payload["body"].(string) + _, err = model.DB.Insert(item) + if err != nil { + fmt.Println("接收TaskResponse数据出错", err) + } + //更新运行进度 + rat := model.UpdateTaskProgressByLog(item) + var ( + ratStr string + ) + if rat > 0 && rat < 1 { + ratStr = fmt.Sprintf("[已处理%2.f,剩余%2.f未处理]", 1-rat, rat) + } else if rat == 1 { + ratStr = "[已全部处理]" + } + taskLog := new(model.TaskLog) + taskLog.TaskId = item.TaskId + taskLog.NodeId = item.NodeId + if len(item.SrcPath) > 0 { + taskLog.Content = fmt.Sprintf("[%s] %s 图片%s处理完成", time.Unix(item.FinishTime, 0).Format("2006-01-02 15:04:05"), + ratStr, item.SrcPath) + } else { + taskLog.Content = fmt.Sprintf("[%s] %s", time.Unix(item.FinishTime, 0).Format("2006-01-02 15:04:05"), + ratStr) + } + model.InsertLog(taskLog) + //fn := payload["fileName"].(string) + //dec := base64.NewDecoder(base64.StdEncoding, strings.NewReader(payload["file"].(string))) + + default: + } + return frame.Tag(cmd.Command), nil +} + +func deliver(topic string, mqType uint, payload interface{}) { + cli := GetMqClient(topic, mqType) + pData, _ := json.Marshal(payload) + _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) +} + +func TaskExecuteLogHandler(data []byte) (frame.Tag, []byte) { + cmd := new(InstructionReq) + err := json.Unmarshal(data, cmd) + if err != nil { + return 0x0B, []byte(err.Error()) + } + payload := cmd.Payload.(map[string]interface{}) + var l sync.Mutex + l.Lock() + taskId := int64(payload["taskId"].(float64)) + if item, ok := TaskList[taskId]; ok { + item.UnfinishedCount -= 1 + if int(payload["status"].(float64)) == 1 { + item.CompletedCount += 1 + } + if int(payload["status"].(float64)) == 2 { + item.FailingCount += 1 + } + if item.UnfinishedCount <= 0 || time.Now().Unix()-item.LastSendTime > 5000 { + //发送完成消息 + taskProgress := &proto.TaskLogProgress{ + PayloadType: 1, + TaskId: item.TaskId, + TotalCount: item.TotalCount, + CompletedCount: item.CompletedCount, + FailingCount: item.FailingCount, + UnfinishedCount: item.UnfinishedCount, + } + //model.UpdateTaskProgress(taskProgress) + taskProgressCmd := &InstructionReq{ + Command: TaskLog, + Payload: taskProgress, + } + deliver("task-log", 1, taskProgressCmd) + if item.UnfinishedCount <= 0 { + delete(TaskList, item.TaskId) + } else { + item.LastSendTime = time.Now().Unix() + } + } + taskLog := &proto.TaskLogPayload{ + PayloadType: 2, + TaskId: item.TaskId, + TaskCode: payload["taskCode"].(string), + NodeId: int64(payload["nodeId"].(float64)), + NodeGuid: payload["nodeGuid"].(string), + TaskContent: payload["taskContent"].(string), + Status: int(payload["status"].(float64)), + EventTime: int64(payload["eventTime"].(float64)), + } + taskLogCmd := &InstructionReq{ + Command: TaskLog, + Payload: taskLog, + } + deliver("task-log", 1, taskLogCmd) + } + l.Unlock() + return frame.Tag(cmd.Command), nil } diff --git a/mq/instruction.go b/mq/instruction.go index abdad42..bdf5a30 100644 --- a/mq/instruction.go +++ b/mq/instruction.go @@ -7,9 +7,36 @@ const ( TaskResponse ModelIssueRepeater ModelIssueResponse + TaskExecuteLog + TaskLog ) type InstructionReq struct { Command int `json:"command"` Payload interface{} `json:"payload"` } + +type FileCapture struct { + FileId int64 `json:"fileId"` + FileName string `json:"fileName"` + File string `json:"file"` + DatasetName string `json:"datasetName"` + CaptureTime int64 `json:"captureTime"` +} + +type TaskLogPayload struct { + TaskId int64 `json:"taskId"` + TaskCode string `json:"taskCode"` + NodeId int64 `json:"nodeId"` + NodeGuid string `json:"nodeGuid"` + Payload string `json:"taskContent"` + EventTime int64 `json:"eventTime"` +} + +type TaskLogProgress struct { + TaskId int64 `json:"taskId"` + TotalCount int64 `json:"totalCount"` + CompletedCount int64 `json:"completedCount"` + FailingCount int64 `json:"failingCount"` + UnfinishedCount int64 `json:"unfinishedCount"` +} diff --git a/test/task_test.go b/test/task_test.go new file mode 100644 index 0000000..a299d99 --- /dev/null +++ b/test/task_test.go @@ -0,0 +1,11 @@ +package test + +// {\"command\":1,\"payload\":{\"datasetArr\":\"2\",\"d":\"绕城道路测试\",\"datasetPath\":\"test_road\",\"httpUrl\":\"/\",\"inPath\":\"/in\",\"modelCommand\":\"\",\"modelId\":3,\"modelVersion\":\"v2.0\",\"nodeId\":1,\"h\":\"/out\",\"subDataTag\":100,\"subDataset\":\"\",\"taskId\":29,\"workflow\":\"\"}} +// +//func TestTaskRequestHandler(t *testing.T) { +// data := []byte(`{"command":1,"payload":{"datasetArr":"2","d":\"绕城道路测试\",\"datasetPath\":\"test_road\",\"httpUrl\":\"/\",\"inPath\":\"/in\",\"modelCommand\":\"\",\"modelId\":3,\"modelVersion\":\"v2.0\",\"nodeId\":1,\"h\":\"/out\",\"subDataTag\":100,\"subDataset\":\"\",\"taskId\":29,\"workflow\":\"\"}}`) +// tag, res := mq.TaskRequestHandler(data) +// fmt.Println("tag ==== >>>> ", tag) +// fmt.Println("res===== >>>> ", string(res)) +// assert.Nil(t, res) +//}