From 46d29e280b0c8e1ccff34a4e02bf553aeb4076e3 Mon Sep 17 00:00:00 2001 From: wangjian Date: Thu, 23 Mar 2023 15:58:12 +0800 Subject: [PATCH] =?UTF-8?q?1=E3=80=81=E5=88=9D=E5=A7=8B=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 1 + cmd/server.go | 154 ++++++++++++++++++ config/config-dev.yaml | 43 +++++ config/config.go | 137 ++++++++++++++++ config/config.yaml | 48 ++++++ go.mod | 102 ++++++++++++ internal/balance/balanceFactory.go | 25 +++ internal/balance/hash.go | 90 +++++++++++ internal/balance/index.go | 8 + internal/balance/randomBalance.go | 32 ++++ internal/balance/roundRobinBalance.go | 35 +++++ internal/balance/weightRoundRobinBalance.go | 78 ++++++++++ internal/minio/index.go | 30 ++++ main.go | 27 ++++ model/index.go | 42 +++++ model/model.go | 28 ++++ model/modelIssue.go | 11 ++ model/node.go | 46 ++++++ model/nodeState.go | 60 +++++++ model/task.go | 24 +++ model/taskResult.go | 15 ++ mq/index.go | 233 ++++++++++++++++++++++++++++ mq/instruction.go | 15 ++ pkg/discover/consul/consul.go | 87 +++++++++++ 24 files changed, 1371 insertions(+) create mode 100644 README.md create mode 100644 cmd/server.go create mode 100644 config/config-dev.yaml create mode 100644 config/config.go create mode 100644 config/config.yaml create mode 100644 go.mod create mode 100644 internal/balance/balanceFactory.go create mode 100644 internal/balance/hash.go create mode 100644 internal/balance/index.go create mode 100644 internal/balance/randomBalance.go create mode 100644 internal/balance/roundRobinBalance.go create mode 100644 internal/balance/weightRoundRobinBalance.go create mode 100644 internal/minio/index.go create mode 100644 main.go create mode 100644 model/index.go create mode 100644 model/model.go create mode 100644 model/modelIssue.go create mode 100644 model/node.go create mode 100644 model/nodeState.go create mode 100644 model/task.go create mode 100644 model/taskResult.go create mode 100644 mq/index.go create mode 100644 mq/instruction.go create mode 100644 pkg/discover/consul/consul.go diff --git a/README.md b/README.md new file mode 100644 index 0000000..2bf7e95 --- /dev/null +++ b/README.md @@ -0,0 +1 @@ +## 任务调度中心 \ No newline at end of file diff --git a/cmd/server.go b/cmd/server.go new file mode 100644 index 0000000..82d7bec --- /dev/null +++ b/cmd/server.go @@ -0,0 +1,154 @@ +package cmd + +import ( + "context" + "fmt" + "hpds_control_center/model" + "hpds_control_center/mq" + "os" + "os/signal" + "syscall" + + "git.hpds.cc/Component/logging" + "github.com/spf13/cobra" + "go.uber.org/zap" + "hpds_control_center/config" + discover "hpds_control_center/pkg/discover/consul" +) + +var ( + ConfigFileFlag string = "./config/config.yaml" + ConsulAddress string = "http://localhost:8500" + NodeName string = "main-node" + Mode string = "dev" +) + +func must(err error) { + if err != nil { + _, _ = fmt.Fprint(os.Stderr, err) + os.Exit(1) + } +} + +func NewStartCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "start", + Short: "Start hpds_web application", + Run: func(cmd *cobra.Command, args []string) { + var ( + cfg *config.WebConfig + err error + ) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + must(err) + configFileFlag, err := cmd.Flags().GetString("c") + if err != nil { + fmt.Println("get local config err: ", err) + return + } + ConsulAddress, err = cmd.Flags().GetString("r") + if err != nil { + fmt.Println("get remote config err: ", err) + return + } + NodeName, err = cmd.Flags().GetString("n") + if err != nil { + fmt.Println("get remote path config err: ", err) + return + } + Mode, err = cmd.Flags().GetString("m") + if err != nil { + fmt.Println("get remote path config err: ", err) + return + } + if len(configFileFlag) > 1 { + cfg, err = config.ParseConfigByFile(configFileFlag) + must(err) + err = config.UpdateRemoteConfig(cfg) + must(err) + ConfigFileFlag = configFileFlag + } else { + //获取consul注册中心的配置文件 + cfg, err = config.GetRemoteConfig(ConsulAddress, fmt.Sprintf("hpds-pavement/hpds_control/%s/%s", Mode, NodeName)) + must(err) + err = config.UpdateLocalConfig(cfg, ConfigFileFlag) + } + //创建注册对象 + tags := make([]string, 1) + tags[0] = "control" + consulCfg, err := discover.NewConsulConfig(cfg.Consul.Host, cfg.Name, cfg.Name, cfg.Consul.Host, cfg.Consul.Port, + tags, 300, 300, 300) + must(err) + + //连接数据库 + model.New(cfg.Db.DriveName, cfg.Db.Conn, cfg.Mode == "dev") + + logger := LoadLoggerConfig(cfg.Logging) + + //创建消息连接点 + mq.MqList, err = mq.NewMqClient(cfg.Funcs, cfg.Node, logger) + must(err) + + // 退出channel + exitChannel := make(chan os.Signal) + defer close(exitChannel) + + // 退出信号监听 + go func(c chan os.Signal) { + _ = consulCfg.ServiceDeregister() + signal.Notify(c, syscall.SIGINT, syscall.SIGTERM) + }(exitChannel) + // start http service + go func() { + //fmt.Printf("Http Server start at port %d \n", cfg.Port) + //启动前执行注册 + err = consulCfg.ServiceRegister() + must(err) + + }() + //服务退出取消注册 + //err = consulCfg.ServiceDeregister() + // + //must(err) + //zap.L().Error("发生错误", zap.Error(err)) + for { + select { + case <-ctx.Done(): + _ = consulCfg.ServiceDeregister() + logger.With( + zap.String("web", "exit"), + ).Error(ctx.Err().Error()) + return + case errs := <-exitChannel: + _ = consulCfg.ServiceDeregister() + logger.With( + zap.String("web", "服务退出"), + ).Info(errs.String()) + return + } + } + }, + } + cmd.Flags().StringVar(&ConfigFileFlag, "c", "./config/config.yaml", "The configuration file path") + cmd.Flags().StringVar(&ConsulAddress, "r", "http://consul.hpds.cc", "The configuration remote consul address") + cmd.Flags().StringVar(&NodeName, "n", "main-node", "The configuration name") + cmd.Flags().StringVar(&Mode, "m", "dev", "run mode : dev | test | releases") + return cmd +} + +func LoadLoggerConfig(opt config.LogOptions) *logging.Logger { + return logging.NewLogger( + logging.SetPath(opt.Path), + logging.SetPrefix(opt.Prefix), + logging.SetDevelopment(opt.Development), + logging.SetDebugFileSuffix(opt.DebugFileSuffix), + logging.SetWarnFileSuffix(opt.WarnFileSuffix), + logging.SetErrorFileSuffix(opt.ErrorFileSuffix), + logging.SetInfoFileSuffix(opt.InfoFileSuffix), + logging.SetMaxAge(opt.MaxAge), + logging.SetMaxBackups(opt.MaxBackups), + logging.SetMaxSize(opt.MaxSize), + logging.SetLevel(logging.LogLevel["debug"]), + ) +} diff --git a/config/config-dev.yaml b/config/config-dev.yaml new file mode 100644 index 0000000..b356215 --- /dev/null +++ b/config/config-dev.yaml @@ -0,0 +1,43 @@ +name: control_center +host: 0.0.0.0 +port: 8088 +mode: dev +logging: + path: ./logs + prefix: hpds-control + errorFileSuffix: error.log + warnFileSuffix: warn.log + infoFileSuffix: info.log + debugFileSuffix: debug.log + maxSize: 100 + maxBackups: 3000 + maxAge: 30 + development: true +consul: + host: http://consul.hpds.cc + port: 80 + interval: 300 + timeout: 5 + deregister: 1 +db: + conn: root:123456@tcp(127.0.0.1:3306)/hpds_jky?charset=utf8mb4 + drive_name: mysql +cache: + host: 127.0.0.1 + port: 6379 + db: 0 + pool_size: 10 +node: + host: 127.0.0.1 + port: 27188 + token: 06d36c6f5705507dae778fdce90d0767 +functions: + - name: task-request + dataTag: 12 + mqType: 2 + - name: task-response + dataTag: 14 + mqType: 1 + - name: task-execute + dataTag: 16 + mqType: 1 \ No newline at end of file diff --git a/config/config.go b/config/config.go new file mode 100644 index 0000000..95b825b --- /dev/null +++ b/config/config.go @@ -0,0 +1,137 @@ +package config + +import ( + "fmt" + "os" + + consulapi "github.com/hashicorp/consul/api" + "github.com/spf13/viper" + _ "github.com/spf13/viper/remote" + yaml "gopkg.in/yaml.v3" +) + +type WebConfig struct { + Name string `yaml:"name,omitempty"` + Host string `yaml:"host,omitempty"` + Port int `yaml:"port,omitempty"` + Mode string `yaml:"mode,omitempty"` + Consul ConsulConfig `yaml:"consul,omitempty"` + Db DbConfig `yaml:"db"` + Cache CacheConfig `yaml:"cache"` + Logging LogOptions `yaml:"logging"` + Minio MinioConfig `yaml:"minio"` + Node HpdsNode `yaml:"node,omitempty"` + Funcs []FuncConfig `yaml:"functions,omitempty"` +} + +type ConsulConfig struct { + Host string `yaml:"host,omitempty"` + Port int `yaml:"port,omitempty"` + Interval int `yaml:"interval,omitempty"` + Timeout int `yaml:"timeout,omitempty"` + Deregister int `yaml:"deregister,omitempty"` + Tags []string `yaml:"tags,omitempty"` +} +type DbConfig struct { + Conn string `yaml:"conn"` + DriveName string `yaml:"drive_name"` +} +type CacheConfig struct { + Host string `yaml:"host,omitempty"` + Port int `yaml:"port,omitempty"` + Pass string `yaml:"pass,omitempty"` + DB int `yaml:"db,omitempty"` + PoolSize int `yaml:"pool_size,omitempty"` +} + +type LogOptions struct { + Path string `yaml:"path" json:"path" toml:"path"` // 文件保存地方 + Prefix string `yaml:"prefix" json:"prefix" toml:"prefix"` // 日志文件前缀 + ErrorFileSuffix string `yaml:"errorFileSuffix" json:"errorFileSuffix" toml:"errorFileSuffix"` // error日志文件后缀 + WarnFileSuffix string `yaml:"warnFileSuffix" json:"warnFileSuffix" toml:"warnFileSuffix"` // warn日志文件后缀 + InfoFileSuffix string `yaml:"infoFileSuffix" json:"infoFileSuffix" toml:"infoFileSuffix"` // info日志文件后缀 + DebugFileSuffix string `yaml:"debugFileSuffix" json:"debugFileSuffix" toml:"debugFileSuffix"` // debug日志文件后缀 + Level string `yaml:"level" json:"level" toml:"level"` // 日志等级 + MaxSize int `yaml:"maxSize" json:"maxSize" toml:"maxSize"` // 日志文件大小（M） + MaxBackups int `yaml:"maxBackups" json:"maxBackups" toml:"maxBackups"` // 最多存在多少个切片文件 + MaxAge int `yaml:"maxAge" json:"maxAge" toml:"maxAge"` // 保存的最大天数 + Development bool `yaml:"development" json:"development" toml:"development"` // 是否是开发模式 +} + +type MinioConfig struct { + Protocol string `yaml:"protocol"` //http or https + Endpoint string `yaml:"endpoint"` + AccessKeyId string `yaml:"accessKeyId"` + SecretAccessKey string `yaml:"secretAccessKey"` + Bucket string `yaml:"bucket"` +} +type HpdsNode struct { + Host string `yaml:"host"` + Port int `yaml:"port"` + Token string `yaml:"token,omitempty"` +} + +type FuncConfig struct { + Name string `yaml:"name"` + DataTag uint8 `yaml:"dataTag"` + MqType uint `yaml:"mqType"` //消息类型， 发布，1；订阅；2 +} + +func ParseConfigByFile(path string) (cfg *WebConfig, err error) { + buffer, err := os.ReadFile(path) + if err != nil { + return nil, err + } + return load(buffer) +} + +func load(buf []byte) (cfg *WebConfig, err error) { + cViper := viper.New() + cViper.SetConfigType("yaml") + cfg = new(WebConfig) + cfg.Funcs = make([]FuncConfig, 0) + //cViper.ReadConfig(bytes.NewBuffer(buf)) + err = yaml.Unmarshal(buf, cfg) + //err = cViper.Unmarshal(cfg) + //if err != nil { + // return nil, err + //} + return +} + +func UpdateLocalConfig(cfg *WebConfig, fn string) error { + data, err := yaml.Marshal(cfg) + if err != nil { + return err + } + err = os.WriteFile(fn, data, 0600) + return err +} + +func UpdateRemoteConfig(cfg *WebConfig) error { + consulClient, err := consulapi.NewClient(&consulapi.Config{Address: fmt.Sprintf("%s:%d", cfg.Consul.Host, cfg.Consul.Port)}) + if err != nil { + return err + } + val, err := yaml.Marshal(cfg) + if err != nil { + return err + } + p := &consulapi.KVPair{Key: fmt.Sprintf("hpds-pavement/hpds_control_center/%s/%s", cfg.Mode, cfg.Name), Value: val} + if _, err = consulClient.KV().Put(p, nil); err != nil { + return err + } + return nil +} + +func GetRemoteConfig(remoteAddr, path string) (cfg *WebConfig, err error) { + consulClient, err := consulapi.NewClient(&consulapi.Config{Address: remoteAddr}) + if err != nil { + return nil, err + } + kv, _, err := consulClient.KV().Get(path, nil) + if err != nil { + return nil, err + } + return load(kv.Value) +} diff --git a/config/config.yaml b/config/config.yaml new file mode 100644 index 0000000..7ef3888 --- /dev/null +++ b/config/config.yaml @@ -0,0 +1,48 @@ +name: control_center +host: 0.0.0.0 +port: 8088 +mode: dev +logging: + path: ./logs + prefix: hpds-control + errorFileSuffix: error.log + warnFileSuffix: warn.log + infoFileSuffix: info.log + debugFileSuffix: debug.log + maxSize: 100 + maxBackups: 3000 + maxAge: 30 + development: true +consul: + host: http://consul.hpds.cc + port: 80 + interval: 300 + timeout: 5 + deregister: 1 +db: + conn: root:123456@tcp(127.0.0.1:3306)/hpds_jky?charset=utf8mb4 + drive_name: mysql +cache: + host: 127.0.0.1 + port: 6379 + db: 0 + pool_size: 10 +minio: + protocol: http + endpoint: 127.0.0.1:9000 + accessKeyId: root + secretAccessKey: OIxv7QptYBO3 +node: + host: 114.55.236.153 + port: 9188 + token: 06d36c6f5705507dae778fdce90d0767 +functions: + - name: task-request + dataTag: 12 + mqType: 2 + - name: task-response + dataTag: 14 + mqType: 2 + - name: task-execute + dataTag: 16 + mqType: 1 \ No newline at end of file diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..d4f508f --- /dev/null +++ b/go.mod @@ -0,0 +1,102 @@ +module hpds_control_center + +go 1.19 + +require ( + git.hpds.cc/Component/logging v0.0.0-20230106105738-e378e873921b + git.hpds.cc/pavement/hpds_node v0.0.0-20230307094826-753c4fe9c877 + 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/spf13/cobra v1.6.1 + github.com/spf13/viper v1.15.0 + go.uber.org/zap v1.23.0 + gopkg.in/yaml.v3 v3.0.1 + xorm.io/xorm v1.3.2 +) + +require ( + cloud.google.com/go v0.105.0 // indirect + cloud.google.com/go/compute v1.14.0 // indirect + cloud.google.com/go/compute/metadata v0.2.3 // indirect + 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/fatih/color v1.13.0 // indirect + github.com/fsnotify/fsnotify v1.6.0 // indirect + github.com/go-ini/ini v1.67.0 // indirect + github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect + github.com/goccy/go-json v0.8.1 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect + github.com/golang/mock v1.6.0 // indirect + 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/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 + github.com/hashicorp/go-hclog v1.2.0 // indirect + github.com/hashicorp/go-immutable-radix v1.3.1 // indirect + github.com/hashicorp/go-rootcerts v1.0.2 // indirect + github.com/hashicorp/golang-lru v0.5.4 // indirect + github.com/hashicorp/hcl v1.0.0 // indirect + 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/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/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/pelletier/go-toml/v2 v2.0.6 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/sagikazarmark/crypt v0.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 + github.com/spf13/pflag v1.0.5 // indirect + github.com/subosito/gotenv v1.4.2 // indirect + github.com/syndtr/goleveldb v1.0.0 // indirect + go.etcd.io/etcd/api/v3 v3.5.6 // indirect + go.etcd.io/etcd/client/pkg/v3 v3.5.6 // indirect + go.etcd.io/etcd/client/v2 v2.305.6 // indirect + go.etcd.io/etcd/client/v3 v3.5.6 // indirect + 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/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/time v0.1.0 // indirect + golang.org/x/tools v0.1.12 // 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 + google.golang.org/genproto v0.0.0-20221227171554-f9683d7f8bef // indirect + google.golang.org/grpc v1.52.0 // indirect + 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 +) + +replace git.hpds.cc/pavement/hpds_node => ../hpds_node diff --git a/internal/balance/balanceFactory.go b/internal/balance/balanceFactory.go new file mode 100644 index 0000000..dcd5dda --- /dev/null +++ b/internal/balance/balanceFactory.go @@ -0,0 +1,25 @@ +package balance + +type LbType int + +const ( + LbRandom LbType = iota + LbRoundRobin + LbWeightRoundRobin + LbConsistentHash +) + +func LoadBalanceFactory(lbType LbType) LoadBalance { + switch lbType { + case LbRandom: + return &RandomBalance{} + case LbConsistentHash: + return NewConsistentHashBalance(10, nil) + case LbRoundRobin: + return &RoundRobinBalance{} + case LbWeightRoundRobin: + return &WeightRoundRobinBalance{} + default: + return &RandomBalance{} + } +} diff --git a/internal/balance/hash.go b/internal/balance/hash.go new file mode 100644 index 0000000..ef111ff --- /dev/null +++ b/internal/balance/hash.go @@ -0,0 +1,90 @@ +package balance + +import ( + "errors" + "fmt" + "hash/crc32" + "hpds_control_center/model" + "sort" + "strconv" + "sync" +) + +// Hash 1 单调性（唯一） 2平衡性 (数据 目标元素均衡) 3分散性(散列) +type Hash func(data []byte) uint32 + +type UInt32Slice []uint32 + +func (s UInt32Slice) Len() int { + return len(s) +} + +func (s UInt32Slice) Less(i, j int) bool { + return s[i] < s[j] +} + +func (s UInt32Slice) Swap(i, j int) { + s[i], s[j] = s[j], s[i] +} + +type ConsistentHashBalance struct { + mux sync.RWMutex + hash Hash + replicas int //复制因子 + keys UInt32Slice //已排序的节点hash切片 + hashMap map[uint32]int64 //节点哈希和key的map, 键是hash值，值是节点key +} + +func NewConsistentHashBalance(replicas int, fn Hash) *ConsistentHashBalance { + m := &ConsistentHashBalance{ + replicas: replicas, + hash: fn, + hashMap: make(map[uint32]int64), + } + if m.hash == nil { + //最多32位，保证是一个2^32-1环 + m.hash = crc32.ChecksumIEEE + } + return m +} + +func (c *ConsistentHashBalance) IsEmpty() bool { + return len(c.keys) == 0 +} + +// Add 方法用来添加缓存节点，参数为节点key，比如使用IP +func (c *ConsistentHashBalance) Add(params model.NodeLastStateItem) error { + + c.mux.Lock() + defer c.mux.Unlock() + + // 结合复制因子计算所有虚拟节点的hash值，并存入m.keys中，同时在m.hashMap中保存哈希值和key的映射 + 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 + } + + // 对所有虚拟节点的哈希值进行排序，方便之后进行二分查找 + sort.Sort(c.keys) + return nil +} + +// Get 方法根据给定的对象获取最靠近它的那个节点 +func (c *ConsistentHashBalance) Get(key int64) (int64, error) { + if c.IsEmpty() { + return 0, errors.New("node is empty") + } + hash := c.hash([]byte(fmt.Sprintf("%d", key))) + + // 通过二分查找获取最优节点，第一个"服务器hash"值大于"数据hash"值的就是最优"服务器节点" + idx := sort.Search(len(c.keys), func(i int) bool { return c.keys[i] >= hash }) + + // 如果查找结果 大于 服务器节点哈希数组的最大索引，表示此时该对象哈希值位于最后一个节点之后，那么放入第一个节点中 + if idx == len(c.keys) { + idx = 0 + } + c.mux.RLock() + defer c.mux.RUnlock() + return c.hashMap[c.keys[idx]], nil +} diff --git a/internal/balance/index.go b/internal/balance/index.go new file mode 100644 index 0000000..bfeb1e2 --- /dev/null +++ b/internal/balance/index.go @@ -0,0 +1,8 @@ +package balance + +import "hpds_control_center/model" + +type LoadBalance interface { + Add(model.NodeLastStateItem) error + Get(int64) (int64, error) +} diff --git a/internal/balance/randomBalance.go b/internal/balance/randomBalance.go new file mode 100644 index 0000000..982baba --- /dev/null +++ b/internal/balance/randomBalance.go @@ -0,0 +1,32 @@ +package balance + +import ( + "hpds_control_center/model" + "math/rand" +) + +// RandomBalance 随机负载均衡 +type RandomBalance struct { + curIndex int + + rss []int64 +} + +func (r *RandomBalance) Add(params model.NodeLastStateItem) error { + nodeId := params.NodeId + r.rss = append(r.rss, nodeId) + + return nil +} + +func (r *RandomBalance) Next() int64 { + if len(r.rss) == 0 { + return 0 + } + r.curIndex = rand.Intn(len(r.rss)) + return r.rss[r.curIndex] +} + +func (r *RandomBalance) Get(int64) (int64, error) { + return r.Next(), nil +} diff --git a/internal/balance/roundRobinBalance.go b/internal/balance/roundRobinBalance.go new file mode 100644 index 0000000..b9a63d8 --- /dev/null +++ b/internal/balance/roundRobinBalance.go @@ -0,0 +1,35 @@ +package balance + +import ( + "hpds_control_center/model" +) + +// RoundRobinBalance 轮询负载均衡 +type RoundRobinBalance struct { + curIndex int + rss []int64 +} + +func (r *RoundRobinBalance) Add(params model.NodeLastStateItem) error { + nodeId := params.NodeId + r.rss = append(r.rss, nodeId) + return nil +} + +func (r *RoundRobinBalance) Next() int64 { + if len(r.rss) == 0 { + return 0 + } + lens := len(r.rss) + if r.curIndex >= lens { + r.curIndex = 0 + } + + curNode := r.rss[r.curIndex] + r.curIndex = (r.curIndex + 1) % lens + return curNode +} + +func (r *RoundRobinBalance) Get(int64) (int64, error) { + return r.Next(), nil +} diff --git a/internal/balance/weightRoundRobinBalance.go b/internal/balance/weightRoundRobinBalance.go new file mode 100644 index 0000000..9831dec --- /dev/null +++ b/internal/balance/weightRoundRobinBalance.go @@ -0,0 +1,78 @@ +package balance + +import ( + "hpds_control_center/model" +) + +// WeightRoundRobinBalance 加权轮询负载 +type WeightRoundRobinBalance struct { + curIndex int + rss []*WeightNode + rsw []int +} + +type WeightNode struct { + node model.NodeLastStateItem + Weight int //初始化时对节点约定的权重 + currentWeight int //节点临时权重，每轮都会变化 + effectiveWeight int //有效权重, 默认与weight相同 , totalWeight = sum(effectiveWeight) //出现故障就-1 +} + +//1, currentWeight = currentWeight + effectiveWeight +//2, 选中最大的currentWeight节点为选中节点 +//3, currentWeight = currentWeight - totalWeight + +func (r *WeightRoundRobinBalance) Add(params model.NodeLastStateItem) error { + weightCpu := params.CpuUsed + weightMem := float64(params.MemUsed*100) / float64(params.MemUsed) + weightSwap := float64(params.SwapTotal*100) / float64(params.SwapTotal) + weightTask := 0.0 + if len(params.ExecTask) > 0 { + weightTask = 0 + } else { + weightTask = 100 + } + node := &WeightNode{ + node: params, + Weight: int(400 - weightCpu - weightMem - weightSwap - weightTask), + } + node.effectiveWeight = node.Weight + r.rss = append(r.rss, node) + return nil +} + +func (r *WeightRoundRobinBalance) Next() int64 { + var best *WeightNode + total := 0 + for i := 0; i < len(r.rss); i++ { + w := r.rss[i] + //1 计算所有有效权重 + total += w.effectiveWeight + //2 修改当前节点临时权重 + w.currentWeight += w.effectiveWeight + //3 有效权重默认与权重相同，通讯异常时-1, 通讯成功+1，直到恢复到weight大小 + if w.effectiveWeight < w.Weight { + w.effectiveWeight++ + } + + //4 选中最大临时权重节点 + if best == nil || w.currentWeight > best.currentWeight { + best = w + } + } + + if best == nil { + return 0 + } + //5 变更临时权重为 临时权重-有效权重之和 + best.currentWeight -= total + return best.node.NodeId +} + +func (r *WeightRoundRobinBalance) Get(int64) (int64, error) { + return r.Next(), nil +} + +func (r *WeightRoundRobinBalance) Update() { + +} diff --git a/internal/minio/index.go b/internal/minio/index.go new file mode 100644 index 0000000..e6b2be4 --- /dev/null +++ b/internal/minio/index.go @@ -0,0 +1,30 @@ +package minio + +import ( + "git.hpds.cc/Component/logging" + "github.com/minio/minio-go" +) + +type MinClient struct { + Client *minio.Client + Logger *logging.Logger +} + +func NewClient(ak, sak, ep string, useSSL bool, logger *logging.Logger) *MinClient { + client, err := minio.New(ep, ak, sak, useSSL) + if err != nil { + return nil + } + return &MinClient{ + Client: client, + Logger: logger, + } +} + +func (cli *MinClient) UploadObject(fn, dst, bucket string) error { + _, err := cli.Client.FPutObject(bucket, dst, fn, minio.PutObjectOptions{}) + if err != nil { + return err + } + return nil +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..4740ec6 --- /dev/null +++ b/main.go @@ -0,0 +1,27 @@ +package main + +import ( + "fmt" + "github.com/spf13/cobra" + "hpds_control_center/cmd" + "os" +) + +var ( + rootCmd = &cobra.Command{ + Use: "hpds_control_center", + Long: "hpds_control_center is a task control center", + Version: "0.1", + } +) + +func init() { + rootCmd.AddCommand(cmd.NewStartCmd()) +} +func main() { + + if err := rootCmd.Execute(); err != nil { + _, _ = fmt.Fprint(os.Stderr, err.Error()) + os.Exit(1) + } +} diff --git a/model/index.go b/model/index.go new file mode 100644 index 0000000..268b07a --- /dev/null +++ b/model/index.go @@ -0,0 +1,42 @@ +package model + +import ( + "go.uber.org/zap" + "os" + + _ "github.com/go-sql-driver/mysql" + "xorm.io/xorm" + "xorm.io/xorm/dialects" +) + +var ( + DB *xorm.Engine +) + +func New(driveName, dsn string, showSql bool) { + DB, _ = NewDbConnection(driveName, dsn) + DB.ShowSQL(showSql) + DB.Dialect().SetQuotePolicy(dialects.QuotePolicyReserved) + err := DB.Sync2( + &IssueModel{}, + &Model{}, + &Node{}, + &NodeState{}, + &Task{}, + &TaskResult{}, + ) + if err != nil { + zap.L().Error("同步数据库表结构", zap.Error(err)) + os.Exit(1) + } +} + +func NewDbConnection(driveName, dsn string) (db *xorm.Engine, err error) { + db, err = xorm.NewEngine(driveName, dsn) + if err != nil { + zap.L().Error("创建数据库连接", zap.Error(err)) + os.Exit(-1) + } + db.SetMaxOpenConns(300) + return +} diff --git a/model/model.go b/model/model.go new file mode 100644 index 0000000..992fb3c --- /dev/null +++ b/model/model.go @@ -0,0 +1,28 @@ +package model + +type Model struct { + ModelId int64 `xorm:"not null pk autoincr INT(11)" json:"modelId"` + ModelName string `xorm:"varchar(200) not null" json:"modelName"` + BizType int `xorm:"int not null default 1" json:"bizType"` + ModelVersion string `xorm:"varchar(50) not null" json:"modelVersion"` + ModelDesc string `xorm:"varchar(200) not null" json:"modelDesc"` + ModelFiles string `xorm:"varchar(200) not null" json:"modelFiles"` + ModelParamsFiles string `xorm:"varchar(200)" json:"modelParamsFiles"` + ModelExecScript string `xorm:"varchar(200)" json:"modelExecScript"` + IsLightWeight bool `xorm:"TINYINT(1) default 0" json:"isLightWeight"` + Status int `xorm:"not null SMALLINT default 0" json:"status"` + CreateAt int64 `xorm:"created" json:"createAt"` + UpdateAt int64 `xorm:"updated" json:"updateAt"` +} + +func GetModelById(id int64) *Model { + item := new(Model) + b, err := DB.ID(id).Get(item) + if err != nil { + return nil + } + if !b { + return nil + } + return item +} diff --git a/model/modelIssue.go b/model/modelIssue.go new file mode 100644 index 0000000..e5505d8 --- /dev/null +++ b/model/modelIssue.go @@ -0,0 +1,11 @@ +package model + +type IssueModel struct { + Id int64 `xorm:"not null pk autoincr INT(11)" json:"id"` + ModelId int64 `xorm:"INT(11) index" json:"modelId"` + NodeId int64 `xorm:"INT(11) index" json:"nodeId"` + Status int `xorm:"not null SMALLINT default 0" json:"status"` + IssueResult string `xorm:"TEXT" json:"issueResult"` + CreateAt int64 `xorm:"created" json:"createAt"` + UpdateAt int64 `xorm:"updated" json:"updateAt"` +} diff --git a/model/node.go b/model/node.go new file mode 100644 index 0000000..bd9ed6e --- /dev/null +++ b/model/node.go @@ -0,0 +1,46 @@ +package model + +// Node 节点信息 +type Node struct { + NodeId int64 `xorm:"not null pk autoincr INT(11)" json:"nodeId"` + NodeGuid string `xorm:"varchar(100) index" json:"nodeGuid"` + NodeName string `xorm:"varchar(100)" json:"nodeName"` + NodeType int `xorm:"not null SMALLINT default 0" json:"nodeType"` + NodeStatus int `xorm:"not null SMALLINT default 0" json:"nodeStatus"` + Platform string `xorm:"varchar(100)" json:"platform,omitempty"` + PlatformVersion string `xorm:"varchar(100)" json:"platformVersion,omitempty"` + CPU string `xorm:"varchar(1000)" json:"cpu,omitempty"` + MemTotal uint64 `xorm:"BIGINT" json:"memTotal,omitempty"` + DiskTotal uint64 `xorm:"BIGINT" json:"diskTotal,omitempty"` + SwapTotal uint64 `xorm:"BIGINT" json:"swapTotal,omitempty"` + Arch string `xorm:"varchar(1000)" json:"arch,omitempty"` + Virtualization string `xorm:"varchar(1000)" json:"virtualization,omitempty"` + BootTime uint64 `xorm:"BIGINT" json:"bootTime,omitempty"` + IP string `xorm:"varchar(100)" json:"ip"` + IsGpu bool `xorm:"TINYINT(1) default 0" json:"isGpu"` + IsTaskExecute bool `xorm:"TINYINT(1) default 0" json:"isTaskExecute"` + CountryCode string `xorm:"varchar(100)" json:"countryCode,omitempty"` + Version string `xorm:"varchar(100)" json:"version,omitempty"` + CreateAt int64 `xorm:"created" json:"createAt"` + UpdateAt int64 `xorm:"updated" json:"updateAt"` +} + +func GetLightWeight(modelId int64) []Node { + list := make([]Node, 0) + err := DB.Table("node").Alias("n").Join("inner", []string{"model_issue", "m"}, "m.node_id = n.node_id"). + Where("n.is_task_execute = 1").And("n.is_gpu = 1").And("m.model_id = ?", modelId).Find(&list) + if err != nil { + return nil + } + return list +} + +func GetAllNode(modelId int64) []Node { + list := make([]Node, 0) + err := DB.Table("node").Alias("n").Join("inner", []string{"model_issue", "m"}, "m.node_id = n.node_id"). + Where("m.model_id = ?", modelId).Find(&list) + if err != nil { + return nil + } + return list +} diff --git a/model/nodeState.go b/model/nodeState.go new file mode 100644 index 0000000..3fe2240 --- /dev/null +++ b/model/nodeState.go @@ -0,0 +1,60 @@ +package model + +// NodeState 节点状态信息 +type NodeState struct { + Uptime uint64 `xorm:"BIGINT pk" json:"uptime,omitempty"` + NodeName string `xorm:"varchar(100) pk" json:"nodeName"` + CPU float64 `xorm:"DECIMAL(18,4)" json:"cpu,omitempty"` + MemUsed uint64 `xorm:"BIGINT" json:"memUsed,omitempty"` + SwapUsed uint64 `xorm:"BIGINT" json:"swapUsed,omitempty"` + DiskUsed uint64 `xorm:"BIGINT" json:"diskUsed,omitempty"` + NetInTransfer uint64 `xorm:"BIGINT" json:"netInTransfer,omitempty"` + NetOutTransfer uint64 `xorm:"BIGINT" json:"netOutTransfer,omitempty"` + NetInSpeed uint64 `xorm:"BIGINT" json:"netInSpeed,omitempty"` + NetOutSpeed uint64 `xorm:"BIGINT" json:"netOutSpeed,omitempty"` + Load1 float64 `xorm:"DECIMAL(18,4)" json:"load1,omitempty"` + Load5 float64 `xorm:"DECIMAL(18,4)" json:"load5,omitempty"` + Load15 float64 `xorm:"DECIMAL(18,4)" json:"load15,omitempty"` + TcpConnCount uint64 `xorm:"BIGINT" json:"tcpConnCount,omitempty"` + UdpConnCount uint64 `xorm:"BIGINT" json:"udpConnCount,omitempty"` + ProcessCount uint64 `xorm:"BIGINT" json:"processCount,omitempty"` +} + +type NodeLastStateItem struct { + NodeId int64 `json:"nodeId"` + NodeGuid string `json:"nodeGuid"` + NodeName string `json:"nodeName"` + NodeType int `json:"nodeType"` + MemTotal uint64 `json:"memTotal"` + SwapTotal uint64 `json:"swapTotal"` + CpuUsed float64 `json:"cpuUsed"` + MemUsed uint64 `json:"memUsed"` + SwapUsed uint64 `json:"swapUsed"` + Load1 float64 `json:"load1"` + Load5 float64 `json:"load5"` + Load15 float64 `json:"load15"` + ExecTask string `json:"execTask"` +} + +func GetNodeState(nodeList []Node) []NodeLastStateItem { + list := make([]NodeLastStateItem, 0) + err := DB.SQL(`select c.node_id,c.node_name,c.node_guid,c.node_type,c.node_type,c.mem_total, + c.swap_total,a.c_p_u cpu_used, a.mem_used,a.swap_used, a.load1, a.load5, a.load15, + d.task_name exec_task from node_state a , (select node_name, max(uptime) uptime from node_state group by node_name) b, node c + left join (select t2.node_id, t2.task_name from task t2, (select node_id, max(start_time) start from task group by node_id) t1 + where t2.node_id = t1.node_id and t2.start_time = t1.start and t2.status = 1) d on c.node_id = d.node_id + where a.node_name = b.node_name and a.uptime = b.uptime and a.node_name = c.node_guid and c.node_status > 0 `).Find(&list) + if err != nil { + return nil + } + resList := make([]NodeLastStateItem, 0) + for _, v := range list { + for _, val := range nodeList { + if v.NodeGuid == val.NodeGuid { + resList = append(resList, v) + break + } + } + } + return resList +} diff --git a/model/task.go b/model/task.go new file mode 100644 index 0000000..11b7860 --- /dev/null +++ b/model/task.go @@ -0,0 +1,24 @@ +package model + +type Task struct { + TaskId int64 `xorm:"not null pk autoincr INT(11)" json:"taskId"` + ModelId int64 `xorm:"INT(11) index" json:"modelId"` + NodeId int64 `xorm:"INT(11) index" json:"nodeId"` + TaskName string `xorm:"VARCHAR(200)" json:"taskName"` + TaskDesc string `xorm:"VARCHAR(500)" json:"taskDesc"` + DatasetArr string `xorm:"TEXT" json:"datasetArr"` + ResultStorage string `xorm:"TEXT" json:"resultStorage"` + AppointmentTime string `xorm:"VARCHAR(30)" json:"appointmentTime"` + StartTime int64 `xorm:"BIGINT" json:"startTime"` + FinishTime int64 `xorm:"BIGINT" json:"finishTime"` + 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"` +} + +func UpdateTaskExecuteNode(id, nodeId int64) { + item := new(Task) + item.TaskId = id + item.NodeId = nodeId + _, _ = DB.ID(id).Cols("node_id").Update(item) +} diff --git a/model/taskResult.go b/model/taskResult.go new file mode 100644 index 0000000..b055467 --- /dev/null +++ b/model/taskResult.go @@ -0,0 +1,15 @@ +package model + +type TaskResult struct { + ResultId int64 `xorm:"not null pk autoincr INT(11)" json:"resultId"` + TaskId int64 `xorm:"INT(11) index" json:"taskId"` + TaskCode string `xorm:"varchar(200)" json:"taskCode"` + ModelId int64 `xorm:"INT(11)" json:"modelId"` + NodeId int64 `xorm:"INT(11)" json:"nodeId"` + StartTime int64 `xorm:"BIGINT" json:"startTime"` + FinishTime int64 `xorm:"BIGINT" json:"finishTime"` + 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"` +} diff --git a/mq/index.go b/mq/index.go new file mode 100644 index 0000000..041e477 --- /dev/null +++ b/mq/index.go @@ -0,0 +1,233 @@ +package mq + +import ( + "encoding/json" + "fmt" + "git.hpds.cc/Component/logging" + "go.uber.org/zap" + "hpds_control_center/config" + "hpds_control_center/internal/balance" + "hpds_control_center/model" + "os" + "time" + + "git.hpds.cc/pavement/hpds_node" +) + +var MqList []HpdsMqNode + +type HpdsMqNode struct { + MqType uint + Topic string + Node config.HpdsNode + EndPoint interface{} + Logger *logging.Logger +} + +func must(logger *logging.Logger, err error) { + if err != nil { + if logger != nil { + logger.With(zap.String("web节点", "错误信息")).Error("启动错误", zap.Error(err)) + } else { + _, _ = fmt.Fprint(os.Stderr, err) + } + os.Exit(1) + } +} + +func NewMqClient(funcs []config.FuncConfig, node config.HpdsNode, logger *logging.Logger) (mqList []HpdsMqNode, err error) { + mqList = make([]HpdsMqNode, 0) + for _, v := range funcs { + switch v.MqType { + case 2: + sf := hpds_node.NewStreamFunction( + v.Name, + hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", node.Host, node.Port)), + hpds_node.WithObserveDataTags(v.DataTag), + hpds_node.WithCredential(node.Token), + ) + err = sf.Connect() + must(logger, err) + nodeInfo := HpdsMqNode{ + MqType: 2, + Topic: v.Name, + Node: node, + EndPoint: sf, + } + switch v.Name { + case "task-request": + _ = sf.SetHandler(TaskRequestHandler) + default: + + } + mqList = append(mqList, nodeInfo) + default: + ap := hpds_node.NewAccessPoint( + v.Name, + hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", node.Host, node.Port)), + hpds_node.WithCredential(node.Token), + ) + err = ap.Connect() + nodeInfo := HpdsMqNode{ + MqType: 1, + Topic: v.Name, + Node: node, + EndPoint: ap, + } + must(logger, err) + ap.SetDataTag(v.DataTag) + mqList = append(mqList, nodeInfo) + } + + } + return mqList, err +} + +func GetMqClient(topic string, mqType uint) *HpdsMqNode { + for _, v := range MqList { + if v.Topic == topic && v.MqType == mqType { + return &v + } + } + return nil +} + +func GenerateAndSendData(stream hpds_node.AccessPoint, data []byte) error { + _, err := stream.Write(data) + if err != nil { + return err + } + time.Sleep(1000 * time.Millisecond) + return nil +} + +func TaskRequestHandler(data []byte) (byte, []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 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) + } 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) + } + } + + case ModelIssue: + payload := cmd.Payload.(map[string]interface{}) + cmd := &InstructionReq{ + Command: ModelIssueRepeater, + Payload: payload, + } + pData, _ := json.Marshal(cmd) + cli := GetMqClient("task-execute", 1) + if cli != nil { + _ = GenerateAndSendData(cli.EndPoint.(hpds_node.AccessPoint), pData) + } + case ModelIssueResponse: + payload := cmd.Payload.(map[string]interface{}) + //查找下发记录 + item := new(model.IssueModel) + h, _ := model.DB.Where("model_id = ? and node_id = ?", payload["modelId"].(int64), payload["nodeId"].(int64)).Get(item) + pData, _ := json.Marshal(payload) + if h { + item.Status = 1 + item.IssueResult = string(pData) + item.UpdateAt = time.Now().Unix() + _, _ = model.DB.ID(item.Id).AllCols().Update(item) + } else { + item.ModelId = payload["modelId"].(int64) + item.NodeId = payload["nodeId"].(int64) + item.Status = 1 + item.IssueResult = string(pData) + item.CreateAt = time.Now().Unix() + 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))) + + default: + + } + return byte(cmd.Command), nil +} diff --git a/mq/instruction.go b/mq/instruction.go new file mode 100644 index 0000000..abdad42 --- /dev/null +++ b/mq/instruction.go @@ -0,0 +1,15 @@ +package mq + +const ( + TaskAdd = iota + 1 + ModelIssue + TaskExecute + TaskResponse + ModelIssueRepeater + ModelIssueResponse +) + +type InstructionReq struct { + Command int `json:"command"` + Payload interface{} `json:"payload"` +} diff --git a/pkg/discover/consul/consul.go b/pkg/discover/consul/consul.go new file mode 100644 index 0000000..64b0148 --- /dev/null +++ b/pkg/discover/consul/consul.go @@ -0,0 +1,87 @@ +package discover + +import ( + "fmt" + + "github.com/hashicorp/consul/api" +) + +type ConsulConfig struct { + Client *api.Client `json:"client"` // consul client + ConsulAddress string `json:"consulAddress"` // consul 服务地址:IP+port + ServiceId string `json:"serviceId"` // 服务ID + ServiceName string `json:"serviceName"` // 服务名称 + ServiceIP string `json:"serviceIP"` // 服务IP + ServicePort int `json:"servicePort"` // 服务端口 + Tags []string `json:"tags"` // 服务标签列表 + DeregisterCriticalServiceAfter int `json:"deregisterCriticalServiceAfter"` // 指定与服务关联的检查应在此时间之后注销 + Interval int `json:"interval"` // 指定运行此检查的频率 + Timeout int `json:"timeout"` // 在脚本、HTTP、TCP 或 gRPC 检查的情况下指定传出连接的超时时间 +} + +func NewConsulConfig(consulAddress string, + serviceId string, + serviceName string, + serviceIP string, + servicePort int, + tags []string, + deregisterCriticalServiceAfter int, + interval int, + timeout int) (*ConsulConfig, error) { + // 1.consul配置 + config := api.DefaultConfig() + config.Address = consulAddress + // 2.client + client, err := api.NewClient(config) + if err != nil { + return nil, err + } + return &ConsulConfig{ + Client: client, + ConsulAddress: consulAddress, + ServiceId: serviceId, + ServiceName: serviceName, + ServiceIP: serviceIP, + ServicePort: servicePort, + Tags: tags, + DeregisterCriticalServiceAfter: deregisterCriticalServiceAfter, + Interval: interval, + Timeout: timeout, + }, nil + +} + +// ServiceRegister 服务注册 +func (cf *ConsulConfig) ServiceRegister() (err error) { + // 注册器 + reg := &api.AgentServiceRegistration{ + ID: cf.ServiceId, + Name: cf.ServiceName, + Address: cf.ServiceIP, + Port: cf.ServicePort, + Tags: cf.Tags, + Check: &api.AgentServiceCheck{ + Interval: fmt.Sprintf("%vs", cf.Interval), // 健康检查间隔 + HTTP: fmt.Sprintf("http://%v:%v/health", cf.ServiceIP, cf.ServicePort), // HTTP 支持，执行健康检查的地址，service 会传到 Health.Check 函数中 + Timeout: fmt.Sprintf("%vs", cf.Timeout), // 健康检查超时时间 + DeregisterCriticalServiceAfter: fmt.Sprintf("%vs", cf.DeregisterCriticalServiceAfter), // 注销时间，相当于过期时间 + Notes: "Consul check service health status.", + }, + } + // 注册服务 + err = cf.Client.Agent().ServiceRegister(reg) + if err != nil { + return err + } + return nil +} + +// ServiceDeregister 服务注销 +func (cf *ConsulConfig) ServiceDeregister() error { + return cf.Client.Agent().ServiceDeregister(cf.ServiceId) +} + +// ServiceDiscover 服务发现 +func (cf *ConsulConfig) ServiceDiscover(service string, tags []string, q *api.QueryOptions) ([]*api.CatalogService, *api.QueryMeta, error) { + return cf.Client.Catalog().ServiceMultipleTags(service, tags, q) +}