c_cloud.go 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173
  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 platform struct {
  11. UrlDeviceAuth string `yaml:"url-device-auth"`
  12. UrlTaskPoll string `yaml:"url-task-poll"`
  13. UrlTask string `yaml:"url-task"`
  14. }
  15. type rosbagStruct struct {
  16. Path string `yaml:"path"`
  17. Envs []string `yaml:"envs"`
  18. }
  19. type hostStruct struct {
  20. Name string `yaml:"name"`
  21. Ip string `yaml:"ip"`
  22. Topics []string `yaml:"topics"`
  23. Rosbag rosbagStruct `yaml:"rosbag"`
  24. }
  25. type ros struct {
  26. MasterAddress string `yaml:"master-address"`
  27. Nodes []string `yaml:"nodes"`
  28. }
  29. type disk struct {
  30. Name string `yaml:"name"`
  31. Used uint64 `yaml:"used"`
  32. }
  33. type trigger struct {
  34. Label string `yaml:"label"`
  35. Topics []string `yaml:"topics"`
  36. }
  37. type cloudConfig struct {
  38. FullCollect bool `yaml:"full-collect"`
  39. ConfigRefreshInterval int `yaml:"config-refresh-interval"` // 配置刷新时间间隔
  40. BagNumber int `yaml:"bag-number"`
  41. TimeWindowSendGap int `yaml:"time-window-send-gap"` // 主节点向从节点发送窗口的最小时间间隔
  42. TimeToLabelJsonPath string `yaml:"time-to-label-json-path"`
  43. BagDataDir string `yaml:"bag-data-dir"`
  44. BagCopyDir string `yaml:"bag-copy-dir"`
  45. TriggersDir string `yaml:"triggers-dir"`
  46. TcpPort string `yaml:"tcp-port"`
  47. RpcPort string `yaml:"rpc-port"`
  48. Triggers []trigger `yaml:"triggers"`
  49. Hosts []hostStruct `yaml:"hosts"`
  50. Ros ros `yaml:"ros"`
  51. Platform platform `yaml:"platform"`
  52. Disk disk `yaml:"disk"`
  53. }
  54. var (
  55. CloudConfig cloudConfig
  56. CloudConfigMutex sync.RWMutex
  57. )
  58. // InitCloudConfig 初始化业务配置
  59. func InitCloudConfig() {
  60. c_log.GlobalLogger.Info("初始化OSS配置文件 - 开始。")
  61. // 获取文件的目录
  62. _ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
  63. // 3 获取 yaml 字符串
  64. var content []byte
  65. cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
  66. // todo 等待时间同步
  67. for {
  68. OssMutex.Lock()
  69. err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
  70. OssMutex.Unlock()
  71. if err != nil {
  72. c_log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
  73. continue
  74. }
  75. break
  76. }
  77. content, err := os.ReadFile(LocalConfig.CloudConfigLocalPath)
  78. if err != nil {
  79. c_log.GlobalLogger.Error("程序退出。配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
  80. os.Exit(-1)
  81. }
  82. // 4 ------- 解析YAML内容 -------
  83. var newCloudConfig cloudConfig
  84. err = yaml.Unmarshal(content, &newCloudConfig)
  85. if err != nil {
  86. c_log.GlobalLogger.Error("程序退出。配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
  87. os.Exit(-1)
  88. }
  89. // 5 ------- 校验 yaml -------
  90. if checkConfig(newCloudConfig) {
  91. CloudConfigMutex.RLock()
  92. CloudConfig = newCloudConfig
  93. CloudConfigMutex.RUnlock()
  94. } else {
  95. c_log.GlobalLogger.Error("配置文件格式错误:", newCloudConfig)
  96. os.Exit(-1)
  97. }
  98. c_log.GlobalLogger.Info("初始化OSS配置文件 - 成功。")
  99. _ = util.CreateDir(CloudConfig.BagDataDir)
  100. _ = util.CreateDir(CloudConfig.BagCopyDir)
  101. timeToLabelJson, _ := util.MapToJsonString(map[string]interface{}{"time": "label"})
  102. _ = util.WriteFile(timeToLabelJson, CloudConfig.TimeToLabelJsonPath)
  103. }
  104. // RefreshCloudConfig 初始化业务配置
  105. func refreshCloudConfig() {
  106. // 获取文件的目录
  107. _ = util.CreateParentDir(LocalConfig.CloudConfigLocalPath)
  108. // 3 ------- 获取 yaml 字符串 -------
  109. var content []byte
  110. cloudConfigObjectKey := LocalConfig.OssBasePrefix + LocalConfig.EquipmentNo + "/" + LocalConfig.CloudConfigFilename
  111. OssMutex.Lock()
  112. err := OssBucket.GetObjectToFile(cloudConfigObjectKey, LocalConfig.CloudConfigLocalPath)
  113. OssMutex.Unlock()
  114. if err != nil {
  115. c_log.GlobalLogger.Error("下载oss上的配置文件"+cloudConfigObjectKey+"失败。", err)
  116. return
  117. }
  118. content, err = os.ReadFile(LocalConfig.CloudConfigLocalPath)
  119. if err != nil {
  120. c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 读取失败:", err)
  121. return
  122. }
  123. // 4 ------- 解析YAML内容 -------
  124. var newCloudConfig cloudConfig
  125. err = yaml.Unmarshal(content, &newCloudConfig)
  126. if err != nil {
  127. c_log.GlobalLogger.Error("配置文件 ", LocalConfig.CloudConfigLocalPath, " 解析失败:", err)
  128. return
  129. }
  130. // 5 ------- 校验 yaml -------
  131. if checkConfig(newCloudConfig) {
  132. CloudConfigMutex.RLock()
  133. CloudConfig = newCloudConfig
  134. CloudConfigMutex.RUnlock()
  135. } else {
  136. c_log.GlobalLogger.Error("配置文件格式错误:", newCloudConfig)
  137. return
  138. }
  139. util.CreateDir(CloudConfig.BagDataDir)
  140. util.CreateDir(CloudConfig.BagCopyDir)
  141. }
  142. // RefreshCloudConfig 轮询oss上的配置文件更新到本地
  143. func RefreshCloudConfig() {
  144. for {
  145. time.Sleep(time.Duration(CloudConfig.ConfigRefreshInterval) * time.Second)
  146. refreshCloudConfig()
  147. }
  148. }
  149. // CheckConfig 校验 cfg.yaml 文件
  150. func checkConfig(check cloudConfig) bool {
  151. if len(check.Hosts) != 2 {
  152. c_log.GlobalLogger.Error("cloud-config.yaml中配置的hosts必须为2。")
  153. os.Exit(-1)
  154. }
  155. return true
  156. }