初始化项目
This commit is contained in:
BBIT-Kai
@@ -0,0 +1,174 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sentinel/pkg/config"
|
||||
"sentinel/pkg/device"
|
||||
"sentinel/pkg/docker"
|
||||
"sentinel/pkg/log"
|
||||
model2 "sentinel/pkg/model"
|
||||
"sentinel/pkg/utils"
|
||||
"time"
|
||||
)
|
||||
|
||||
type BusinessService struct {
|
||||
mqtt *MQTTService
|
||||
dockerManager docker.DockerManager
|
||||
deviceID string
|
||||
deptId string
|
||||
cmdTopic string
|
||||
deviceType string
|
||||
}
|
||||
|
||||
func NewBusinessService(m *MQTTService, deviceID string, dm docker.DockerManager) *BusinessService {
|
||||
return &BusinessService{
|
||||
mqtt: m,
|
||||
dockerManager: dm,
|
||||
deviceID: deviceID,
|
||||
}
|
||||
}
|
||||
|
||||
func (b *BusinessService) Start() error {
|
||||
if err := b.mqtt.SubscribeTopic(getInitTopic(b.deviceID), 1); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
b.mqtt.SetMessageHandler(b.onMQTTMessage)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func getInitTopic(deviceID string) string {
|
||||
return "+/+/+/" + deviceID + "/#"
|
||||
}
|
||||
|
||||
func (b *BusinessService) getOwnTopic(deviceID string) string {
|
||||
return b.deptId + "/cmd/" + b.deviceType + "/" + deviceID + "/#"
|
||||
}
|
||||
|
||||
// 消息处理
|
||||
func (b *BusinessService) onMQTTMessage(topic string, payload []byte) {
|
||||
model := model2.FromStringToMqttTopic(topic)
|
||||
payload_json, err := utils.PayloadToMap(payload)
|
||||
if err != nil {
|
||||
fmt.Println("解析失败:", err)
|
||||
return
|
||||
}
|
||||
|
||||
// 指令
|
||||
if model.Domain == "cmd" && model.DeviceType == b.deviceType {
|
||||
log.Println("收到指令：", model.Resource)
|
||||
switch model.Resource {
|
||||
case "shutdown":
|
||||
payload_json["massage"] = b.handleShutdown()
|
||||
case "restart":
|
||||
payload_json["massage"] = b.handleRestart()
|
||||
case "check_update":
|
||||
payload_json["massage"] = b.handleCheckUpdate()
|
||||
default:
|
||||
log.Println("未知的命令：", model.Resource)
|
||||
payload_json["massage"] = "未知的命令"
|
||||
}
|
||||
topic := "x/receipt/x/" + device.GetDeviceID() + "/info"
|
||||
payload := utils.JSONToString(payload_json)
|
||||
b.mqtt.PublicMsg(topic, payload, false) // 回执
|
||||
} else if model.Domain == "status" && model.Resource == "receipt" {
|
||||
b.deviceType = model.DeviceType
|
||||
b.deptId = model.DeptId
|
||||
// 取消订阅之前的初始化主题
|
||||
if b.mqtt.UnsubscribeTopic(getInitTopic(b.deviceID)) != nil {
|
||||
log.Error("无法取消初始化主题")
|
||||
return
|
||||
}
|
||||
// 新订阅属于自己的主题
|
||||
if b.mqtt.SubscribeTopic(b.getOwnTopic(b.deviceID), 1) != nil {
|
||||
log.Error("无法定于属于自己的主题")
|
||||
return
|
||||
}
|
||||
log.Println("设备初始化成功：所属项目：", model.DeptId, "\t设备类型：", model.DeviceType)
|
||||
}
|
||||
}
|
||||
|
||||
func getStatusInfoTopicPayload(isOnline bool) (string, string) {
|
||||
info := map[string]interface{}{
|
||||
"version": config.APP_VERSION,
|
||||
"online": isOnline,
|
||||
"ip": utils.GetLocalIP(),
|
||||
"hostname": utils.GetHostname(),
|
||||
"mac": utils.GetMacAddress(),
|
||||
"os": utils.GetOSInfo(),
|
||||
"cpu": utils.GetCPUInfo(),
|
||||
"memory_total": utils.GetMemory(),
|
||||
"disk_total": utils.GetDisk(),
|
||||
"last_seen": time.Now().UTC().Format(time.RFC3339),
|
||||
}
|
||||
payloadBytes, _ := json.Marshal(info)
|
||||
payload := string(payloadBytes) // 转成 string
|
||||
return "x/status/x/" + device.GetDeviceID() + "/info", payload
|
||||
}
|
||||
|
||||
// 关闭程序（立即退出）
|
||||
func (b *BusinessService) handleShutdown() string {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
b.dockerManager.StopAndRemoveContainer(ctx)
|
||||
return "程序已退出"
|
||||
}
|
||||
|
||||
// 重启程序
|
||||
// 重启程序
|
||||
func (b *BusinessService) handleRestart() string {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
// 先停止并移除旧容器
|
||||
if err := b.dockerManager.StopAndRemoveContainer(ctx); err != nil {
|
||||
log.Println("警告：移除容器失败，可能容器不存在：", err)
|
||||
}
|
||||
|
||||
// 拉取最新镜像
|
||||
log.Println("正在拉取镜像：", config.DOCKER_IMAGE)
|
||||
if err := b.dockerManager.PullImage(ctx, config.DOCKER_IMAGE); err != nil {
|
||||
log.Println("镜像拉取失败：", err)
|
||||
return "程序重启失败，拉取镜像失败"
|
||||
}
|
||||
|
||||
// 启动新容器
|
||||
if err := b.dockerManager.RunContainer(ctx); err != nil {
|
||||
log.Println("容器启动失败：", err)
|
||||
return "程序重启失败，容器启动失败"
|
||||
}
|
||||
|
||||
return "程序已重启"
|
||||
}
|
||||
|
||||
func (b *BusinessService) handleCheckUpdate() string {
|
||||
cfg := config.LoadConfig()
|
||||
cfg.NeedUpdate = true
|
||||
if err := config.WriteLocalConfig(&cfg); err != nil {
|
||||
log.Println("load config failed:", err)
|
||||
}
|
||||
return "程序更新机制已触发"
|
||||
}
|
||||
|
||||
//func (b *BusinessService) handleCheckUpdate() {
|
||||
// args := []string{
|
||||
// "--version", strconv.Itoa(config.APP_VERSION),
|
||||
// }
|
||||
//
|
||||
// launcher := newUpdaterLauncher()
|
||||
//
|
||||
// if err := launcher.Start(args); err != nil {
|
||||
// log.Println("[BUS] failed to start updater:", err)
|
||||
// return
|
||||
// }
|
||||
//
|
||||
// log.Println(
|
||||
// "[BUS] updater started, exiting main program",
|
||||
// )
|
||||
//
|
||||
// time.Sleep(500 * time.Millisecond)
|
||||
// os.Exit(0)
|
||||
//}
|
||||
+122
@@ -0,0 +1,122 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"sentinel/pkg/config"
|
||||
"sentinel/pkg/device"
|
||||
"sentinel/pkg/docker"
|
||||
"sentinel/pkg/log"
|
||||
)
|
||||
|
||||
func main() {
|
||||
deviceID := device.GetDeviceID()
|
||||
|
||||
// APP Logo
|
||||
InitPrint(deviceID)
|
||||
// Config
|
||||
InitConfigFile()
|
||||
// MQTT
|
||||
InitMQTT(deviceID)
|
||||
|
||||
}
|
||||
func InitConfigFile() {
|
||||
// 第一次写入默认配置
|
||||
cfg := config.AppConfig{
|
||||
VersionCode: config.APP_VERSION,
|
||||
PID: os.Getpid(),
|
||||
NeedUpdate: false,
|
||||
ControlState: "DEGRADED",
|
||||
LastAliveAt: time.Now().Unix(),
|
||||
}
|
||||
|
||||
// 写入文件
|
||||
if err := config.WriteLocalConfig(&cfg); err != nil {
|
||||
log.Println("[config] 写入初始配置失败:", err)
|
||||
}
|
||||
}
|
||||
|
||||
func InitPrint(deviceID string) {
|
||||
banner := `
|
||||
==============================================================================
|
||||
|
||||
██████ █████ ██████ ██████ ██ ██████ ██████ ██ ██ ██████
|
||||
██ ██ ██ ██ ██ ████ ██ ██ ███ ██ ██
|
||||
█████ ██ ██ ██ ██ █████ ██ ██ ██ ██ █████ ████ ██ ██
|
||||
██ ██ ██ ██ ██ ██ ██████ ██ ██ ██ ██ ████ ██
|
||||
██████ █████ █████ ██████ ██ ██ █████ ██████ ██ ██ ██
|
||||
|
||||
==============================================================================
|
||||
`
|
||||
log.Println(banner)
|
||||
log.Println("Device id: "+deviceID, "\t版本号: ", config.APP_VERSION) // 第一次启动记录
|
||||
}
|
||||
|
||||
func InitMQTT(deviceID string) {
|
||||
var mqttSvc *MQTTService
|
||||
firstFail := true // 标记是否第一次失败
|
||||
mqttSvc = NewMQTTService(config.MQTT_BROKER, deviceID, deviceID, config.PASSWORD, 60)
|
||||
for {
|
||||
err := mqttSvc.Connect()
|
||||
if err != nil {
|
||||
if firstFail {
|
||||
log.Error("物联网服务器连接失败，如未注册设备，请先注册: " + deviceID)
|
||||
firstFail = false
|
||||
}
|
||||
time.Sleep(3 * time.Second) // 5秒后重试
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
defer mqttSvc.Close()
|
||||
|
||||
dm, err := docker.NewDockerManager()
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
biz := NewBusinessService(mqttSvc, deviceID, *dm)
|
||||
|
||||
for {
|
||||
// MQTT业务
|
||||
err := biz.Start()
|
||||
if err != nil {
|
||||
log.Error("business service start failed: " + err.Error())
|
||||
fmt.Println("业务启动失败，5秒后重试...")
|
||||
time.Sleep(5 * time.Second)
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
// 第一次运行直接启动
|
||||
biz.handleRestart()
|
||||
|
||||
// 4️⃣ T2 超时自毁检查
|
||||
tickerLost := time.NewTicker(10 * time.Second)
|
||||
defer tickerLost.Stop()
|
||||
go func() {
|
||||
for range tickerLost.C {
|
||||
cfg := config.LoadConfig()
|
||||
now := time.Now().Unix()
|
||||
if cfg.ControlState == "DEGRADED" && now-cfg.LastAliveAt > config.DAEMON_ALIVE_GAP_SECONDS {
|
||||
log.Println("[main] MQTT 长期失联，进入 CONTROL_LOST")
|
||||
cfg.ControlState = "CONTROL_LOST"
|
||||
cfg.LastAliveAt = time.Now().Unix()
|
||||
if err := config.WriteLocalConfig(cfg); err != nil {
|
||||
log.Println("[main] 写状态失败:", err)
|
||||
}
|
||||
// 停止业务容器
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
dm.StopAndRemoveContainer(ctx)
|
||||
// 自杀退出，让 daemon 重启
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
<-make(chan struct{})
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package main
|
||||
|
||||
import "sentinel/pkg/platform"
|
||||
|
||||
func newUpdaterLauncher() platform.UpdaterLauncher {
|
||||
return &updaterLauncher{}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
//go:build linux
|
||||
// +build linux
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"os/exec"
|
||||
)
|
||||
|
||||
type updaterLauncher struct{}
|
||||
|
||||
func (u *updaterLauncher) Start(args []string) error {
|
||||
cmd := exec.Command("./updater", args...)
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
|
||||
// Linux 下先不做进程组处理，保证能跑
|
||||
return cmd.Start()
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
//go:build windows
|
||||
// +build windows
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
type updaterLauncher struct{}
|
||||
|
||||
func (u *updaterLauncher) Start(args []string) error {
|
||||
cmd := exec.Command("./updater.exe", args...)
|
||||
cmd.Stdout = os.Stdout
|
||||
cmd.Stderr = os.Stderr
|
||||
|
||||
cmd.SysProcAttr = &syscall.SysProcAttr{
|
||||
CreationFlags: syscall.CREATE_NEW_PROCESS_GROUP,
|
||||
}
|
||||
|
||||
return cmd.Start()
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"sentinel/pkg/config"
|
||||
"sentinel/pkg/log"
|
||||
"time"
|
||||
|
||||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||||
)
|
||||
|
||||
type MQTTService struct {
|
||||
client mqtt.Client
|
||||
handler func(topic string, payload []byte)
|
||||
subscriptions map[string]struct{} // 记录已订阅 topic
|
||||
}
|
||||
|
||||
func NewMQTTService(broker, clientID, username, password string, keepalive int) *MQTTService {
|
||||
opts := mqtt.NewClientOptions()
|
||||
opts.AddBroker(broker)
|
||||
opts.SetClientID(clientID)
|
||||
opts.SetUsername(username)
|
||||
opts.SetPassword(password)
|
||||
opts.AutoReconnect = true
|
||||
opts.OnReconnecting = func(c mqtt.Client, opts *mqtt.ClientOptions) {
|
||||
log.Println("正在重连到物联网服务器...")
|
||||
}
|
||||
opts.SetKeepAlive(time.Duration(keepalive) * time.Second)
|
||||
opts.SetTLSConfig(&tls.Config{InsecureSkipVerify: true})
|
||||
// 设置遗嘱消息
|
||||
lwtTopic, lwtPayload := getStatusInfoTopicPayload(false)
|
||||
opts.SetWill(
|
||||
lwtTopic, // topic
|
||||
lwtPayload, // payload
|
||||
1, // qos
|
||||
false, // retained
|
||||
)
|
||||
ms := &MQTTService{subscriptions: make(map[string]struct{})}
|
||||
opts.SetDefaultPublishHandler(func(c mqtt.Client, m mqtt.Message) {
|
||||
if ms.handler != nil {
|
||||
ms.handler(m.Topic(), m.Payload())
|
||||
}
|
||||
})
|
||||
opts.OnConnect = func(c mqtt.Client) {
|
||||
log.Println("物联网服务器已连接")
|
||||
cfg := config.LoadConfig()
|
||||
cfg.ControlState = "CONNECTED"
|
||||
cfg.LastAliveAt = time.Now().Unix()
|
||||
config.WriteLocalConfig(cfg)
|
||||
// 连接成功 发送状态信息
|
||||
topic, payload := getStatusInfoTopicPayload(true)
|
||||
ms.PublicMsg(topic, payload, false)
|
||||
}
|
||||
opts.OnConnectionLost = func(c mqtt.Client, err error) {
|
||||
log.Println("物联网服务器已断开：", err)
|
||||
cfg := config.LoadConfig()
|
||||
cfg.ControlState = "DEGRADED" // 表示短暂断开
|
||||
cfg.LastAliveAt = time.Now().Unix() // 更新这个时间戳，让 daemon 与 主程序看到尚可忍耐
|
||||
|
||||
if err := config.WriteLocalConfig(cfg); err != nil {
|
||||
log.Println("[main] 写状态失败:", err)
|
||||
}
|
||||
}
|
||||
|
||||
ms.client = mqtt.NewClient(opts)
|
||||
return ms
|
||||
}
|
||||
|
||||
func (m *MQTTService) Connect() error {
|
||||
token := m.client.Connect()
|
||||
if token.Wait() && token.Error() != nil {
|
||||
return token.Error()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MQTTService) Subscribe(topic string, qos byte) error {
|
||||
token := m.client.Subscribe(topic, qos, nil)
|
||||
if token.Wait() && token.Error() != nil {
|
||||
return token.Error()
|
||||
}
|
||||
log.Debug("物联网服务消息订阅：", topic)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MQTTService) Publish(topic string, qos byte, retained bool, payload interface{}) error {
|
||||
token := m.client.Publish(topic, qos, retained, payload)
|
||||
token.Wait()
|
||||
return token.Error()
|
||||
}
|
||||
|
||||
func (m *MQTTService) SetMessageHandler(h func(topic string, payload []byte)) {
|
||||
m.handler = h
|
||||
}
|
||||
|
||||
func (m *MQTTService) Close() {
|
||||
if m.client != nil {
|
||||
m.client.Disconnect(250)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MQTTService) UnsubscribeTopic(topic string) error {
|
||||
token := m.client.Unsubscribe(topic)
|
||||
go func() {
|
||||
if token.Wait() && token.Error() != nil {
|
||||
log.Println("取消订阅失败:", token.Error())
|
||||
} else {
|
||||
delete(m.subscriptions, topic)
|
||||
log.Debug("取消订阅成功:", topic)
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// UnsubscribeAll 取消所有已订阅 topic
|
||||
func (m *MQTTService) UnsubscribeAll() {
|
||||
for topic := range m.subscriptions {
|
||||
_ = m.client.Unsubscribe(topic)
|
||||
delete(m.subscriptions, topic)
|
||||
}
|
||||
}
|
||||
|
||||
// SubscribeTopic 订阅指定 topic，并记录可取消
|
||||
func (m *MQTTService) SubscribeTopic(topic string, qos byte) error {
|
||||
token := m.client.Subscribe(topic, qos, nil)
|
||||
go func() {
|
||||
if token.Wait() && token.Error() != nil {
|
||||
log.Println("订阅失败:", token.Error())
|
||||
} else {
|
||||
m.subscriptions[topic] = struct{}{}
|
||||
log.Debug("订阅成功:", topic)
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MQTTService) PublicMsg(topic string, payload string, retained bool) {
|
||||
qos := byte(1)
|
||||
log.Debug("发送消息：", topic)
|
||||
if err := m.Publish(topic, qos, retained, payload); err != nil {
|
||||
log.Println("发送信息出错：", err)
|
||||
} else {
|
||||
log.Debug("发送信息：", string(payload))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user