control.go 3.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. package main
  2. import (
  3. "cicv-data-closedloop/common/config/c_log"
  4. "cicv-data-closedloop/common/util"
  5. commonConfig "cicv-data-closedloop/pji/common/cfg"
  6. commonService "cicv-data-closedloop/pji/common/svc"
  7. controlConfig "cicv-data-closedloop/pji/control/package/cfg"
  8. "net/rpc"
  9. "os"
  10. "runtime"
  11. "time"
  12. )
  13. func init() {
  14. runtime.GOMAXPROCS(1)
  15. // 初始化日志配置
  16. c_log.InitLog("pji-control")
  17. // 初始化本地配置文件(第1处配置,在本地文件)
  18. commonConfig.InitLocalConfig()
  19. // 初始化Oss连接信息
  20. commonConfig.InitOssConfig()
  21. // 初始化业务逻辑配置信息,配置文件在oss上(第2处配置,在oss文件)
  22. commonConfig.InitCloudConfig()
  23. // 初始化rpc客户端,用于杀死旧的采集程序
  24. controlConfig.InitNacos()
  25. }
  26. func main() {
  27. lastStatus := "NONE"
  28. // 轮询任务接口判断是否有更新
  29. for {
  30. time.Sleep(time.Duration(2) * time.Second)
  31. // 1 获取当前设备的任务的 status
  32. status, err := commonConfig.GetStatus(commonConfig.PlatformConfig.TaskConfigId)
  33. if err != nil {
  34. c_log.GlobalLogger.Error("获取配置status失败:", err)
  35. continue
  36. }
  37. // 2 判断 status
  38. // UN_CHANGE 没有新的任务,无需更改
  39. // CHANGE 有新的任务,需要杀死旧的任务并重启
  40. // NONE 设备没有配置任务,需要杀死旧的任务
  41. if status == "UN_CHANGE" {
  42. lastStatus = "UN_CHANGE"
  43. continue
  44. } else if status == "CHANGE" || status == "NONE" {
  45. if lastStatus == "NONE" && status == "NONE" {
  46. continue
  47. }
  48. // 3 发送rpc信号杀死采集程序
  49. if lastStatus == "NONE" && status == "CHANGE" {
  50. if _, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...); err != nil {
  51. c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, ":", err)
  52. os.Exit(-1)
  53. }
  54. c_log.GlobalLogger.Info("启动任务,本地执行启动命令:【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)
  55. lastStatus = status
  56. c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
  57. commonConfig.InitPlatformConfig()
  58. continue
  59. }
  60. var killArgs *commonService.KillSignal
  61. if lastStatus == "UN_CHANGE" && status == "CHANGE" {
  62. killArgs = &commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: true}
  63. c_log.GlobalLogger.Info("更新任务,发送rpc重启信号:", killArgs)
  64. }
  65. if lastStatus == "UN_CHANGE" && status == "NONE" {
  66. killArgs = &commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
  67. c_log.GlobalLogger.Info("杀死任务,发送rpc结束信号:", killArgs)
  68. }
  69. KillRpcClient, err := rpc.Dial("tcp", "localhost:"+commonConfig.CloudConfig.RpcPort)
  70. if err != nil {
  71. c_log.GlobalLogger.Error("创建rpc客户端连接master失败:", err)
  72. // 此处关闭client会报错
  73. continue
  74. }
  75. reply := 0
  76. if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
  77. c_log.GlobalLogger.Error("发送 rpc 请求到 master 报错:", err)
  78. //TODO 这里可能会报错unexpected EOF但是不影响,先注释 close 和 continue
  79. //KillRpcClient.Close()
  80. //continue
  81. }
  82. lastStatus = status
  83. c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")
  84. commonConfig.InitPlatformConfig()
  85. KillRpcClient.Close()
  86. } else {
  87. c_log.GlobalLogger.Error("未知的采集任务状态。【status】=", status)
  88. }
  89. }
  90. }