|
|
@@ -5,11 +5,9 @@ import (
|
|
|
commonService "cicv-data-closedloop/aarch64/jili/common/service"
|
|
|
"cicv-data-closedloop/common/config/c_log"
|
|
|
"cicv-data-closedloop/common/util"
|
|
|
- "fmt"
|
|
|
"net/rpc"
|
|
|
"os"
|
|
|
"runtime"
|
|
|
- "strings"
|
|
|
"time"
|
|
|
)
|
|
|
|
|
|
@@ -23,6 +21,8 @@ func init() {
|
|
|
commonConfig.InitOssConfig()
|
|
|
// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
|
|
|
commonConfig.InitCloudConfig()
|
|
|
+ // 初始化平台任务配置
|
|
|
+ commonConfig.InitPlatformConfig()
|
|
|
}
|
|
|
|
|
|
func main() {
|
|
|
@@ -43,66 +43,50 @@ func RunWithInternet() {
|
|
|
wait = true
|
|
|
|
|
|
// 1. 获取当前设备的任务配置
|
|
|
- platformTasks, err := commonConfig.GetConfig()
|
|
|
- fmt.Println("platformTasks", platformTasks)
|
|
|
- commonConfig.PlatformTasks = platformTasks
|
|
|
+ task, err := commonConfig.GetConfig()
|
|
|
if err != nil {
|
|
|
- c_log.GlobalLogger.Error("获取平台任务失败：", err)
|
|
|
+ c_log.GlobalLogger.Error("获取今日任务信息失败：", err)
|
|
|
continue
|
|
|
}
|
|
|
|
|
|
- // 2. 判断当前时间是否有需要执行的任务
|
|
|
- for _, task := range platformTasks {
|
|
|
- flag := shouldExecuteTask(task.StartTime, task.EndTime)
|
|
|
- fmt.Println("flag", flag)
|
|
|
- // 2.1 当前时间不执行此任务，跳过
|
|
|
- if !flag {
|
|
|
+ // 2 当前时间需要执行此任务
|
|
|
+ // 2.1 判断此任务是否正在执行
|
|
|
+ if commonConfig.CurrTask.TaskConfigId == task.TaskConfigId { // 此任务正在执行，退出循环
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ // 此任务没有在执行，需要启动此任务
|
|
|
+ // 2.2 判断是否有其他任务正在执行
|
|
|
+ if commonConfig.CurrTask.TaskConfigId != "" { // 存在其他任务正在执行，杀死旧的任务
|
|
|
+ c_log.GlobalLogger.Info("需要执行的任务id为", task.TaskConfigId, ", 当前任务id为：", commonConfig.CurrTask.TaskConfigId, ", 需要杀死该任务。")
|
|
|
+ // 杀死旧的任务
|
|
|
+ var killArgs commonService.KillSignal
|
|
|
+ // todo 吉利新平台没有DropUploadData字段（旧平台，已采数据：保留/丢弃）
|
|
|
+ killArgs = commonService.KillSignal{DropUploadData: true, Restart: false}
|
|
|
+ c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号到本地：", killArgs)
|
|
|
+ KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
|
|
|
+ if err != nil {
|
|
|
+ // 此处如果连接失败说明采集程序已经停止了
|
|
|
+ commonConfig.CurrTask = commonConfig.PlatformConfigStruct{}
|
|
|
+ c_log.GlobalLogger.Error("采集程序已经停止：", err)
|
|
|
continue
|
|
|
}
|
|
|
- // 2.2 当前时间需要执行此任务
|
|
|
- // 2.2.1 判断此任务是否正在执行
|
|
|
- if commonConfig.CurrTask.TaskConfigId == task.TaskConfigId { // 此任务正在执行，退出循环
|
|
|
- break
|
|
|
+ reply := 0
|
|
|
+ if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
|
|
|
+ c_log.GlobalLogger.Error("发送rpc请求到master失败：", err)
|
|
|
+ // 这里可能会报错 unexpected EOF 但是不影响，先注释 close 和 continue
|
|
|
+ //KillRpcClient.Close()
|
|
|
+ //continue
|
|
|
}
|
|
|
- // 此任务没有在执行，需要启动此任务
|
|
|
- // 2.2.2 判断是否有其他任务正在执行
|
|
|
- if commonConfig.CurrTask.TaskConfigId != "" { // 存在其他任务正在执行，杀死旧的任务
|
|
|
- c_log.GlobalLogger.Info("需要执行的任务id为", task.TaskConfigId, ", 当前任务id为：", commonConfig.CurrTask.TaskConfigId, ", 需要杀死该任务。")
|
|
|
- // 杀死旧的任务
|
|
|
- var killArgs commonService.KillSignal
|
|
|
- // todo 吉利新平台没有DropUploadData字段（旧平台，已采数据：保留/丢弃）
|
|
|
- killArgs = commonService.KillSignal{DropUploadData: true, Restart: false}
|
|
|
- c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号到本地：", killArgs)
|
|
|
- KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
|
|
|
- if err != nil {
|
|
|
- // 此处如果连接失败说明采集程序已经停止了
|
|
|
- commonConfig.RecordTopics = []string{}
|
|
|
- commonConfig.CurrTask = commonConfig.PlatformConfigStruct{}
|
|
|
- c_log.GlobalLogger.Error("采集程序已经停止：", err)
|
|
|
- continue
|
|
|
- }
|
|
|
- reply := 0
|
|
|
- if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
|
|
|
- c_log.GlobalLogger.Error("发送rpc请求到master失败：", err)
|
|
|
- // 这里可能会报错 unexpected EOF 但是不影响，先注释 close 和 continue
|
|
|
- //KillRpcClient.Close()
|
|
|
- //continue
|
|
|
- }
|
|
|
- c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
|
|
|
- commonConfig.RecordTopics = []string{}
|
|
|
- commonConfig.CurrTask = commonConfig.PlatformConfigStruct{}
|
|
|
- if err = KillRpcClient.Close(); err != nil {
|
|
|
- // 不做处理
|
|
|
- }
|
|
|
+ c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
|
|
|
+ commonConfig.CurrTask = commonConfig.PlatformConfigStruct{}
|
|
|
+ if err = KillRpcClient.Close(); err != nil {
|
|
|
+ // 不做处理
|
|
|
}
|
|
|
-
|
|
|
- // 开启新的任务
|
|
|
- startMasterOrSlave()
|
|
|
- commonConfig.RecordTopics = getUniqueCollectionTopics(task.TaskTriggers)
|
|
|
- commonConfig.CurrTask = task
|
|
|
- // 退出循环
|
|
|
- break
|
|
|
}
|
|
|
+
|
|
|
+ // 开启新的任务
|
|
|
+ startMasterOrSlave()
|
|
|
+ commonConfig.CurrTask = task
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -117,57 +101,3 @@ func startMasterOrSlave() {
|
|
|
}
|
|
|
c_log.GlobalLogger.Info("启动任务，本地执行启动命令：【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)
|
|
|
}
|
|
|
-
|
|
|
-// shouldExecuteTask 给定任务起止时间，判断当前时刻此任务是否应该执行
|
|
|
-func shouldExecuteTask(startTimeStr, endTimeStr string) bool {
|
|
|
- // 定义时间格式（根据输入的时间字符串格式）
|
|
|
- timeFormat := "2006-01-02 15:04:05"
|
|
|
-
|
|
|
- // 注意： time.Parse 时间没有考虑时区，当做默认的 +0 时区 UTC 时间来解析了，就造成了实际时间比预期的 +8 小时
|
|
|
- // 解析 startTime 和 endTime
|
|
|
- startTime, err := time.ParseInLocation(timeFormat, startTimeStr, time.Local)
|
|
|
- //fmt.Println("startTime", startTime)
|
|
|
- if err != nil {
|
|
|
- fmt.Println("Error parsing startTime:", err)
|
|
|
- return false
|
|
|
- }
|
|
|
- endTime, err := time.ParseInLocation(timeFormat, endTimeStr, time.Local)
|
|
|
- //fmt.Println("endTime", endTime)
|
|
|
- if err != nil {
|
|
|
- fmt.Println("Error parsing endTime:", err)
|
|
|
- return false
|
|
|
- }
|
|
|
-
|
|
|
- // 获取当前时间
|
|
|
- currentTime := time.Now()
|
|
|
- //fmt.Println("currentTime", currentTime)
|
|
|
-
|
|
|
- // 判断当前时间是否在 startTime 和 endTime 之间
|
|
|
- return !currentTime.Before(startTime) && !currentTime.After(endTime)
|
|
|
-}
|
|
|
-
|
|
|
-// getUniqueCollectionTopics 统计需要采集的 topic 并去重
|
|
|
-func getUniqueCollectionTopics(taskTriggers []commonConfig.TaskTrigger) []string {
|
|
|
- // 使用 map 去重
|
|
|
- topicMap := make(map[string]struct{})
|
|
|
-
|
|
|
- // 遍历 taskTriggers
|
|
|
- for _, trigger := range taskTriggers {
|
|
|
- // 分割 collectionTopic 字段
|
|
|
- topics := strings.Split(trigger.CollectionTopic, ",")
|
|
|
- for _, topic := range topics {
|
|
|
- trimmedTopic := strings.TrimSpace(topic) // 去除空格
|
|
|
- if trimmedTopic != "" { // 忽略空字符串
|
|
|
- topicMap[trimmedTopic] = struct{}{}
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // 将 map 中的 key 转换为切片
|
|
|
- var uniqueTopics []string
|
|
|
- for topic := range topicMap {
|
|
|
- uniqueTopics = append(uniqueTopics, topic)
|
|
|
- }
|
|
|
-
|
|
|
- return uniqueTopics
|
|
|
-}
|
