environmentCaptureAgent/cmd/server.go

package cmd

import (
	"environmentCaptureAgent/config"
	"environmentCaptureAgent/internal/monitor"
	"fmt"
	"git.hpds.cc/pavement/hpds_node"
	"go.uber.org/zap"
	"os"
	"os/signal"
	"syscall"
	"time"

	"git.hpds.cc/Component/logging"
	"github.com/spf13/cobra"
)

var (
	ConfigFileFlag string = "./config/config.yaml"
	logger         *logging.Logger
)

func must(err error) {
	if err != nil {
		if logger != nil {
			logger.With(zap.String("capture agent", "错误信息")).Error("启动错误", zap.Error(err))
		} else {
			_, _ = fmt.Fprint(os.Stderr, err)
		}
		os.Exit(1)
	}
}

func NewStartCmd() *cobra.Command {
	cmd := &cobra.Command{
		Use:   "start",
		Short: "Start hpds environment capture agent",
		Run: func(cmd *cobra.Command, args []string) {
			var (
				cfg *config.AgentConfig
				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)
			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)
			ap := hpds_node.NewAccessPoint(
				cfg.Name,
				hpds_node.WithMqAddr(fmt.Sprintf("%s:%d", cfg.Node.Host, cfg.Node.Port)),
				hpds_node.WithCredential(cfg.Node.Token),
			)
			err = ap.Connect()
			must(err)
			ap.SetDataTag(18)
			node := monitor.GetHost(cfg)
			byteNode := node.ToByte()
			_ = generateAndSendData(ap, byteNode)

			statTicker := time.NewTicker(time.Duration(cfg.Delay) * time.Second)
			nodeTicker := time.NewTicker(1 * time.Hour)
			count := 0
			for {
				select {
				case <-statTicker.C:
					monitor.TrackNetworkSpeed()
					stat := monitor.GetState(node.NodeName).ToByte()
					go func() {
						_ = generateAndSendData(ap, stat)
					}()
				case <-nodeTicker.C:
					nodeInfo := monitor.GetHost(cfg).ToByte()
					go func() {
						_ = generateAndSendData(ap, nodeInfo)
					}()
				case errs := <-exitChannel:
					count++
					if count > 3 {
						logger.With(
							zap.String("agent", "服务退出"),
						).Info(errs.String())
						return
					}
				}
			}
		},
	}
	cmd.Flags().StringVar(&ConfigFileFlag, "c", "./config/config.yaml", "The configuration file path")
	return cmd
}

func generateAndSendData(stream hpds_node.AccessPoint, data []byte) error {
	logger.With(zap.String("agent", "发送数据")).Info("数据报文", zap.ByteString("原文", data))
	_, err := stream.Write(data)
	if err != nil {
		return err
	}
	time.Sleep(1000 * time.Millisecond)
	return nil
}

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"]),
	)
}