cicv_data_closed_loop/aarch64/pjisuv/common/config/c_cloud.go

package config

import (
	"cicv-data-closedloop/common/config/c_log"
	"cicv-data-closedloop/common/util"
	"gopkg.in/yaml.v3"
	"os"
	"sync"
	"time"
)

type MonitorStruct struct {
	Url string `yaml:"url"`
}

type platform struct {
	UrlDeviceAuth string `yaml:"url-device-auth"`
	UrlTaskPoll   string `yaml:"url-task-poll"`
	UrlTask       string `yaml:"url-task"`
}

type rosbagStruct struct {
	Path string   `yaml:"path"`
	Envs []string `yaml:"envs"`
}

type hostStruct struct {
	Name   string       `yaml:"name"`
	Ip     string       `yaml:"ip"`
	Topics []string     `yaml:"topics"`
	Rosbag rosbagStruct `yaml:"rosbag"`
}
type ros struct {
	MasterAddress string   `yaml:"master-address"`
	Nodes         []string `yaml:"nodes"`
}
type disk struct {
	Name string `yaml:"name"`
	Used uint64 `yaml:"used"`
}

type trigger struct {
	Label  string   `yaml:"label"`
	Topics []string `yaml:"topics"`
}

type CloudConfigStruct struct {
	RefreshCloudConfig    bool          `yaml:"refresh-cloud-config"`
	CompressBag           bool          `yaml:"compress-bag"`
	CleanBeforeStart      bool          `yaml:"clean-before-start"`
	FullCollect           bool          `yaml:"full-collect"`
	ConfigRefreshInterval int           `yaml:"config-refresh-interval"` // 配置刷新时间间隔
	BagNumber             int           `yaml:"bag-number"`
	TimeWindowSendGap     int           `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔
	TimeToLabelJsonPath   string        `yaml:"time-to-label-json-path"`
	BagDataDir            string        `yaml:"bag-data-dir"`
	BagCopyDir            string        `yaml:"bag-copy-dir"`
	TriggersDir           string        `yaml:"triggers-dir"`
	TcpPort               string        `yaml:"tcp-port"`
	RpcPort               string        `yaml:"rpc-port"`
	Triggers              []trigger     `yaml:"triggers"`
	Hosts                 []hostStruct  `yaml:"hosts"`
	Ros                   ros           `yaml:"ros"`
	Platform              platform      `yaml:"platform"`
	Disk                  disk          `yaml:"disk"`
	Monitor               MonitorStruct `yaml:"monitor"`
}

var (
	CloudConfig      CloudConfigStruct
	CloudConfigMutex sync.RWMutex
)

// 初始化业务配置
func InitCloudConfig() {
	c_log.GlobalLogger.Info("初始化OSS配置文件 - 开始。")
	var content []byte // cloud.yaml 内容
	var err error
	if LocalConfig.Internet {
		c_log.GlobalLogger.Info("车辆可以访问互联网。")
		// 获取文件的目录
		_ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
		// 3 ------- 获取 yaml 字符串 -------
		cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
		// todo 等待时间同步
		// 判断文件是否存在。如果不存在则使用默认的
		isExist, err := OssBucket.IsObjectExist(cloudConfigObjectKey)
		if err != nil {
			c_log.GlobalLogger.Errorf("判断配置文件是否存在失败，错误信息为：%v", err)
		}
		if isExist {
			c_log.GlobalLogger.Info("使用机器人自定义配置文件:", cloudConfigObjectKey)
		} else {
			cloudConfigObjectKey = LocalConfig.OssBasePrefix + LocalConfig.CloudConfigFilename // 默认配置文件路径
			c_log.GlobalLogger.Info("使用默认配置文件:", cloudConfigObjectKey)
		}
		for {
			OssMutex.Lock()
			err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
			OssMutex.Unlock()
			if err != nil {
				c_log.GlobalLogger.Error("下载oss上的配置文件 "+cloudConfigObjectKey+" 失败。网络授时未完成或者未配置阿里云网络映射到/etc/hosts：", err)
				time.Sleep(time.Duration(2) * time.Second)
				continue
			}
			break
		}
		if CloudConfig.RefreshCloudConfig {
			go RefreshCloudConfig()
		}
	} else {
		c_log.GlobalLogger.Infof("车辆不可以访问互联网，请提前将配置文件放到【%v】。", LocalConfig.CloudConfigLocalPath)
		for {
			time.Sleep(time.Duration(1) * time.Second)
			if util.FileExists(LocalConfig.CloudConfigLocalPath) {
				c_log.GlobalLogger.Infof("配置文件【%v】已准备好。", LocalConfig.CloudConfigLocalPath)
				break
			} else {
				c_log.GlobalLogger.Infof("配置文件【%v】没有准备好。", LocalConfig.CloudConfigLocalPath)
				continue
			}
		}
	}

	content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
	if err != nil {
		c_log.GlobalLogger.Error("程序退出。配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
		os.Exit(-1)
	}

	// 4 ------- 解析YAML内容 -------
	var newCloudConfig CloudConfigStruct
	err = yaml.Unmarshal(content, &newCloudConfig)
	if err != nil {
		c_log.GlobalLogger.Error("程序退出。配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
		os.Exit(-1)
	}

	// 5 ------- 校验 yaml -------
	if checkConfig(newCloudConfig) {
		CloudConfigMutex.RLock()
		CloudConfig = newCloudConfig
		CloudConfigMutex.RUnlock()
	} else {
		c_log.GlobalLogger.Error("配置文件格式错误：", newCloudConfig)
		os.Exit(-1)
	}
	c_log.GlobalLogger.Info("初始化OSS配置文件 - 成功。")
	util.CreateDir(CloudConfig.BagDataDir)
	util.CreateDir(CloudConfig.BagCopyDir)

	_ = util.RemoveSubFiles(CloudConfig.BagDataDir)
	if CloudConfig.CleanBeforeStart { // 判断是否需要清空原数据缓存
		if LocalConfig.Node.Name == "node1" { // 只有cloud配置了同时节点为node1才清除
			// 清空 timeToLabel.json
			timeToLabelJson, _ := util.MapToJsonString(map[string]interface{}{"time": "label"})
			_ = util.WriteFile(timeToLabelJson, CloudConfig.TimeToLabelJsonPath)
			c_log.GlobalLogger.Info("清空timeToLabel.json - 成功。")
		}
		_ = util.RemoveSubFiles(CloudConfig.BagCopyDir)
	}
}

// 刷新配置文件
func refreshCloudConfig() {
	// 获取文件的目录
	_ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
	// 3 ------- 获取 yaml 字符串 -------
	var content []byte
	cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
	// 判断文件是否存在。如果不存在则使用默认的
	isExist, err := OssBucket.IsObjectExist(cloudConfigObjectKey)
	if err != nil {
		c_log.GlobalLogger.Errorf("判断配置文件是否存在失败，错误信息为：%v", err)
		return
	}
	if !isExist {
		cloudConfigObjectKey = LocalConfig.OssBasePrefix + LocalConfig.CloudConfigFilename // 默认配置文件路径
	}
	OssMutex.Lock()
	err = OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
	OssMutex.Unlock()
	if err != nil {
		c_log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
		return
	}

	content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
	if err != nil {
		c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
		return
	}

	// 4 ------- 解析YAML内容 -------
	var newCloudConfig CloudConfigStruct
	err = yaml.Unmarshal(content, &newCloudConfig)
	if err != nil {
		c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
		return
	}

	// 5 ------- 校验 yaml -------
	if checkConfig(newCloudConfig) {
		CloudConfigMutex.RLock()
		CloudConfig = newCloudConfig
		CloudConfigMutex.RUnlock()
	} else {
		c_log.GlobalLogger.Error("配置文件格式错误：", newCloudConfig)
		return
	}
	util.CreateDir(CloudConfig.BagDataDir)
	util.CreateDir(CloudConfig.BagCopyDir)
}

// RefreshCloudConfig 轮询oss上的配置文件更新到本地
func RefreshCloudConfig() {
	for {
		time.Sleep(time.Duration(CloudConfig.ConfigRefreshInterval) * time.Second)
		refreshCloudConfig()
	}
}

// CheckConfig 校验 cfg.yaml 文件
func checkConfig(check CloudConfigStruct) bool {
	if len(check.Hosts) != 2 {
		c_log.GlobalLogger.Error("cloud-config.yaml中配置的hosts必须为2。")
		os.Exit(-1)
	}
	return true
}