1、初始化
This commit is contained in:
wangjian
commit
46d29e280b
|
|
@ -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"]),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
|
@ -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)
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
|
@ -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
|
||||||
|
|
@ -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{}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,8 @@
|
||||||
|
package balance
|
||||||
|
|
||||||
|
import "hpds_control_center/model"
|
||||||
|
|
||||||
|
type LoadBalance interface {
|
||||||
|
Add(model.NodeLastStateItem) error
|
||||||
|
Get(int64) (int64, error)
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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() {
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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"`
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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)
|
||||||
|
}
|
||||||
|
|
@ -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"`
|
||||||
|
}
|
||||||
|
|
@ -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
|
||||||
|
}
|
||||||
|
|
@ -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"`
|
||||||
|
}
|
||||||
|
|
@ -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)
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue