ipole_edge_server/edge/ipole/healthcheck.go

package main

import (
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"time"

	"lc/common/protocol"
	"lc/common/util"
)

const (
	phase1Timeout = 30 * time.Second
	phase2Delay   = 5 * time.Minute
	phase2Window  = 5 * time.Minute
)

// HealthCheckRunner 健康检测运行器
type HealthCheckRunner struct {
	marker     *DeployMarker
	running    int32
	phase2Done int32
}

var _hcRunner *HealthCheckRunner

// StartHealthCheckIfNeeded 启动时检查部署标记，如有则运行健康检测
func StartHealthCheckIfNeeded() {
	markerPath := filepath.Join(util.GetPath(0), "deploy_marker.json")
	data, err := os.ReadFile(markerPath)
	if err != nil {
		// 检查是否有回滚标记
		rollbackPath := filepath.Join(util.GetPath(0), "rollback_marker.json")
		if rbData, err := os.ReadFile(rollbackPath); err == nil {
			var marker DeployMarker
			json.Unmarshal(rbData, &marker)
			util.GetTagLog().Infof("sys", "检测到回滚标记，已回滚到版本:%s", marker.Version)
			reportRollback(marker)
			os.Remove(rollbackPath)
		}
		return
	}

	var marker DeployMarker
	if err := json.Unmarshal(data, &marker); err != nil {
		util.GetTagLog().Errorf("sys", "解析部署标记失败,err=%v", err)
		return
	}

	util.GetTagLog().Infof("sys", "检测到部署标记，开始健康检测 version=%s", marker.Version)

	_hcRunner = &HealthCheckRunner{
		marker: &marker,
	}

	go func() {
		time.Sleep(3 * time.Second) // 等待 MQTT 初始化
		_hcRunner.runPhase1()
	}()
}

func (o *HealthCheckRunner) runPhase1() {
	if !atomic.CompareAndSwapInt32(&o.running, 0, 1) {
		return
	}
	defer atomic.StoreInt32(&o.running, 0)

	util.GetTagLog().Infof("sys", "HealthCheck:Phase 1 开始 version=%s", o.marker.Version)
	deadline := time.Now().Add(phase1Timeout)

	checks := []protocol.HealthCheck{
		o.checkMQTT(deadline),
		o.checkRedis(deadline),
	}

	allOK := true
	for _, c := range checks {
		if c.Status != "ok" {
			allOK = false
			break
		}
	}

	o.reportPhase(1, allOK, checks)

	if !allOK {
		util.GetTagLog().Errorf("sys", "HealthCheck:Phase 1 失败，开始回滚")
		o.rollback(checks)
		return
	}

	util.GetTagLog().Infof("sys", "HealthCheck:Phase 1 全部通过，清理部署标记")
	os.Remove(filepath.Join(util.GetPath(0), "deploy_marker.json"))
	DeployResultSave(o.marker.Version, 1, allOK, checks)

	go func() {
		time.Sleep(phase2Delay)
		o.runPhase2()
	}()
}

func (o *HealthCheckRunner) checkMQTT(deadline time.Time) protocol.HealthCheck {
	start := time.Now()
	check := protocol.HealthCheck{Name: "mqtt"}

	for time.Now().Before(deadline) {
		mgr := GetMQTTMgr()
		if mgr.Cloud != nil && mgr.Cloud.IsConnected() {
			check.Status = "ok"
			check.Detail = "MQTT Broker 已连接"
			check.DurationMs = int(time.Since(start).Milliseconds())
			util.GetTagLog().Infof("sys", "HealthCheck:MQTT检测通过")
			return check
		}
		time.Sleep(1 * time.Second)
	}

	check.Status = "timeout"
	check.Detail = "MQTT Broker 不可达，请检查 Server/User/Password 配置、网络连通性"
	check.DurationMs = int(time.Since(start).Milliseconds())
	util.GetTagLog().Errorf("sys", "HealthCheck:MQTT检测失败")
	return check
}

func (o *HealthCheckRunner) checkRedis(deadline time.Time) protocol.HealthCheck {
	start := time.Now()
	check := protocol.HealthCheck{Name: "redis"}

	for time.Now().Before(deadline) {
		if redisEdgeData != nil {
			if _, err := redisEdgeData.Ping().Result(); err == nil {
				check.Status = "ok"
				check.Detail = "Redis PING 正常"
				check.DurationMs = int(time.Since(start).Milliseconds())
				util.GetTagLog().Infof("sys", "HealthCheck:Redis检测通过")
				return check
			}
		}
		time.Sleep(2 * time.Second)
	}

	check.Status = "timeout"
	check.Detail = "Redis 连接失败，请检查 Redis Server/Password 配置"
	check.DurationMs = int(time.Since(start).Milliseconds())
	util.GetTagLog().Errorf("sys", "HealthCheck:Redis检测失败")
	return check
}

func (o *HealthCheckRunner) runPhase2() {
	atomic.StoreInt32(&o.phase2Done, 1)
	util.GetTagLog().Infof("sys", "HealthCheck:Phase 2 开始 version=%s", o.marker.Version)

	checks := []protocol.HealthCheck{
		o.checkSerial(),
		o.checkModbusData(),
	}

	allOK := true
	for _, c := range checks {
		if c.Status != "ok" {
			allOK = false
			break
		}
	}

	o.reportPhase(2, allOK, checks)
	DeployResultSave(o.marker.Version, 2, allOK, checks)

	if !allOK {
		util.GetTagLog().Warnf("sys", "HealthCheck:Phase 2 存在告警，不回滚")
	} else {
		util.GetTagLog().Infof("sys", "HealthCheck:Phase 2 全部通过")
	}
}

func (o *HealthCheckRunner) checkSerial() protocol.HealthCheck {
	start := time.Now()
	check := protocol.HealthCheck{Name: "serial"}

	sc := GetSerialMgr()
	failedPorts := sc.GetFailedPorts()
	if len(failedPorts) == 0 {
		check.Status = "ok"
		check.Detail = "所有串口打开正常"
	} else {
		check.Status = "fail"
		parts := make([]string, len(failedPorts))
		for i, p := range failedPorts {
			parts[i] = fmt.Sprintf("%d", p)
		}
		check.Detail = "串口打开失败: " + strings.Join(parts, ", ")
	}
	check.DurationMs = int(time.Since(start).Milliseconds())
	return check
}

func (o *HealthCheckRunner) checkModbusData() protocol.HealthCheck {
	start := time.Now()
	check := protocol.HealthCheck{Name: "modbus"}

	deviceCount := 0
	hasData := false
	mapRtuUploadManager.Range(func(key, value interface{}) bool {
		mgr, ok := value.(*RtuUploadManager)
		if !ok {
			return true
		}
		deviceCount++
		mgr.DataLock.Lock()
		lastDataTime := mgr.Datatime
		mgr.DataLock.Unlock()
		if time.Since(lastDataTime) < phase2Window {
			hasData = true
			return false
		}
		return true
	})

	if deviceCount == 0 {
		check.Status = "ok"
		check.Detail = fmt.Sprintf("无Modbus设备配置，跳过采集检测")
	} else if hasData {
		check.Status = "ok"
		check.Detail = fmt.Sprintf("设备数据采集正常（%d台设备）", deviceCount)
	} else {
		check.Status = "fail"
		check.Detail = fmt.Sprintf("%d台设备5分钟内无数据上报，请检查设备配置和物模型是否正确加载", deviceCount)
	}
	check.DurationMs = int(time.Since(start).Milliseconds())
	return check
}

func (o *HealthCheckRunner) reportPhase(phase int, success bool, checks []protocol.HealthCheck) {
	var obj protocol.Pack_HealthCheck
	seq := GetNextUint64()
	if str, err := obj.EnCode(appConfig.GID, appConfig.GID, seq, o.marker.Version, phase, success, checks); err == nil {
		topic := GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_DEPLOY_HEALTH)
		GetMQTTMgr().Publish(topic, str, 0, ToCloud)
		util.GetTagLog().Infof("sys", "HealthCheck:Phase %d 结果已上报 success=%v", phase, success)
	}
}

func (o *HealthCheckRunner) rollback(checks []protocol.HealthCheck) {
	cwd, _ := os.Getwd()
	targetPath := filepath.Join(cwd, appname)
	backupPath := targetPath + ".bak"

	rbMarker := DeployMarker{
		Version:   o.marker.Version,
		Timestamp: time.Now().Unix(),
		Action:    "rollback",
	}
	rbContent, _ := json.MarshalToString(rbMarker)
	os.WriteFile(filepath.Join(util.GetPath(0), "rollback_marker.json"), []byte(rbContent), os.ModePerm)

	os.Remove(filepath.Join(util.GetPath(0), "deploy_marker.json"))

	DeployResultSave(o.marker.Version, 1, false, checks)

	if _, err := os.Stat(backupPath); os.IsNotExist(err) {
		util.GetTagLog().Errorf("sys", "HealthCheck:备份文件不存在，无法自动回滚")
		return
	}

	if err := os.Rename(backupPath, targetPath); err != nil {
		util.GetTagLog().Errorf("sys", "HealthCheck:回滚失败,err=%v", err)
		return
	}
	os.Chmod(targetPath, 0755)
	util.GetTagLog().Infof("sys", "HealthCheck:回滚成功，退出进程")

	time.Sleep(500 * time.Millisecond)
	os.Exit(2)
}

func reportRollback(marker DeployMarker) {
	var obj protocol.Pack_DeployAck
	seq := GetNextUint64()
	if str, err := obj.EnCode(appConfig.GID, appConfig.GID, seq, marker.Version, false,
		"已自动回滚到版本 "+marker.Version); err == nil {
		topic := GetTopic(protocol.DT_GATEWAY, appConfig.GID, protocol.TP_GW_DEPLOY_ACK)
		GetMQTTMgr().Publish(topic, str, 0, ToCloud)
	}
}

// ---- 部署结果持久化 ----

type DeployResult struct {
	Version   string             `json:"version"`
	Timestamp int64              `json:"timestamp"`
	Phase1    *DeployPhaseResult `json:"phase1,omitempty"`
	Phase2    *DeployPhaseResult `json:"phase2,omitempty"`
}

type DeployPhaseResult struct {
	Success bool                   `json:"success"`
	Checks  []protocol.HealthCheck `json:"checks"`
}

var _deployResult *DeployResult

func DeployResultSave(version string, phase int, success bool, checks []protocol.HealthCheck) {
	if _deployResult == nil || _deployResult.Version != version {
		_deployResult = &DeployResult{
			Version:   version,
			Timestamp: time.Now().Unix(),
		}
	}
	pr := &DeployPhaseResult{Success: success, Checks: checks}
	if phase == 1 {
		_deployResult.Phase1 = pr
	} else {
		_deployResult.Phase2 = pr
	}

	content, _ := json.MarshalToString(_deployResult)
	resultPath := filepath.Join(util.GetPath(0), "deploy_result.json")
	os.WriteFile(resultPath, []byte(content), os.ModePerm)
}

func DeployResultGet() *DeployResult {
	if _deployResult != nil {
		return _deployResult
	}
	resultPath := filepath.Join(util.GetPath(0), "deploy_result.json")
	data, err := os.ReadFile(resultPath)
	if err != nil {
		return nil
	}
	var result DeployResult
	if err := json.Unmarshal(data, &result); err != nil {
		return nil
	}
	_deployResult = &result
	return &result
}