c_cloud.go 5.0 KB

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