cicv_data_closed_loop/aarch64/pjibot_delivery/control/main.go

package main

import (
	commonConfig "cicv-data-closedloop/aarch64/pjibot_delivery/common/config"
	commonService "cicv-data-closedloop/aarch64/pjibot_delivery/common/service"
	"cicv-data-closedloop/aarch64/pjibot_delivery/common/variable"
	"cicv-data-closedloop/aarch64/pjibot_delivery/control/pkg"
	"cicv-data-closedloop/common/config/c_log"
	"cicv-data-closedloop/common/entity"
	"cicv-data-closedloop/common/util"
	commonUtil "cicv-data-closedloop/common/util"
	"encoding/json"
	"fmt"
	"net/rpc"
	"os"
	"runtime"
	"time"
)

var (
	applicationName  = "pji-control"
	localStatus      = "idle"
	cloudStatus      = "NONE"
	lastLocalStatus  = "idle"
	lastCloudStatus  = "NONE"
	timeAllowedFlag  = false
	limitReachedFlag = false
	localTurnLength  = 1  // s，本地状态刷新时间
	cloudTurnLength  = 60 // s，云端状态刷新时间
	renewTurnLength  = 3  // s，续约状态刷新时间
	waitStopLength   = 1  // min，停止master前等待时间
	checkLimitLength = 1  // min, 查询是否达到采集限制等待时间
	launchedFlag     = false
	renewedFlag      = false
	renewTimer       *time.Timer // 续约定时器
	RenewDur         = 5         // min， 续约时间
	maxRetryCount    = 10        // 查询配置最大重试次数
)

func init() {
	runtime.GOMAXPROCS(1)
	// 初始化日志配置
	c_log.InitLog(variable.LogDir, applicationName)
	// 初始化本地配置文件（第1处配置，在本地文件）
	commonConfig.InitLocalConfig(variable.LocalConfigPath)
	// 初始化Oss连接信息
	commonConfig.InitOssConfig()
	// 初始化业务逻辑配置信息，配置文件在oss上（第2处配置，在oss文件）
	commonConfig.InitCloudConfig()
	// 初始化websocket配置
	commonConfig.InitWebsocketConfig()
}

func isTimeAllowed(currentTime time.Time) bool {
	cw := commonConfig.CloudConfig.CollectWindow
	if cw.Flag == 0 { // 关闭固定时间段采集, 则都返回true
		return true
	}
	if len(cw.Days) > 0 { // 如果指定了周几
		currentDay := currentTime.Weekday().String()
		included := false
		for _, day := range cw.Days {
			if day == currentDay {
				included = true
				break
			}
		}
		if !included {
			return false
		}
	}

	start := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), cw.StartTime.Hour(), cw.StartTime.Minute(), cw.StartTime.Second(), 0, currentTime.Location())
	end := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), cw.EndTime.Hour(), cw.EndTime.Minute(), cw.EndTime.Second(), 0, currentTime.Location())

	if start.After(end) { // 如果时段跨天
		end = end.AddDate(0, 0, 1)
	}
	return !currentTime.Before(start) && currentTime.Before(end)
}

func checkCollectLimit() {
	for {
		time.Sleep(time.Duration(checkLimitLength) * time.Minute)
		// 如果开启了采集频率限制，则云端判断采集数量是否超过限额
		if commonConfig.CloudConfig.CollectLimit.Flag == 1 {
			//c_log.GlobalLogger.Error("当前设备已开启数采频率限制，需判断采集数量是否达到限额。")
			responseString, err := commonUtil.HttpPostJsonWithHeaders(
				commonConfig.CloudConfig.CollectLimit.Url,
				map[string]string{"Authorization": "U9yKpD6kZZDDe4LFKK6myAxBUT1XRrDM"},
				map[string]string{
					"snCode":            commonConfig.LocalConfig.SecretKey,
					"collectLimitDay":   util.ToString(commonConfig.CloudConfig.CollectLimit.Day),
					"collectLimitWeek":  util.ToString(commonConfig.CloudConfig.CollectLimit.Week),
					"collectLimitMonth": util.ToString(commonConfig.CloudConfig.CollectLimit.Month),
					"collectLimitYear":  util.ToString(commonConfig.CloudConfig.CollectLimit.Year),
				},
			)
			if err != nil {
				c_log.GlobalLogger.Error("发送http请求获取是否允许采集失败：", err)
				continue
			}
			// 解析JSON字符串到Response结构体
			var resp entity.Response
			err = json.Unmarshal([]byte(responseString), &resp)
			if err != nil {
				c_log.GlobalLogger.Error("解析是否允许采集接口返回结果失败：", err)
				continue
			}
			if resp.Code != 200 { // 不是200 代表采集数量已超过限额不允许采集
				//c_log.GlobalLogger.Info("采集数量已超过限额，当前周期内不再采集。", resp.Code)
				limitReachedFlag = true
			}
		}
		limitReachedFlag = false
	}
}

func initRenew() {
	c_log.GlobalLogger.Info("启动定时器 - 开始。")
	if renewTimer != nil {
		renewTimer.Stop()
	}
	renewedFlag = true
	renewTimer = time.AfterFunc(time.Duration(RenewDur)*time.Minute, func() {
		renewedFlag = false
	})
	c_log.GlobalLogger.Infof("定时时间【%v】分钟 - 成功。", RenewDur)
}

func renew() {
	for {
		time.Sleep(time.Duration(renewTurnLength) * time.Second)
		if localStatus == "running" && launchedFlag && !renewedFlag { // 设备处于运行状态,数采程序已启动，且尚未续约
			c_log.GlobalLogger.Info("设备仍处于运行状态，续约 - 开始。")
			if renewTimer != nil {
				renewTimer.Stop()
			}
			renewedFlag = true
			renewTimer = time.AfterFunc(time.Duration(RenewDur)*time.Minute, func() {
				renewedFlag = false
			})
			c_log.GlobalLogger.Infof("续约时间【%v】分钟 - 成功。", RenewDur)
		}
	}
}

func startMasterNode() {
	c_log.GlobalLogger.Info("获取数据闭环平台最新配置。")

	if commonConfig.CheckPlatformConfigStatus(maxRetryCount) {
		c_log.GlobalLogger.Info("查询到数据闭环平台有配置任务。")
		commonConfig.InitPlatformConfig()

		if _, err := util.ExecuteWithPath(commonConfig.LocalConfig.RestartCmd.Dir, commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args...); err != nil {
			c_log.GlobalLogger.Info("启动新程序失败,【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args, "：", err)
			os.Exit(-1)
		}
		c_log.GlobalLogger.Info("启动任务，本地执行启动命令：【path】=", commonConfig.LocalConfig.RestartCmd.Dir, "【cmd】=", commonConfig.LocalConfig.RestartCmd.Name, commonConfig.LocalConfig.RestartCmd.Args)

		initRenew()
		launchedFlag = true

		c_log.GlobalLogger.Info("数采程序启动 - 成功。")
	} else {
		c_log.GlobalLogger.Error("查询到数据闭环平台没有配置任务，不启动数采程序。")
	}
}

func stopMasterNode() {
	// 发送rpc信号杀死采集程序
	var killArgs commonService.KillSignal
	killArgs = commonService.KillSignal{NodeName: "master", DropUploadData: commonConfig.PlatformConfig.DropUploadData, Restart: false}
	c_log.GlobalLogger.Info("杀死任务，发送rpc结束信号：", killArgs)
	KillRpcClient, err := rpc.Dial("tcp", commonConfig.LocalConfig.Node.Ip+":"+commonConfig.CloudConfig.RpcPort)
	if err != nil {
		// 此处如果连接失败说明采集程序已经停止了
		lastCloudStatus = "NONE"
		c_log.GlobalLogger.Error("采集程序已经停止：", err)
		return
	}

	reply := 0
	if err = KillRpcClient.Call("KillService.Kill", killArgs, &reply); err != nil {
		c_log.GlobalLogger.Error("发送 rpc 请求到 master 报错：", err)
	}

	c_log.GlobalLogger.Info("结束任务后，将数据闭环平台配置置空。")
	commonConfig.PlatformConfig = commonConfig.PlatformConfigStruct{}
	if err = KillRpcClient.Close(); err != nil {
		// 不做处理
	}

	launchedFlag = false
	c_log.GlobalLogger.Info("数采程序关闭 - 成功。")
}

func main() {
	// 更新本地任务状态
	go pkg.GetLocalStatus(&localStatus, &lastLocalStatus, localTurnLength)
	// 更新云端任务状态
	go pkg.GetCloudStatus(&cloudStatus, &lastCloudStatus, cloudTurnLength)

	// 定期检查本地任务状态，执行续约，避免短时间内多次启停
	go renew()

	// 检查是否达到限额
	go checkCollectLimit()

	// 云端任务状态负责更新配置
	go pkg.GetCloudConfig(&cloudStatus, &lastCloudStatus, cloudTurnLength)

	for {
		if launchedFlag { // 当前已启动master节点
			time.Sleep(time.Duration(cloudTurnLength) * time.Second)
		} else {
			time.Sleep(time.Duration(localTurnLength) * time.Second)
		}

		fmt.Println("localStatus: ", localStatus, "lastLocalStatus: ", lastLocalStatus)
		fmt.Println("cloudStatus: ", cloudStatus, "lastCloudStatus: ", lastCloudStatus)

		// 采集时间限制
		currentTime := time.Now()
		timeAllowedFlag = isTimeAllowed(currentTime)

		// 采集频率限制

		// 综合判断 cloudStatus 和 localStatus
		// cloudStatus
		// UN_CHANGE 没有新的任务，无需更改
		// CHANGE 有新的任务，需要杀死旧的数采任务并重启
		// NONE 设备没有配置任务，需要杀死旧的数采任务
		// localStatus
		// idle 空闲状态，此状态下不启动数采任务
		// running 繁忙状态，此状态需要启动数采任务
		// error 错误状态，此状态下不启动数采任务

		// 启停master
		if localStatus == "running" && timeAllowedFlag && !limitReachedFlag {
			if !launchedFlag { // 目前未启动数采程序
				c_log.GlobalLogger.Info("数采程序启动 - 开始。")
				startMasterNode()
			}
		} else if localStatus == "idle" || !timeAllowedFlag || !limitReachedFlag {
			if !renewedFlag && launchedFlag && len(entity.TimeWindowConsumerQueue) == 0 {
				time.Sleep(time.Duration(waitStopLength) * time.Minute)
				c_log.GlobalLogger.Info("数采程序关闭 - 开始。")
				stopMasterNode()
			}
		} else if localStatus == "error" {
			c_log.GlobalLogger.Error("设备运行状态出错，停止数采程序。")
			stopMasterNode()
		} else {
			c_log.GlobalLogger.Error("未知的设备运行状态。【status】=", localStatus)
		}
	}
}