c_cloud.go 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210
  1. package config
  2. import (
  3. "cicv-data-closedloop/common/config/c_log"
  4. "cicv-data-closedloop/common/util"
  5. "gopkg.in/yaml.v3"
  6. "os"
  7. "sync"
  8. "time"
  9. )
  10. type MonitorStruct struct {
  11. Url string `yaml:"url"`
  12. }
  13. type platform struct {
  14. UrlDeviceAuth string `yaml:"url-device-auth"`
  15. UrlTaskPoll string `yaml:"url-task-poll"`
  16. UrlTask string `yaml:"url-task"`
  17. }
  18. type rosbagStruct struct {
  19. Path string `yaml:"path"`
  20. Envs []string `yaml:"envs"`
  21. }
  22. type hostStruct struct {
  23. Name string `yaml:"name"`
  24. Ip string `yaml:"ip"`
  25. Topics []string `yaml:"topics"`
  26. Rosbag rosbagStruct `yaml:"rosbag"`
  27. }
  28. type ros struct {
  29. MasterAddress string `yaml:"master-address"`
  30. Nodes []string `yaml:"nodes"`
  31. }
  32. type disk struct {
  33. Name string `yaml:"name"`
  34. Used uint64 `yaml:"used"`
  35. }
  36. type trigger struct {
  37. Label string `yaml:"label"`
  38. Topics []string `yaml:"topics"`
  39. }
  40. type cloudConfig struct {
  41. RefreshCloudConfig bool `yaml:"refresh-cloud-config"`
  42. CompressBag bool `yaml:"compress-bag"`
  43. CleanBeforeStart bool `yaml:"clean-before-start"`
  44. FullCollect bool `yaml:"full-collect"`
  45. ConfigRefreshInterval int `yaml:"config-refresh-interval"` // 配置刷新时间间隔
  46. BagNumber int `yaml:"bag-number"`
  47. TimeWindowSendGap int `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔
  48. TimeToLabelJsonPath string `yaml:"time-to-label-json-path"`
  49. BagDataDir string `yaml:"bag-data-dir"`
  50. BagCopyDir string `yaml:"bag-copy-dir"`
  51. TriggersDir string `yaml:"triggers-dir"`
  52. TcpPort string `yaml:"tcp-port"`
  53. RpcPort string `yaml:"rpc-port"`
  54. TriggersOnline bool `yaml:"triggers-online"`
  55. Triggers []trigger `yaml:"triggers"`
  56. Hosts []hostStruct `yaml:"hosts"`
  57. Ros ros `yaml:"ros"`
  58. Platform platform `yaml:"platform"`
  59. Disk disk `yaml:"disk"`
  60. Monitor MonitorStruct `yaml:"monitor"`
  61. }
  62. var (
  63. CloudConfig cloudConfig
  64. CloudConfigMutex sync.RWMutex
  65. )
  66. // InitCloudConfig 初始化业务配置
  67. func InitCloudConfig() {
  68. c_log.GlobalLogger.Info("初始化OSS配置文件 - 开始。")
  69. // 获取文件的目录
  70. _ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
  71. // 3 ------- 获取 yaml 字符串 -------
  72. cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
  73. // todo 等待时间同步
  74. // 判断文件是否存在。如果不存在则使用默认的
  75. isExist, err := OssBucket.IsObjectExist(cloudConfigObjectKey)
  76. if err != nil {
  77. c_log.GlobalLogger.Errorf("判断配置文件是否存在失败,错误信息为:%v", err)
  78. }
  79. if isExist {
  80. c_log.GlobalLogger.Info("使用机器人自定义配置文件:", cloudConfigObjectKey)
  81. } else {
  82. cloudConfigObjectKey = LocalConfig.OssBasePrefix + LocalConfig.CloudConfigFilename // 默认配置文件路径
  83. c_log.GlobalLogger.Info("使用机器人默认配置文件:", cloudConfigObjectKey)
  84. }
  85. for {
  86. OssMutex.Lock()
  87. err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
  88. OssMutex.Unlock()
  89. if err != nil {
  90. c_log.GlobalLogger.Error("下载oss上的配置文件 "+cloudConfigObjectKey+" 失败。网络授时未完成或者未配置阿里云网络映射到/etc/hosts:", err)
  91. time.Sleep(time.Duration(2) * time.Second)
  92. continue
  93. }
  94. break
  95. }
  96. content, err := os.ReadFile(LocalConfig.CloudConfigLocalPath)
  97. if err != nil {
  98. c_log.GlobalLogger.Error("程序退出。配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
  99. os.Exit(-1)
  100. }
  101. // 4 ------- 解析YAML内容 -------
  102. var newCloudConfig cloudConfig
  103. err = yaml.Unmarshal(content, &newCloudConfig)
  104. if err != nil {
  105. c_log.GlobalLogger.Error("程序退出。配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
  106. os.Exit(-1)
  107. }
  108. // 5 ------- 校验 yaml -------
  109. if checkConfig(newCloudConfig) {
  110. CloudConfigMutex.RLock()
  111. CloudConfig = newCloudConfig
  112. CloudConfigMutex.RUnlock()
  113. } else {
  114. c_log.GlobalLogger.Error("配置文件格式错误:", newCloudConfig)
  115. os.Exit(-1)
  116. }
  117. c_log.GlobalLogger.Info("初始化OSS配置文件 - 成功。")
  118. util.CreateDir(CloudConfig.BagDataDir)
  119. util.CreateDir(CloudConfig.BagCopyDir)
  120. _ = util.RemoveSubFiles(CloudConfig.BagDataDir)
  121. if CloudConfig.CleanBeforeStart { // 判断是否需要清空原数据缓存
  122. if LocalConfig.Node.Name == "node1" { // 只有cloud配置了同时节点为node1才清除
  123. // 清空 timeToLabel.json
  124. timeToLabelJson, _ := util.MapToJsonString(map[string]interface{}{"time": "label"})
  125. _ = util.WriteFile(timeToLabelJson, CloudConfig.TimeToLabelJsonPath)
  126. c_log.GlobalLogger.Info("清空timeToLabel.json - 成功。")
  127. }
  128. _ = util.RemoveSubFiles(CloudConfig.BagCopyDir)
  129. }
  130. }
  131. // RefreshCloudConfig 刷新配置文件
  132. func refreshCloudConfig() {
  133. // 获取文件的目录
  134. _ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
  135. // 3 ------- 获取 yaml 字符串 -------
  136. var content []byte
  137. cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
  138. // 判断文件是否存在。如果不存在则使用默认的
  139. isExist, err := OssBucket.IsObjectExist(cloudConfigObjectKey)
  140. if err != nil {
  141. c_log.GlobalLogger.Errorf("判断配置文件是否存在失败,错误信息为:%v", err)
  142. }
  143. if !isExist {
  144. cloudConfigObjectKey = LocalConfig.OssBasePrefix + LocalConfig.CloudConfigFilename // 默认配置文件路径
  145. }
  146. OssMutex.Lock()
  147. err = OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
  148. OssMutex.Unlock()
  149. if err != nil {
  150. c_log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
  151. return
  152. }
  153. content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
  154. if err != nil {
  155. c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
  156. return
  157. }
  158. // 4 ------- 解析YAML内容 -------
  159. var newCloudConfig cloudConfig
  160. err = yaml.Unmarshal(content, &newCloudConfig)
  161. if err != nil {
  162. c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
  163. return
  164. }
  165. // 5 ------- 校验 yaml -------
  166. if checkConfig(newCloudConfig) {
  167. CloudConfigMutex.RLock()
  168. CloudConfig = newCloudConfig
  169. CloudConfigMutex.RUnlock()
  170. } else {
  171. c_log.GlobalLogger.Error("配置文件格式错误:", newCloudConfig)
  172. return
  173. }
  174. util.CreateDir(CloudConfig.BagDataDir)
  175. util.CreateDir(CloudConfig.BagCopyDir)
  176. }
  177. // RefreshCloudConfig 轮询oss上的配置文件更新到本地
  178. func RefreshCloudConfig() {
  179. for {
  180. time.Sleep(time.Duration(CloudConfig.ConfigRefreshInterval) * time.Second)
  181. refreshCloudConfig()
  182. }
  183. }
  184. // CheckConfig 校验 cfg.yaml 文件
  185. func checkConfig(check cloudConfig) bool {
  186. if len(check.Hosts) != 2 {
  187. c_log.GlobalLogger.Error("cloud-config.yaml中配置的hosts必须为2。")
  188. os.Exit(-1)
  189. }
  190. return true
  191. }