control.go 3.7 KB

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