refactor(watchdog): remove tcp socket connection and add local unix sock connection

This commit is contained in:
2026-05-10 14:38:52 +08:00
parent 9c684241d1
commit ef27e7e0a9
10 changed files with 86 additions and 82 deletions
-5
View File
@@ -65,7 +65,6 @@ func LoadConfig(configPath string, getInitSystem func() string) error {
global.CurrentConfig.InstancePath = fileConfig.InstancePath global.CurrentConfig.InstancePath = fileConfig.InstancePath
global.CurrentConfig.Debug = fileConfig.Debug global.CurrentConfig.Debug = fileConfig.Debug
global.CurrentConfig.Watchdog.Enabled = fileConfig.Watchdog.Enabled global.CurrentConfig.Watchdog.Enabled = fileConfig.Watchdog.Enabled
global.CurrentConfig.Watchdog.Port = fileConfig.Watchdog.Port
global.CurrentConfig.Notification.Enabled = fileConfig.Notification.Enabled global.CurrentConfig.Notification.Enabled = fileConfig.Notification.Enabled
global.CurrentConfig.Notification.Method = fileConfig.Notification.Method global.CurrentConfig.Notification.Method = fileConfig.Notification.Method
global.CurrentConfig.Webhook.Method = fileConfig.Webhook.Method global.CurrentConfig.Webhook.Method = fileConfig.Webhook.Method
@@ -93,10 +92,6 @@ func LoadConfig(configPath string, getInitSystem func() string) error {
global.CurrentConfig.InstancePath = "./configs" global.CurrentConfig.InstancePath = "./configs"
} }
if fileConfig.Watchdog.Port == 0 {
global.CurrentConfig.Watchdog.Port = 12380
}
if err := os.MkdirAll(global.CurrentConfig.InstancePath, 0755); err != nil { if err := os.MkdirAll(global.CurrentConfig.InstancePath, 0755); err != nil {
return fmt.Errorf("failed to create config directory: %w", err) return fmt.Errorf("failed to create config directory: %w", err)
} }
-1
View File
@@ -1655,7 +1655,6 @@ key=ListenAddr
"InstancePath": "./configs", "InstancePath": "./configs",
"Debug": true, "Debug": true,
"Watchdog.Enabled": true, "Watchdog.Enabled": true,
"Watchdog.Port": 12380,
"Notification.Enabled": true, "Notification.Enabled": true,
"Notification.Method": "webhook", "Notification.Method": "webhook",
"Webhook.Method": "POST", "Webhook.Method": "POST",
+8 -11
View File
@@ -51,15 +51,14 @@ type Config struct {
Debug bool `json:"debug"` Debug bool `json:"debug"`
Watchdog struct { Watchdog struct {
Enabled bool `json:"enabled"` Enabled bool `json:"enabled"`
Port int `json:"port"`
} `json:"watchdog"` } `json:"watchdog"`
Notification struct { Notification struct {
Enabled bool `json:"enabled"` Enabled bool `json:"enabled"`
Method string `json:"method"` Method string `json:"method"`
} `json:"notification"` } `json:"notification"`
Webhook struct { Webhook struct {
Method string `json:"method"` Method string `json:"method"`
URL string `json:"url"` URL string `json:"url"`
Headers string `json:"headers"` Headers string `json:"headers"`
Body string `json:"body"` Body string `json:"body"`
} `json:"webhook"` } `json:"webhook"`
@@ -73,10 +72,8 @@ var CurrentConfig = Config{
Debug: false, Debug: false,
Watchdog: struct { Watchdog: struct {
Enabled bool `json:"enabled"` Enabled bool `json:"enabled"`
Port int `json:"port"`
}{ }{
Enabled: false, Enabled: false,
Port: 0,
}, },
Notification: struct { Notification: struct {
Enabled bool `json:"enabled"` Enabled bool `json:"enabled"`
@@ -86,14 +83,14 @@ var CurrentConfig = Config{
Method: "webhook", Method: "webhook",
}, },
Webhook: struct { Webhook: struct {
Method string `json:"method"` Method string `json:"method"`
URL string `json:"url"` URL string `json:"url"`
Headers string `json:"headers"` Headers string `json:"headers"`
Body string `json:"body"` Body string `json:"body"`
}{ }{
Method: "", Method: "",
URL: "", URL: "",
Headers: "Content-Type: application/json", Headers: "Content-Type: application/json",
Body: "", Body: "",
}, },
} }
-7
View File
@@ -37,8 +37,6 @@ func GetSettingsHandler(w http.ResponseWriter, r *http.Request) {
value = global.CurrentConfig.Debug value = global.CurrentConfig.Debug
case "Watchdog.Enabled": case "Watchdog.Enabled":
value = global.CurrentConfig.Watchdog.Enabled value = global.CurrentConfig.Watchdog.Enabled
case "Watchdog.Port":
value = global.CurrentConfig.Watchdog.Port
case "Notification.Enabled": case "Notification.Enabled":
value = global.CurrentConfig.Notification.Enabled value = global.CurrentConfig.Notification.Enabled
case "Notification.Method": case "Notification.Method":
@@ -68,7 +66,6 @@ func GetSettingsHandler(w http.ResponseWriter, r *http.Request) {
"InstancePath": global.CurrentConfig.InstancePath, "InstancePath": global.CurrentConfig.InstancePath,
"Debug": global.CurrentConfig.Debug, "Debug": global.CurrentConfig.Debug,
"Watchdog.Enabled": global.CurrentConfig.Watchdog.Enabled, "Watchdog.Enabled": global.CurrentConfig.Watchdog.Enabled,
"Watchdog.Port": global.CurrentConfig.Watchdog.Port,
"Notification.Enabled": global.CurrentConfig.Notification.Enabled, "Notification.Enabled": global.CurrentConfig.Notification.Enabled,
"Notification.Method": global.CurrentConfig.Notification.Method, "Notification.Method": global.CurrentConfig.Notification.Method,
"Webhook.Method": global.CurrentConfig.Webhook.Method, "Webhook.Method": global.CurrentConfig.Webhook.Method,
@@ -131,10 +128,6 @@ func SetSettingsHandler(w http.ResponseWriter, r *http.Request) {
if v, ok := value.(bool); ok { if v, ok := value.(bool); ok {
global.CurrentConfig.Watchdog.Enabled = v global.CurrentConfig.Watchdog.Enabled = v
} }
case "Watchdog.Port":
if v, ok := value.(float64); ok {
global.CurrentConfig.Watchdog.Port = int(v)
}
case "Notification.Enabled": case "Notification.Enabled":
if v, ok := value.(bool); ok { if v, ok := value.(bool); ok {
global.CurrentConfig.Notification.Enabled = v global.CurrentConfig.Notification.Enabled = v
+3 -4
View File
@@ -76,11 +76,10 @@ func main() {
if err != nil { if err != nil {
postLog.Error(fmt.Sprintf("Unable to initialize Watchdog: %s", err)) postLog.Error(fmt.Sprintf("Unable to initialize Watchdog: %s", err))
} else { } else {
if !watchdog.Connect("127.0.0.1", global.CurrentConfig.Watchdog.Port) { if !watchdog.Connect() {
postLog.Error(fmt.Sprintf("Failed to connect to Watchdog at %s:%d", "127.0.0.1", global.CurrentConfig.Watchdog.Port)) postLog.Error("Failed to connect to Watchdog")
} else { } else {
postLog.Info(fmt.Sprintf("Connected to Watchdog at %s:%d", "127.0.0.1", global.CurrentConfig.Watchdog.Port)) postLog.Info("Connected to Watchdog successfully")
global.Is.WatchdogConnected = true
go func() { go func() {
if err := watchdog.StartKeepAlive(); err != nil { if err := watchdog.StartKeepAlive(); err != nil {
@@ -16,26 +16,28 @@ import (
"time" "time"
) )
const socketPath = "/tmp/super-frpc-watchdog.sock"
var ( var (
tcpConn net.Conn localConn net.Conn
tcpConnMutex sync.Mutex localConnMutex sync.Mutex
isConnected bool isConnected bool
recvChan chan string recvChan chan string
stopRecvChan chan struct{} stopRecvChan chan struct{}
recvWg sync.WaitGroup recvWg sync.WaitGroup
) )
func Init() error { func Init() error {
tcpConnMutex.Lock() localConnMutex.Lock()
if recvChan != nil || stopRecvChan != nil { if recvChan != nil || stopRecvChan != nil {
tcpConnMutex.Unlock() localConnMutex.Unlock()
return nil return nil
} }
if tcpConn != nil { if localConn != nil {
tcpConnMutex.Unlock() localConnMutex.Unlock()
return fmt.Errorf("TCP client already initialized") return fmt.Errorf("local socket client already initialized")
} }
tcpConnMutex.Unlock() localConnMutex.Unlock()
if global.CurrentConfig.Watchdog.Enabled { if global.CurrentConfig.Watchdog.Enabled {
if err := ensureWatchdogProcess(); err != nil { if err := ensureWatchdogProcess(); err != nil {
@@ -44,8 +46,8 @@ func Init() error {
} }
} }
tcpConnMutex.Lock() localConnMutex.Lock()
defer tcpConnMutex.Unlock() defer localConnMutex.Unlock()
recvChan = make(chan string, 100) recvChan = make(chan string, 100)
stopRecvChan = make(chan struct{}) stopRecvChan = make(chan struct{})
@@ -135,21 +137,20 @@ func isWatchdogProcessRunning(watchdogName string) bool {
} }
} }
func tcpConnect(ipaddr string, port int) error { func localSocketConnect() error {
tcpConnMutex.Lock() localConnMutex.Lock()
defer tcpConnMutex.Unlock() defer localConnMutex.Unlock()
if tcpConn != nil { if localConn != nil {
return fmt.Errorf("already connected") return fmt.Errorf("already connected")
} }
address := net.JoinHostPort(ipaddr, fmt.Sprintf("%d", port)) conn, err := net.DialTimeout("unix", socketPath, 3*time.Second)
conn, err := net.DialTimeout("tcp", address, 3*time.Second)
if err != nil { if err != nil {
return fmt.Errorf("failed to connect to %s: %v", address, err) return fmt.Errorf("failed to connect to %s: %v", socketPath, err)
} }
tcpConn = conn localConn = conn
isConnected = true isConnected = true
recvWg.Add(1) recvWg.Add(1)
@@ -159,20 +160,20 @@ func tcpConnect(ipaddr string, port int) error {
} }
func sendMsg(message string, timeout int) (string, error) { func sendMsg(message string, timeout int) (string, error) {
tcpConnMutex.Lock() localConnMutex.Lock()
if tcpConn == nil { if localConn == nil {
tcpConnMutex.Unlock() localConnMutex.Unlock()
return "", fmt.Errorf("not connected") return "", fmt.Errorf("not connected")
} }
_, err := tcpConn.Write([]byte(message + "\n")) _, err := localConn.Write([]byte(message + "\n"))
if err != nil { if err != nil {
tcpConnMutex.Unlock() localConnMutex.Unlock()
return "", fmt.Errorf("failed to send message: %v", err) return "", fmt.Errorf("failed to send message: %v", err)
} }
tcpConnMutex.Unlock() localConnMutex.Unlock()
select { select {
case response := <-recvChan: case response := <-recvChan:
@@ -188,13 +189,13 @@ func sendMsg(message string, timeout int) (string, error) {
func recvMsg() { func recvMsg() {
defer recvWg.Done() defer recvWg.Done()
tcpConnMutex.Lock() localConnMutex.Lock()
if tcpConn == nil { if localConn == nil {
tcpConnMutex.Unlock() localConnMutex.Unlock()
return return
} }
conn := tcpConn conn := localConn
tcpConnMutex.Unlock() localConnMutex.Unlock()
reader := bufio.NewReader(conn) reader := bufio.NewReader(conn)
@@ -215,9 +216,7 @@ func recvMsg() {
default: default:
// drop the message // drop the message
} }
// Here add logic to handle the message
if !isResponseMessage(line) { if !isResponseMessage(line) {
// postLog.Debug(fmt.Sprintf("[Watchdog] TCP Socket received message: %s", line))
parseCommand(line) parseCommand(line)
} }
} }
@@ -228,13 +227,13 @@ func recvMsg() {
continue continue
} }
tcpConnMutex.Lock() localConnMutex.Lock()
if tcpConn != nil { if localConn != nil {
tcpConn.Close() localConn.Close()
tcpConn = nil localConn = nil
isConnected = false isConnected = false
} }
tcpConnMutex.Unlock() localConnMutex.Unlock()
return return
} }
} }
@@ -246,23 +245,23 @@ func isResponseMessage(msg string) bool {
} }
func Destroy() error { func Destroy() error {
tcpConnMutex.Lock() localConnMutex.Lock()
if stopRecvChan != nil { if stopRecvChan != nil {
close(stopRecvChan) close(stopRecvChan)
stopRecvChan = nil stopRecvChan = nil
} }
tcpConnMutex.Unlock() localConnMutex.Unlock()
recvWg.Wait() recvWg.Wait()
tcpConnMutex.Lock() localConnMutex.Lock()
defer tcpConnMutex.Unlock() defer localConnMutex.Unlock()
if tcpConn != nil { if localConn != nil {
err := tcpConn.Close() err := localConn.Close()
tcpConn = nil localConn = nil
isConnected = false isConnected = false
if err != nil { if err != nil {
return fmt.Errorf("failed to close connection: %v", err) return fmt.Errorf("failed to close connection: %v", err)
@@ -278,7 +277,7 @@ func Destroy() error {
} }
func IsConnected() bool { func IsConnected() bool {
tcpConnMutex.Lock() localConnMutex.Lock()
defer tcpConnMutex.Unlock() defer localConnMutex.Unlock()
return isConnected && tcpConn != nil return isConnected && localConn != nil
} }
+4 -2
View File
@@ -6,7 +6,7 @@ import (
"time" "time"
) )
func Connect(ipaddr string, port int) bool { func Connect() bool {
if IsConnected() { if IsConnected() {
return true return true
} }
@@ -18,7 +18,7 @@ func Connect(ipaddr string, port int) bool {
deadline := time.Now().Add(5 * time.Second) deadline := time.Now().Add(5 * time.Second)
var lastErr error var lastErr error
for { for {
if err := tcpConnect(ipaddr, port); err == nil { if err := localSocketConnect(); err == nil {
break break
} else { } else {
lastErr = err lastErr = err
@@ -58,6 +58,8 @@ func Disconnect() bool {
return false return false
} }
KillWatchdogProcess()
time.Sleep(100 * time.Millisecond) time.Sleep(100 * time.Millisecond)
return true return true
+2 -2
View File
@@ -29,7 +29,7 @@ func StartKeepAlive() error {
continue continue
} }
if Connect("127.0.0.1", global.CurrentConfig.Watchdog.Port) { if err = Init(); err == nil && Connect() {
global.Is.WatchdogConnected = true global.Is.WatchdogConnected = true
postLog.Info("[watchdog] successfully reconnected to watchdog") postLog.Info("[watchdog] successfully reconnected to watchdog")
lastErr = nil lastErr = nil
@@ -37,7 +37,7 @@ func StartKeepAlive() error {
} }
lastErr = fmt.Errorf("failed to connect to watchdog") lastErr = fmt.Errorf("failed to connect to watchdog")
time.Sleep(500 * time.Millisecond) time.Sleep(1000 * time.Millisecond)
} }
if lastErr != nil { if lastErr != nil {
+10
View File
@@ -6,3 +6,13 @@ import "os/exec"
func configureWatchdogCommand(cmd *exec.Cmd) { func configureWatchdogCommand(cmd *exec.Cmd) {
} }
func KillWatchdogProcess() error {
watchdogName, err := getWatchdogBinaryName()
if err != nil {
return err
}
cmd := exec.Command("pkill", "-f", watchdogName)
return cmd.Run()
}
+10
View File
@@ -10,3 +10,13 @@ import (
func configureWatchdogCommand(cmd *exec.Cmd) { func configureWatchdogCommand(cmd *exec.Cmd) {
cmd.SysProcAttr = &syscall.SysProcAttr{HideWindow: true} cmd.SysProcAttr = &syscall.SysProcAttr{HideWindow: true}
} }
func KillWatchdogProcess() error {
watchdogName, err := getWatchdogBinaryName()
if err != nil {
return err
}
cmd := exec.Command("taskkill", "/IM", watchdogName, "/F")
return cmd.Run()
}