From 45b91925712592dd2ec3a9f2dc7c6b99f2cb9c74 Mon Sep 17 00:00:00 2001 From: wangjian Date: Wed, 1 Mar 2023 14:10:50 +0800 Subject: [PATCH] =?UTF-8?q?=E7=89=88=E6=9C=AC=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 | 3 ++ cmd/server.go | 102 +++++++++++++++++++++++++++++++++++++++ config/config.go | 67 ++++++++++++++++++++++++++ config/config.yaml | 23 +++++++++ config/config_real.yaml | 23 +++++++++ go.mod | 57 ++++++++++++++++++++++ main.go | 27 +++++++++++ model/index.go | 40 ++++++++++++++++ model/node.go | 123 ++++++++++++++++++++++++++++++++++++++++++++++++ 9 files changed, 465 insertions(+) create mode 100644 README.md create mode 100644 cmd/server.go create mode 100644 config/config.go create mode 100644 config/config.yaml create mode 100644 config/config_real.yaml create mode 100644 go.mod create mode 100644 main.go create mode 100644 model/index.go create mode 100644 model/node.go diff --git a/README.md b/README.md new file mode 100644 index 0000000..9dc1e48 --- /dev/null +++ b/README.md @@ -0,0 +1,3 @@ +## 采集数据落地应用 + +将环境采集的数据保存到数据库 \ No newline at end of file diff --git a/cmd/server.go b/cmd/server.go new file mode 100644 index 0000000..ed6183e --- /dev/null +++ b/cmd/server.go @@ -0,0 +1,102 @@ +package cmd + +import ( + "encoding/json" + "fmt" + "github.com/spf13/cobra" + "go.uber.org/zap" + "os" + "os/signal" + "stream_capture_db/config" + "stream_capture_db/model" + "syscall" + + "git.hpds.cc/Component/logging" + "git.hpds.cc/pavement/hpds_node" +) + +var ( + ConfigFileFlag string = "./config/config.yaml" +) + +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.StreamFuncCaptureDBConfig + err error + ) + configFileFlag, err := cmd.Flags().GetString("c") + if err != nil { + fmt.Println("get local config err: ", err) + return + } + must(err) + cfg, err = config.ParseConfigByFile(configFileFlag) + must(err) + //连接数据库 + model.New(cfg.Db.DriveName, cfg.Db.Conn) + + model.Logger = LoadLoggerConfig(cfg.Logging) + exitChannel := make(chan os.Signal) + defer close(exitChannel) + // 退出信号监听 + go func(c chan os.Signal) { + signal.Notify(c, syscall.SIGINT, syscall.SIGTERM) + }(exitChannel) + sf := hpds_node.NewStreamFunction( + cfg.Functions.Name, + hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", cfg.Node.Host, cfg.Node.Port)), + hpds_node.WithObserveDataTags(cfg.Functions.DataTag), + hpds_node.WithCredential(cfg.Node.Token), + ) + _ = sf.SetHandler(handler) + // start + _ = sf.Connect() + select {} + }, + } + cmd.Flags().StringVar(&ConfigFileFlag, "c", "./config/config.yaml", "The configuration file path") + 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"]), + ) +} + +func handler(data []byte) (byte, []byte) { + node := new(model.Node) + err := json.Unmarshal(data, node) + if err != nil { + stat := new(model.NodeState) + err = json.Unmarshal(data, stat) + if err != nil { + model.Logger.With(zap.String("The received data", string(data))).Error("数据无法解析", zap.Error(err)) + return 0x12, []byte(err.Error()) + } + stat.Insert() + } else { + node.Insert() + } + return 0x12, nil +} diff --git a/config/config.go b/config/config.go new file mode 100644 index 0000000..332cce4 --- /dev/null +++ b/config/config.go @@ -0,0 +1,67 @@ +package config + +import ( + "bytes" + "os" + + "github.com/spf13/viper" +) + +type StreamFuncCaptureDBConfig struct { + Name string `yaml:"name,omitempty"` + Mode string `yaml:"mode,omitempty"` + Logging LogOptions `yaml:"logging"` + Node HpdsNode `yaml:"node,omitempty"` + Db DbConfig `yaml:"db"` + Functions FuncConfig `yaml:"functions,omitempty"` +} + +type DbConfig struct { + Conn string `yaml:"conn"` + DriveName string `yaml:"driveName"` +} + +type FuncConfig struct { + Name string `yaml:"name"` + DataTag uint8 `yaml:"dataTag"` +} + +type HpdsNode struct { + Host string `yaml:"host"` + Port int `yaml:"port"` + Token string `yaml:"token,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"` // 是否是开发模式 +} + +func ParseConfigByFile(path string) (cfg *StreamFuncCaptureDBConfig, err error) { + buffer, err := os.ReadFile(path) + if err != nil { + return nil, err + } + return load(buffer) +} + +func load(buf []byte) (cfg *StreamFuncCaptureDBConfig, err error) { + cViper := viper.New() + cViper.SetConfigType("yaml") + cfg = new(StreamFuncCaptureDBConfig) + cViper.ReadConfig(bytes.NewBuffer(buf)) + err = cViper.Unmarshal(cfg) + if err != nil { + return nil, err + } + return +} diff --git a/config/config.yaml b/config/config.yaml new file mode 100644 index 0000000..1dddb41 --- /dev/null +++ b/config/config.yaml @@ -0,0 +1,23 @@ +name: capture-db +mode: dev +node: + host: 114.55.236.153 + port: 9188 + token: 06d36c6f5705507dae778fdce90d0767 +logging: + path: ./logs + prefix: capture-agent + errorFileSuffix: error.log + warnFileSuffix: warn.log + infoFileSuffix: info.log + debugFileSuffix: debug.log + maxSize: 100 + maxBackups: 3000 + maxAge: 30 + development: true +db: + conn: root:123456@tcp(127.0.0.1:3306)/hpds_jky?charset=utf8mb4 + driveName: mysql +functions: + name: capture-agent + dataTag: 18 \ No newline at end of file diff --git a/config/config_real.yaml b/config/config_real.yaml new file mode 100644 index 0000000..b70a6e1 --- /dev/null +++ b/config/config_real.yaml @@ -0,0 +1,23 @@ +name: capture-db +mode: dev +node: + host: 127.0.0.1 + port: 9188 + token: 06d36c6f5705507dae778fdce90d0767 +logging: + path: ./logs + prefix: capture2db + errorFileSuffix: error.log + warnFileSuffix: warn.log + infoFileSuffix: info.log + debugFileSuffix: debug.log + maxSize: 100 + maxBackups: 3000 + maxAge: 30 + development: true +db: + conn: root:OIxv7QptYBO3@tcp(172.28.125.20:23306)/diagnostic_platform?charset=utf8mb4 + driveName: mysql +functions: + name: capture-agent + dataTag: 18 \ No newline at end of file diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..d4f75c5 --- /dev/null +++ b/go.mod @@ -0,0 +1,57 @@ +module stream_capture_db + +go 1.19 + +require ( + git.hpds.cc/Component/logging v0.0.0-20230106105738-e378e873921b + git.hpds.cc/pavement/hpds_node v0.0.0-20221023053316-37f7ba99eab3 + github.com/go-sql-driver/mysql v1.6.0 + github.com/spf13/cobra v0.0.3 + github.com/spf13/viper v1.15.0 + go.uber.org/zap v1.23.0 + xorm.io/xorm v1.3.2 +) + +require ( + 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/fsnotify/fsnotify v1.6.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/golang/mock v1.6.0 // indirect + github.com/golang/snappy v0.0.4 // indirect + github.com/hashicorp/hcl v1.0.0 // indirect + github.com/inconshreveable/mousetrap v1.0.0 // 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/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/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.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/sys v0.3.0 // indirect + golang.org/x/text v0.5.0 // indirect + golang.org/x/tools v0.1.12 // 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 + gopkg.in/yaml.v3 v3.0.1 // indirect + xorm.io/builder v0.3.11-0.20220531020008-1bd24a7dc978 // indirect +) diff --git a/main.go b/main.go new file mode 100644 index 0000000..d5d2e35 --- /dev/null +++ b/main.go @@ -0,0 +1,27 @@ +package main + +import ( + "fmt" + "github.com/spf13/cobra" + "os" + "stream_capture_db/cmd" +) + +var ( + rootCmd = &cobra.Command{ + Use: "hpds_stream_functions_capture_to_database", + Long: "The stream functions is capture data to database service", + 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..0c2f45a --- /dev/null +++ b/model/index.go @@ -0,0 +1,40 @@ +package model + +import ( + "git.hpds.cc/Component/logging" + _ "github.com/go-sql-driver/mysql" + "go.uber.org/zap" + "os" + + "xorm.io/xorm" + "xorm.io/xorm/dialects" +) + +var ( + DB *xorm.Engine + Logger *logging.Logger +) + +func New(driveName, dsn string) { + DB, _ = NewDbConnection(driveName, dsn) + DB.ShowSQL(true) + DB.Dialect().SetQuotePolicy(dialects.QuotePolicyReserved) + err := DB.Sync2( + &Node{}, + &NodeState{}, + ) + 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/node.go b/model/node.go new file mode 100644 index 0000000..7e5142d --- /dev/null +++ b/model/node.go @@ -0,0 +1,123 @@ +package model + +import ( + "encoding/json" + "go.uber.org/zap" +) + +// Node 节点信息 +type Node struct { + NodeId int64 `xorm:"not null pk autoincr INT(11)" json:"-"` + 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"` + 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 (n Node) ToByte() []byte { + data, err := json.Marshal(n) + if err != nil { + return []byte("") + } + return data +} +func (n Node) ToString() string { + data, err := json.Marshal(n) + if err != nil { + return "" + } + return string(data) +} + +func (n Node) Insert() { + node := new(Node) + b, err := DB.Where("node_guid = ?", n.NodeName).Get(node) + if err != nil { + Logger.With(zap.String("Node", n.ToString())).Error("查询节点失败", zap.Error(err)) + return + } + node.NodeGuid = n.NodeName + node.Platform = n.Platform + node.PlatformVersion = n.PlatformVersion + node.CPU = n.CPU + node.MemTotal = n.MemTotal + node.DiskTotal = n.DiskTotal + node.SwapTotal = n.SwapTotal + node.Arch = n.Arch + node.Virtualization = n.Virtualization + node.BootTime = n.BootTime + node.IP = n.IP + node.CountryCode = n.CountryCode + node.Version = n.Version + if b { + _, err = DB.ID(node.NodeId).Update(node) + if err != nil { + Logger.With(zap.String("Node", node.ToString())).Error("更新节点失败", zap.Error(err)) + return + } + } else { + _, err := DB.Insert(node) + if err != nil { + Logger.With(zap.String("Node", node.ToString())).Error("入库失败", zap.Error(err)) + return + } + } +} + +// 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"` +} + +func (ns NodeState) ToByte() []byte { + data, err := json.Marshal(ns) + if err != nil { + return []byte("") + } + return data +} + +func (ns NodeState) ToString() string { + data, err := json.Marshal(ns) + if err != nil { + return "" + } + return string(data) +} + +func (ns NodeState) Insert() { + _, err := DB.Insert(ns) + if err != nil { + Logger.With(zap.String("NodeState", ns.ToString())).Error("入库失败", zap.Error(err)) + return + } +}