From 79ed8fd6aba51279d2625dfc0c4db5c174860afc Mon Sep 17 00:00:00 2001 From: ryan Date: Fri, 29 May 2026 09:52:34 +0800 Subject: [PATCH] =?UTF-8?q?[=E6=96=B0=E5=A2=9E]=20=E5=AE=9E=E7=8E=B0=20Age?= =?UTF-8?q?nt=20WebSocket=20=E8=BF=9E=E6=8E=A5=E5=8D=87=E7=BA=A7=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=EF=BC=8C=E6=94=AF=E6=8C=81=E7=8A=B6=E6=80=81=E4=B8=8A?= =?UTF-8?q?=E6=8A=A5=E5=92=8C=E9=85=8D=E7=BD=AE=E5=B9=BF=E6=92=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- docs/design/architecture.md | 2 + docs/design/development.md | 1 + docs/design/release-model.md | 2 +- docs/guide/deployment.md | 2 + docs/reference/configuration.md | 2 + openflare_agent/cmd/agent/main.go | 3 + openflare_agent/go.mod | 5 +- openflare_agent/go.sum | 2 + openflare_agent/internal/agent/runner.go | 301 +++++++++++++++--- openflare_agent/internal/agent/runner_test.go | 111 +++++++ .../internal/protocol/agent_api.go | 43 ++- openflare_agent/internal/wsclient/client.go | 165 ++++++++++ openflare_server/common/constants.go | 1 + openflare_server/controller/agent.go | 121 +++++++ openflare_server/controller/option.go | 19 ++ openflare_server/controller/option_test.go | 12 + openflare_server/model/option.go | 3 + openflare_server/router/api-router.go | 1 + openflare_server/service/agent.go | 54 ++-- openflare_server/service/agent_ws.go | 226 +++++++++++++ openflare_server/service/config_version.go | 8 + openflare_server/service/dashboard.go | 36 +-- openflare_server/service/node.go | 12 +- openflare_server/service/node_update_test.go | 37 +++ .../components/dashboard-overview.tsx | 5 +- .../features/nodes/components/nodes-page.tsx | 5 +- openflare_server/web/features/nodes/utils.ts | 20 +- .../settings/components/settings-page.tsx | 22 ++ 29 files changed, 1128 insertions(+), 95 deletions(-) create mode 100644 openflare_agent/go.sum create mode 100644 openflare_agent/internal/wsclient/client.go create mode 100644 openflare_server/service/agent_ws.go diff --git a/README.md b/README.md index b65aa063..89a412c8 100644 --- a/README.md +++ b/README.md @@ -138,7 +138,7 @@ curl -fsSL https://raw.githubusercontent.com/Rain-kl/OpenFlare/main/scripts/unin 1. 登录管理端并新增反代规则 2. 在发布前查看预览或变更摘要 3. 激活新版本 -4. 等待 Agent 在后续 heartbeat 中拉取并应用配置 +4. Agent 通过 WebSocket 通知或后续 heartbeat 拉取并应用配置 版本号格式固定为 `YYYYMMDD-NNN`，历史版本不可变，回滚通过重新激活旧版本完成。 diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 96726f2a..d3e076d6 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -89,6 +89,8 @@ Agent 执行 OpenResty 校验与 reload Agent 上报应用结果 ``` +默认启用 WS 连接升级时，Agent 会先通过 HTTP heartbeat 获取设置，随后尝试连接 Agent WebSocket。WS 成功后，周期性状态上报改由 WS 承载；Server 发布或激活版本后会向已连接 Agent 广播激活版本摘要，使 Agent 立即进入既有同步流程。WS 断开或建立失败时，Agent 自动退回 HTTP heartbeat。 + ### 反向代理流 ```text diff --git a/docs/design/development.md b/docs/design/development.md index 923d8be0..d33dc7dd 100644 --- a/docs/design/development.md +++ b/docs/design/development.md @@ -234,6 +234,7 @@ Agent 必须满足： * 启动后读取或生成本地 `node_id`。 * 周期性心跳与同步。 * 常规同步优先依据 heartbeat 返回的版本摘要判断。 +* WS 连接升级开启且连接成功时，Agent 可通过 WS 接收激活版本摘要并立即同步；WS 失败或断开必须退回 HTTP heartbeat。 * 发现新版本时先备份旧文件。 * 写入主配置、路由配置与必要证书文件。 * 写入新配置后执行 `openresty -t -c `，再 reload；reload 发现运行时未启动时允许直接启动 OpenResty。 diff --git a/docs/design/release-model.md b/docs/design/release-model.md index 834c100b..3d12acd1 100644 --- a/docs/design/release-model.md +++ b/docs/design/release-model.md @@ -35,7 +35,7 @@ Server 发布时必须： 全局同时只能有一个激活版本。当前不做按节点分组的差异化版本。 -Agent 通过 heartbeat 获取激活版本摘要；当远端版本或 checksum 与本地状态不一致时，Agent 才进入同步流程。 +Agent 通过 heartbeat 获取激活版本摘要；当远端版本或 checksum 与本地状态不一致时，Agent 才进入同步流程。当 Agent WS 连接升级开启且连接可用时，Server 在发布或激活版本成功后会广播最新激活版本摘要，Agent 收到后复用普通同步流程立即拉取并应用配置。WS 不可用时仍按 HTTP heartbeat 间隔发现变更。 ## 不可变历史 diff --git a/docs/guide/deployment.md b/docs/guide/deployment.md index 998847a8..cc891af7 100644 --- a/docs/guide/deployment.md +++ b/docs/guide/deployment.md @@ -224,6 +224,8 @@ export LOG_LEVEL='info' 未配置 `openresty_path` 时，Agent 默认调用 `openresty`。 +默认情况下，Agent 在 HTTP 心跳成功后会尝试升级为 WebSocket。升级成功时，Server 发布或激活配置会立即通知 Agent；如果 WebSocket 无法建立或意外断开，Agent 会自动退回 HTTP 心跳同步。 + ## 最小联调步骤 1. 启动 Server 并完成首次登录。 diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index ab1824a4..d8d932d7 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -72,6 +72,7 @@ go run . --port 3000 --log-dir ./logs | 配置项 | 作用 | 默认值 | | --- | --- | --- | | `AgentHeartbeatInterval` | Agent 心跳间隔（毫秒） | `10000` | +| `AgentWebsocketUpgradeEnabled` | 是否允许 Agent 在 HTTP 心跳成功后升级为 WebSocket | `true` | | `NodeOfflineThreshold` | 节点离线阈值（毫秒） | `120000` | | `AgentUpdateRepo` | Agent 自更新仓库 | `Rain-kl/OpenFlare` | | `GeoIPProvider` | 节点/IP 归属解析方式 | `ipinfo` | @@ -186,6 +187,7 @@ OpenResty 性能参数与缓存参数继续统一保存在 `Option` 表。当前 * `agent_token` 与 `discovery_token` 不能同时为空。 * `heartbeat_interval` 与 `request_timeout` 支持毫秒整数或 Go duration 字符串。 +* Server 运行时配置 `AgentWebsocketUpgradeEnabled` 开启时，Agent 会在 HTTP 心跳成功后尝试升级为 WebSocket；连接失败或断开后自动退回 HTTP 心跳。 * 未配置 `openresty_path` 时默认调用 `openresty`。 * 如果 `agent.json` 不存在，但 `OPENFLARE_SERVER_URL` 与 Token 等环境变量足够，Agent 可以直接启动；两者同时存在时环境变量优先。 * Agent 自动探测到私网 `node_ip` 时，Server 会在注册/心跳阶段优先保留 Agent 直连来源的公网地址，避免 NAT/多网卡场景误登记内网网卡地址。 diff --git a/openflare_agent/cmd/agent/main.go b/openflare_agent/cmd/agent/main.go index 778eb5fa..aa568e3f 100644 --- a/openflare_agent/cmd/agent/main.go +++ b/openflare_agent/cmd/agent/main.go @@ -17,6 +17,7 @@ import ( "openflare-agent/internal/state" syncservice "openflare-agent/internal/sync" "openflare-agent/internal/updater" + "openflare-agent/internal/wsclient" ) func main() { @@ -56,6 +57,7 @@ func main() { ) client := httpclient.New(cfg.ServerURL, cfg.InitialAuthToken(), cfg.RequestTimeout.Duration()) + wsClient := wsclient.New(cfg.ServerURL, cfg.InitialAuthToken(), cfg.RequestTimeout.Duration()) stateStore := state.NewStore(cfg.StatePath) observabilityBuffer := state.NewObservabilityBufferStore(cfg.ObservabilityBufferPath) runtimeManager := &nginx.Manager{ @@ -93,6 +95,7 @@ func main() { SyncService: syncservice.New(client, runtimeManager, stateStore), Updater: updater.New(), RuntimeManager: runtimeManager, + WebSocketService: wsClient, } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) diff --git a/openflare_agent/go.mod b/openflare_agent/go.mod index be41283e..d4b50a2c 100644 --- a/openflare_agent/go.mod +++ b/openflare_agent/go.mod @@ -2,6 +2,9 @@ module openflare-agent go 1.25.0 -require openflare v0.0.0 +require ( + golang.org/x/net v0.53.0 + openflare v0.0.0 +) replace openflare => ../openflare_server diff --git a/openflare_agent/go.sum b/openflare_agent/go.sum new file mode 100644 index 00000000..eea2ecc7 --- /dev/null +++ b/openflare_agent/go.sum @@ -0,0 +1,2 @@ +golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= diff --git a/openflare_agent/internal/agent/runner.go b/openflare_agent/internal/agent/runner.go index cba04312..e8fa5cc4 100644 --- a/openflare_agent/internal/agent/runner.go +++ b/openflare_agent/internal/agent/runner.go @@ -2,6 +2,7 @@ package agent import ( "context" + "encoding/json" "errors" "log/slog" "strings" @@ -33,6 +34,12 @@ type RuntimeManager interface { Restart(ctx context.Context) error } +type WebSocketService interface { + Connect(ctx context.Context) (protocol.WebSocketConnection, error) + SetToken(token string) + URL() string +} + type UpdateOptions struct { Channel string TagName string @@ -47,13 +54,15 @@ type Runner struct { SyncService SyncService Updater Updater RuntimeManager RuntimeManager + WebSocketService WebSocketService - autoUpdate bool - updateNow bool - updateRepo string - updateChan string - updateTag string - restartOpenrestyNow bool + autoUpdate bool + updateNow bool + updateRepo string + updateChan string + updateTag string + restartOpenrestyNow bool + websocketUpgradeEnabled bool } func (r *Runner) Run(ctx context.Context) error { @@ -63,26 +72,8 @@ func (r *Runner) Run(ctx context.Context) error { } slog.Info("agent runner started", "node_id", nodeID, "node", r.Config.NodeName, "ip", r.Config.NodeIP) if r.hasAgentToken() { - r.refreshOpenrestyHealth(ctx) - payload, ackWindows := r.prepareHeartbeatPayload(nodeID) - heartbeatResult, hbErr := r.HeartbeatService.Heartbeat(ctx, payload) - if hbErr != nil { + if _, hbErr := r.performHeartbeatCycle(ctx, nodeID, true); hbErr != nil { slog.Error("agent startup heartbeat failed", "error", hbErr) - } else { - r.ackObservabilityWindows(ackWindows) - if heartbeatResult == nil { - heartbeatResult = &protocol.HeartbeatResult{} - } - slog.Debug("agent startup heartbeat succeeded", "node_id", nodeID) - r.applySettings(heartbeatResult.AgentSettings) - if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil { - r.recordSyncError(err) - slog.Error("agent startup sync failed", "error", err) - } else { - slog.Debug("agent startup sync completed") - } - r.tryRestartOpenresty(ctx) - r.tryAutoUpdate(ctx) } } else if err = r.tryRegister(ctx, &nodeID); err != nil { slog.Error("agent initial discovery register failed", "error", err) @@ -90,43 +81,266 @@ func (r *Runner) Run(ctx context.Context) error { heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval.Duration()) defer heartbeatTicker.Stop() + var wsDone <-chan error + wsBackoff := newWebSocketBackoff() + nextWSAttempt := time.Now() + tryStartWebSocket := func() { + if wsDone != nil || !r.shouldUseWebSocket() || time.Now().Before(nextWSAttempt) { + return + } + done, startErr := r.startWebSocket(ctx, nodeID) + if startErr != nil { + delay := wsBackoff.Next() + nextWSAttempt = time.Now().Add(delay) + slog.Debug("agent ws upgrade failed; falling back to http heartbeat", + "enabled", r.websocketUpgradeEnabled, + "url", r.websocketURL(), + "retry_after", delay, + "error", startErr, + ) + return + } + wsBackoff.Reset() + wsDone = done + slog.Debug("agent switched to websocket mode", "url", r.websocketURL()) + } + tryStartWebSocket() for { select { case <-ctx.Done(): slog.Info("agent runner shutting down", "error", ctx.Err()) return ctx.Err() + case wsErr := <-wsDone: + wsDone = nil + delay := wsBackoff.Next() + nextWSAttempt = time.Now().Add(delay) + slog.Debug("agent ws disconnected; resuming http heartbeat", "retry_after", delay, "error", wsErr) + if r.hasAgentToken() { + if _, hbErr := r.performHeartbeatCycle(ctx, nodeID, false); hbErr != nil { + slog.Error("agent heartbeat after ws disconnect failed", "error", hbErr) + } + } case <-heartbeatTicker.C: + if wsDone != nil { + continue + } if !r.hasAgentToken() { if err = r.tryRegister(ctx, &nodeID); err != nil { slog.Error("agent discovery register failed", "error", err) } continue } - r.refreshOpenrestyHealth(ctx) - payload, ackWindows := r.prepareHeartbeatPayload(nodeID) - heartbeatResult, hbErr := r.HeartbeatService.Heartbeat(ctx, payload) - if hbErr != nil { + if changed, hbErr := r.performHeartbeatCycle(ctx, nodeID, false); hbErr != nil { slog.Error("agent heartbeat failed", "error", hbErr) } else { - r.ackObservabilityWindows(ackWindows) - if heartbeatResult == nil { - heartbeatResult = &protocol.HeartbeatResult{} - } - if changed := r.applySettings(heartbeatResult.AgentSettings); changed { + if changed { heartbeatTicker.Reset(r.Config.HeartbeatInterval.Duration()) } - if err = r.SyncService.SyncOnce(ctx, heartbeatResult.ActiveConfig); err != nil { - r.recordSyncError(err) - slog.Error("agent sync failed", "error", err) - } - r.tryRestartOpenresty(ctx) - r.tryAutoUpdate(ctx) + tryStartWebSocket() } } } } +func (r *Runner) performHeartbeatCycle(ctx context.Context, nodeID string, startup bool) (bool, error) { + r.refreshOpenrestyHealth(ctx) + payload, ackWindows := r.prepareHeartbeatPayload(nodeID) + heartbeatResult, err := r.HeartbeatService.Heartbeat(ctx, payload) + if err != nil { + return false, err + } + r.ackObservabilityWindows(ackWindows) + if heartbeatResult == nil { + heartbeatResult = &protocol.HeartbeatResult{} + } + mode := "periodic" + if startup { + mode = "startup" + } + slog.Debug("agent heartbeat succeeded", "mode", mode, "node_id", nodeID) + changed := r.applySettings(heartbeatResult.AgentSettings) + if startup { + if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil { + r.recordSyncError(err) + slog.Error("agent startup sync failed", "error", err) + } else { + slog.Debug("agent startup sync completed") + } + } else if err = r.SyncService.SyncOnce(ctx, heartbeatResult.ActiveConfig); err != nil { + r.recordSyncError(err) + slog.Error("agent sync failed", "error", err) + } + r.tryRestartOpenresty(ctx) + r.tryAutoUpdate(ctx) + return changed, nil +} + +func (r *Runner) shouldUseWebSocket() bool { + enabled := r.WebSocketService != nil && r.websocketUpgradeEnabled && r.hasAgentToken() + slog.Debug("agent ws upgrade eligibility checked", "enabled", enabled, "server_enabled", r.websocketUpgradeEnabled, "url", r.websocketURL()) + return enabled +} + +func (r *Runner) websocketURL() string { + if r.WebSocketService == nil { + return "" + } + return r.WebSocketService.URL() +} + +func (r *Runner) startWebSocket(ctx context.Context, nodeID string) (<-chan error, error) { + if r.WebSocketService == nil { + return nil, errors.New("websocket service is not configured") + } + conn, err := r.WebSocketService.Connect(ctx) + if err != nil { + return nil, err + } + done := make(chan error, 1) + go func() { + defer func() { + _ = conn.Close() + }() + done <- r.runWebSocket(ctx, nodeID, conn) + }() + return done, nil +} + +func (r *Runner) runWebSocket(ctx context.Context, nodeID string, conn protocol.WebSocketConnection) error { + slog.Debug("agent ws connected", "url", conn.URL(), "node_id", nodeID) + statusTicker := time.NewTicker(r.Config.HeartbeatInterval.Duration()) + defer statusTicker.Stop() + + messages := make(chan protocol.WSMessage, 8) + readDone := make(chan error, 1) + go func() { + for { + message, err := conn.Receive() + if err != nil { + readDone <- err + return + } + select { + case messages <- message: + case <-ctx.Done(): + readDone <- ctx.Err() + return + } + } + }() + + if err := r.sendWebSocketStatus(ctx, nodeID, conn); err != nil { + return err + } + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case err := <-readDone: + return err + case <-statusTicker.C: + if err := r.sendWebSocketStatus(ctx, nodeID, conn); err != nil { + return err + } + case message := <-messages: + changed, err := r.handleWebSocketMessage(ctx, message, conn) + if err != nil { + return err + } + if changed { + statusTicker.Reset(r.Config.HeartbeatInterval.Duration()) + } + } + } +} + +func (r *Runner) sendWebSocketStatus(ctx context.Context, nodeID string, conn protocol.WebSocketConnection) error { + r.refreshOpenrestyHealth(ctx) + payload, ackWindows := r.prepareHeartbeatPayload(nodeID) + if err := conn.SendStatus(payload); err != nil { + return err + } + r.ackObservabilityWindows(ackWindows) + return nil +} + +func (r *Runner) handleWebSocketMessage(ctx context.Context, message protocol.WSMessage, conn protocol.WebSocketConnection) (bool, error) { + switch message.Type { + case protocol.WSMessageTypeSettings: + var settings protocol.AgentSettings + if err := json.Unmarshal(message.Payload, &settings); err != nil { + slog.Debug("agent ws settings decode failed", "error", err) + return false, nil + } + changed := r.applySettings(&settings) + r.tryRestartOpenresty(ctx) + r.tryAutoUpdate(ctx) + if !r.websocketUpgradeEnabled { + slog.Debug("agent ws disabled by server settings; falling back to http heartbeat") + return changed, errors.New("websocket upgrade disabled by server") + } + return changed, nil + case protocol.WSMessageTypeActiveConfig: + var target protocol.ActiveConfigMeta + if err := json.Unmarshal(message.Payload, &target); err != nil { + slog.Debug("agent ws active config decode failed", "error", err) + return false, nil + } + slog.Debug("agent ws active config received", "version", target.Version, "checksum", target.Checksum, "trigger_sync", true) + if err := r.SyncService.SyncOnce(ctx, &target); err != nil { + r.recordSyncError(err) + slog.Error("agent ws triggered sync failed", "version", target.Version, "error", err) + } + return false, nil + case protocol.WSMessageTypePing: + slog.Debug("agent ws ping received") + return false, conn.SendPong() + case protocol.WSMessageTypePong: + slog.Debug("agent ws pong received") + return false, nil + default: + slog.Debug("agent ws unsupported message type", "type", message.Type) + return false, nil + } +} + +type webSocketBackoff struct { + delays []time.Duration + index int +} + +func newWebSocketBackoff() *webSocketBackoff { + return &webSocketBackoff{ + delays: []time.Duration{ + time.Second, + 2 * time.Second, + 5 * time.Second, + 10 * time.Second, + 30 * time.Second, + }, + } +} + +func (backoff *webSocketBackoff) Next() time.Duration { + if backoff == nil || len(backoff.delays) == 0 { + return 30 * time.Second + } + if backoff.index >= len(backoff.delays) { + return backoff.delays[len(backoff.delays)-1] + } + delay := backoff.delays[backoff.index] + backoff.index++ + return delay +} + +func (backoff *webSocketBackoff) Reset() { + if backoff != nil { + backoff.index = 0 + } +} + func (r *Runner) hasAgentToken() bool { return strings.TrimSpace(r.Config.AgentToken) != "" } @@ -144,6 +358,10 @@ func (r *Runner) applySettings(settings *protocol.AgentSettings) bool { changed = true } } + if settings.WebsocketUpgradeEnabled != r.websocketUpgradeEnabled { + slog.Debug("agent websocket upgrade setting updated", "from", r.websocketUpgradeEnabled, "to", settings.WebsocketUpgradeEnabled) + } + r.websocketUpgradeEnabled = settings.WebsocketUpgradeEnabled r.autoUpdate = settings.AutoUpdate r.updateNow = settings.UpdateNow r.updateRepo = strings.TrimSpace(settings.UpdateRepo) @@ -222,6 +440,9 @@ func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error { return err } r.HeartbeatService.SetToken(response.AgentToken) + if r.WebSocketService != nil { + r.WebSocketService.SetToken(response.AgentToken) + } *nodeID = response.NodeID slog.Info("agent discovery registration succeeded", "node_id", response.NodeID) r.refreshOpenrestyHealth(ctx) diff --git a/openflare_agent/internal/agent/runner_test.go b/openflare_agent/internal/agent/runner_test.go index 2884be0a..5f169116 100644 --- a/openflare_agent/internal/agent/runner_test.go +++ b/openflare_agent/internal/agent/runner_test.go @@ -2,6 +2,7 @@ package agent import ( "context" + "encoding/json" "errors" "os" "path/filepath" @@ -67,6 +68,7 @@ type fakeSyncService struct { syncOnceErr error startupCalls int syncOnceCalls int + lastTarget *protocol.ActiveConfigMeta onSyncOnceCall func(int) } @@ -104,6 +106,10 @@ func (f *fakeSyncService) SyncOnStartup(ctx context.Context, target *protocol.Ac func (f *fakeSyncService) SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error { f.mu.Lock() f.syncOnceCalls++ + if target != nil { + copied := *target + f.lastTarget = &copied + } callIndex := f.syncOnceCalls callback := f.onSyncOnceCall f.mu.Unlock() @@ -113,6 +119,31 @@ func (f *fakeSyncService) SyncOnce(ctx context.Context, target *protocol.ActiveC return f.syncOnceErr } +type fakeWebSocketConnection struct { + pongCalls int +} + +func (f *fakeWebSocketConnection) URL() string { + return "ws://127.0.0.1/api/agent/ws" +} + +func (f *fakeWebSocketConnection) SendStatus(payload protocol.NodePayload) error { + return nil +} + +func (f *fakeWebSocketConnection) SendPong() error { + f.pongCalls++ + return nil +} + +func (f *fakeWebSocketConnection) Receive() (protocol.WSMessage, error) { + return protocol.WSMessage{}, errors.New("not implemented") +} + +func (f *fakeWebSocketConnection) Close() error { + return nil +} + func TestRunnerKeepsHeartbeatWhenStartupSyncFails(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -495,3 +526,83 @@ func TestRunnerDiscoveryRegisterUpdatesTokenAndNodeID(t *testing.T) { t.Fatal("expected config token rotation to complete") } } + +func TestRunnerHandlesWebSocketActiveConfigMessage(t *testing.T) { + syncService := &fakeSyncService{} + runner := &Runner{SyncService: syncService} + payload, err := json.Marshal(protocol.ActiveConfigMeta{ + Version: "20260529-001", + Checksum: "checksum-ws", + }) + if err != nil { + t.Fatalf("marshal active config: %v", err) + } + + changed, err := runner.handleWebSocketMessage(context.Background(), protocol.WSMessage{ + Type: protocol.WSMessageTypeActiveConfig, + Payload: payload, + }, &fakeWebSocketConnection{}) + if err != nil { + t.Fatalf("handle websocket active config: %v", err) + } + if changed { + t.Fatal("active config message should not change heartbeat interval") + } + if syncService.syncOnceCalls != 1 { + t.Fatalf("expected one sync call, got %d", syncService.syncOnceCalls) + } + if syncService.lastTarget == nil || syncService.lastTarget.Version != "20260529-001" || syncService.lastTarget.Checksum != "checksum-ws" { + t.Fatalf("unexpected sync target: %+v", syncService.lastTarget) + } +} + +func TestRunnerHandlesWebSocketSettingsDisabled(t *testing.T) { + runner := &Runner{ + Config: &config.Config{ + HeartbeatInterval: config.MillisecondDuration(10 * time.Second), + }, + websocketUpgradeEnabled: true, + } + payload, err := json.Marshal(protocol.AgentSettings{ + HeartbeatInterval: 15000, + WebsocketUpgradeEnabled: false, + }) + if err != nil { + t.Fatalf("marshal settings: %v", err) + } + + changed, err := runner.handleWebSocketMessage(context.Background(), protocol.WSMessage{ + Type: protocol.WSMessageTypeSettings, + Payload: payload, + }, &fakeWebSocketConnection{}) + if err == nil { + t.Fatal("expected disabled websocket setting to request fallback") + } + if !changed { + t.Fatal("expected heartbeat interval change to be reported") + } + if runner.websocketUpgradeEnabled { + t.Fatal("expected websocket upgrade to be disabled") + } +} + +func TestWebSocketBackoffSequence(t *testing.T) { + backoff := newWebSocketBackoff() + expected := []time.Duration{ + time.Second, + 2 * time.Second, + 5 * time.Second, + 10 * time.Second, + 30 * time.Second, + 30 * time.Second, + } + for _, want := range expected { + if got := backoff.Next(); got != want { + t.Fatalf("unexpected backoff: got %s want %s", got, want) + } + } + backoff.Reset() + if got := backoff.Next(); got != time.Second { + t.Fatalf("expected reset backoff to return 1s, got %s", got) + } +} diff --git a/openflare_agent/internal/protocol/agent_api.go b/openflare_agent/internal/protocol/agent_api.go index ed4c8957..64898141 100644 --- a/openflare_agent/internal/protocol/agent_api.go +++ b/openflare_agent/internal/protocol/agent_api.go @@ -1,5 +1,7 @@ package protocol +import "encoding/json" + type APIResponse[T any] struct { Success bool `json:"success"` Message string `json:"message"` @@ -20,13 +22,40 @@ type HeartbeatResult struct { } type AgentSettings struct { - HeartbeatInterval int `json:"heartbeat_interval"` - AutoUpdate bool `json:"auto_update"` - UpdateRepo string `json:"update_repo"` - UpdateNow bool `json:"update_now"` - UpdateChannel string `json:"update_channel"` - UpdateTag string `json:"update_tag"` - RestartOpenrestyNow bool `json:"restart_openresty_now"` + HeartbeatInterval int `json:"heartbeat_interval"` + WebsocketUpgradeEnabled bool `json:"websocket_upgrade_enabled"` + AutoUpdate bool `json:"auto_update"` + UpdateRepo string `json:"update_repo"` + UpdateNow bool `json:"update_now"` + UpdateChannel string `json:"update_channel"` + UpdateTag string `json:"update_tag"` + RestartOpenrestyNow bool `json:"restart_openresty_now"` +} + +const ( + WSMessageTypeStatus = "status" + WSMessageTypeSettings = "settings" + WSMessageTypeActiveConfig = "active_config" + WSMessageTypePing = "ping" + WSMessageTypePong = "pong" +) + +type WSMessage struct { + Type string `json:"type"` + Payload json.RawMessage `json:"payload,omitempty"` +} + +type WSOutboundMessage struct { + Type string `json:"type"` + Payload any `json:"payload,omitempty"` +} + +type WebSocketConnection interface { + URL() string + SendStatus(payload NodePayload) error + SendPong() error + Receive() (WSMessage, error) + Close() error } const ( diff --git a/openflare_agent/internal/wsclient/client.go b/openflare_agent/internal/wsclient/client.go new file mode 100644 index 00000000..d587730a --- /dev/null +++ b/openflare_agent/internal/wsclient/client.go @@ -0,0 +1,165 @@ +package wsclient + +import ( + "context" + "errors" + "log/slog" + "net" + "net/http" + "net/url" + "strings" + "time" + + "golang.org/x/net/websocket" + + "openflare-agent/internal/protocol" +) + +type Client struct { + baseURL string + token string + timeout time.Duration +} + +type Connection struct { + conn *websocket.Conn + url string + readTimeout time.Duration +} + +func New(baseURL string, token string, timeout time.Duration) *Client { + return &Client{ + baseURL: strings.TrimRight(baseURL, "/"), + token: strings.TrimSpace(token), + timeout: timeout, + } +} + +func (c *Client) SetToken(token string) { + c.token = strings.TrimSpace(token) + slog.Debug("agent ws client token updated") +} + +func (c *Client) URL() string { + wsURL, err := buildWebsocketURL(c.baseURL) + if err != nil { + return "" + } + return wsURL +} + +func (c *Client) Connect(ctx context.Context) (protocol.WebSocketConnection, error) { + wsURL, err := buildWebsocketURL(c.baseURL) + if err != nil { + return nil, err + } + if strings.TrimSpace(c.token) == "" { + return nil, errors.New("agent ws token is empty") + } + origin := strings.TrimSpace(c.baseURL) + if origin == "" { + origin = "http://localhost" + } + config, err := websocket.NewConfig(wsURL, origin) + if err != nil { + return nil, err + } + config.Header = http.Header{} + config.Header.Set("X-Agent-Token", c.token) + if c.timeout > 0 { + config.Dialer = &net.Dialer{Timeout: c.timeout} + } + slog.Debug("agent ws dialing server", "url", wsURL) + conn, err := config.DialContext(ctx) + if err != nil { + return nil, err + } + slog.Debug("agent ws dial succeeded", "url", wsURL) + return &Connection{conn: conn, url: wsURL, readTimeout: websocketReadTimeout(c.timeout)}, nil +} + +func buildWebsocketURL(baseURL string) (string, error) { + parsed, err := url.Parse(strings.TrimRight(baseURL, "/")) + if err != nil { + return "", err + } + switch parsed.Scheme { + case "http": + parsed.Scheme = "ws" + case "https": + parsed.Scheme = "wss" + case "ws", "wss": + default: + return "", errors.New("server_url scheme must be http, https, ws, or wss") + } + parsed.Path = strings.TrimRight(parsed.Path, "/") + "/api/agent/ws" + parsed.RawQuery = "" + parsed.Fragment = "" + return parsed.String(), nil +} + +func (conn *Connection) URL() string { + if conn == nil { + return "" + } + return conn.url +} + +func (conn *Connection) SendStatus(payload protocol.NodePayload) error { + if conn == nil || conn.conn == nil { + return errors.New("agent ws connection is nil") + } + slog.Debug("agent ws sending status", + "node_id", payload.NodeID, + "current_version", payload.CurrentVersion, + "openresty_status", payload.OpenrestyStatus, + ) + return websocket.JSON.Send(conn.conn, protocol.WSOutboundMessage{ + Type: protocol.WSMessageTypeStatus, + Payload: payload, + }) +} + +func (conn *Connection) SendPong() error { + if conn == nil || conn.conn == nil { + return errors.New("agent ws connection is nil") + } + slog.Debug("agent ws sending pong") + return websocket.JSON.Send(conn.conn, protocol.WSOutboundMessage{ + Type: protocol.WSMessageTypePong, + }) +} + +func (conn *Connection) Receive() (protocol.WSMessage, error) { + var message protocol.WSMessage + if conn == nil || conn.conn == nil { + return message, errors.New("agent ws connection is nil") + } + if conn.readTimeout > 0 { + _ = conn.conn.SetReadDeadline(time.Now().Add(conn.readTimeout)) + } + err := websocket.JSON.Receive(conn.conn, &message) + if err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + slog.Debug("agent ws receive timeout waiting for server message", "timeout", conn.readTimeout) + } + return message, err + } + slog.Debug("agent ws received message", "type", message.Type) + return message, nil +} + +func websocketReadTimeout(requestTimeout time.Duration) time.Duration { + timeout := requestTimeout * 6 + if timeout < 75*time.Second { + return 75 * time.Second + } + return timeout +} + +func (conn *Connection) Close() error { + if conn == nil || conn.conn == nil { + return nil + } + return conn.conn.Close() +} diff --git a/openflare_server/common/constants.go b/openflare_server/common/constants.go index c5d82d6c..b8d77591 100644 --- a/openflare_server/common/constants.go +++ b/openflare_server/common/constants.go @@ -53,6 +53,7 @@ var NodeOfflineThreshold = 2 * time.Minute // V3 operational settings (hot-reloadable via Option table) var AgentHeartbeatInterval = 10000 // milliseconds +var AgentWebsocketUpgradeEnabled = true var AgentUpdateRepo = "Rain-kl/OpenFlare" var GeoIPProvider = "ipinfo" var DatabaseAutoCleanupEnabled = false diff --git a/openflare_server/controller/agent.go b/openflare_server/controller/agent.go index f00203d6..4555c61e 100644 --- a/openflare_server/controller/agent.go +++ b/openflare_server/controller/agent.go @@ -1,11 +1,17 @@ package controller import ( + "encoding/json" + "log/slog" + "net" + "openflare/common" "openflare/model" "openflare/service" "strconv" + "time" "github.com/gin-gonic/gin" + "golang.org/x/net/websocket" ) // AgentRegister godoc @@ -122,6 +128,121 @@ func AgentReportApplyLog(c *gin.Context) { respondSuccess(c, log) } +// AgentWebSocket godoc +// @Summary Upgrade agent connection to websocket +// @Tags Agent +// @Security AgentTokenAuth +// @Router /api/agent/ws [get] +func AgentWebSocket(c *gin.Context) { + authNode, ok := c.Get("agent_node") + if !ok { + respondUnauthorized(c, "无权进行此操作，Agent Token 无效") + return + } + node := authNode.(*model.Node) + slog.Debug("agent ws upgrade requested", "node_id", node.NodeID, "remote", c.Request.RemoteAddr) + websocket.Handler(func(conn *websocket.Conn) { + client := service.RegisterAgentWSClient(node.NodeID) + defer service.UnregisterAgentWSClient(client) + defer func() { + _ = conn.Close() + slog.Debug("agent ws connection closed", "node_id", node.NodeID) + }() + + slog.Debug("agent ws upgrade succeeded", "node_id", node.NodeID, "remote", c.Request.RemoteAddr) + go streamAgentWSMessages(c, conn, client) + + for { + var message service.AgentWSInboundMessage + _ = conn.SetReadDeadline(time.Now().Add(agentWSReadTimeout())) + if err := websocket.JSON.Receive(conn, &message); err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + slog.Debug("agent ws receive timeout waiting for status or pong", "node_id", node.NodeID, "timeout", agentWSReadTimeout()) + return + } + slog.Debug("agent ws receive failed", "node_id", node.NodeID, "error", err) + return + } + slog.Debug("agent ws message received", "node_id", node.NodeID, "type", message.Type) + switch message.Type { + case service.AgentWSMessageTypeStatus: + handleAgentWSStatus(c, node, message) + case service.AgentWSMessageTypePing: + if !service.SendAgentWSPong(node.NodeID) { + slog.Debug("agent ws pong enqueue failed", "node_id", node.NodeID) + } + case service.AgentWSMessageTypePong: + slog.Debug("agent ws pong received", "node_id", node.NodeID) + default: + slog.Debug("agent ws unsupported message type", "node_id", node.NodeID, "type", message.Type) + } + } + }).ServeHTTP(c.Writer, c.Request) +} + +func agentWSReadTimeout() time.Duration { + timeout := time.Duration(common.AgentHeartbeatInterval) * time.Millisecond * 3 + if timeout < 30*time.Second { + return 30 * time.Second + } + return timeout +} + +func streamAgentWSMessages(c *gin.Context, conn *websocket.Conn, client *service.AgentWSClient) { + pingTicker := time.NewTicker(30 * time.Second) + defer pingTicker.Stop() + for { + select { + case message := <-client.Messages(): + slog.Debug("agent ws sending message", "node_id", client.NodeID(), "type", message.Type) + if err := websocket.JSON.Send(conn, message); err != nil { + slog.Debug("agent ws send failed", "node_id", client.NodeID(), "type", message.Type, "error", err) + client.Close() + return + } + case <-pingTicker.C: + message := service.AgentWSOutboundMessage{Type: service.AgentWSMessageTypePing} + slog.Debug("agent ws sending ping", "node_id", client.NodeID()) + if err := websocket.JSON.Send(conn, message); err != nil { + slog.Debug("agent ws ping failed", "node_id", client.NodeID(), "error", err) + client.Close() + return + } + case <-client.Done(): + return + case <-c.Request.Context().Done(): + client.Close() + return + } + } +} + +func handleAgentWSStatus(c *gin.Context, node *model.Node, message service.AgentWSInboundMessage) { + var payload service.AgentNodePayload + if err := json.Unmarshal(message.Payload, &payload); err != nil { + slog.Debug("agent ws status payload decode failed", "node_id", node.NodeID, "error", err) + return + } + payload.IP = service.ResolveReportedNodeIP(payload.IP, c.Request.RemoteAddr) + response, err := service.HeartbeatNode(node, payload) + if err != nil { + slog.Debug("agent ws status handling failed", "node_id", node.NodeID, "error", err) + return + } + settingsSent := service.SendAgentWSSettings(node.NodeID, response.AgentSettings) + activeConfigSent := false + if response.ActiveConfig != nil { + activeConfigSent = service.SendAgentWSActiveConfig(node.NodeID, response.ActiveConfig) + } + slog.Debug("agent ws status processed", + "node_id", node.NodeID, + "current_version", payload.CurrentVersion, + "openresty_status", payload.OpenrestyStatus, + "settings_sent", settingsSent, + "active_config_sent", activeConfigSent, + ) +} + // GetNodes godoc // @Summary List nodes // @Tags Nodes diff --git a/openflare_server/controller/option.go b/openflare_server/controller/option.go index 09f0a396..1e649e0c 100644 --- a/openflare_server/controller/option.go +++ b/openflare_server/controller/option.go @@ -92,6 +92,15 @@ func validateDatabaseCleanupOption(key string, value string) error { } } +func validateAgentOption(key string, value string) error { + switch key { + case "AgentWebsocketUpgradeEnabled": + return validateBooleanOption(key, strings.TrimSpace(value)) + default: + return nil + } +} + func validateOpenRestyOption(key string, value string) error { trimmed := strings.TrimSpace(value) @@ -245,6 +254,9 @@ func validateOptionWithState(option model.Option, state map[string]string) error if err := validateDatabaseCleanupOption(option.Key, option.Value); err != nil { return err } + if err := validateAgentOption(option.Key, option.Value); err != nil { + return err + } return nil } @@ -366,6 +378,13 @@ func UpdateOption(c *gin.Context) { }) return } + if err = validateAgentOption(option.Key, option.Value); err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": err.Error(), + }) + return + } err = model.UpdateOption(option.Key, option.Value) if err != nil { c.JSON(http.StatusOK, gin.H{ diff --git a/openflare_server/controller/option_test.go b/openflare_server/controller/option_test.go index c5a10c8d..17b25c2f 100644 --- a/openflare_server/controller/option_test.go +++ b/openflare_server/controller/option_test.go @@ -46,3 +46,15 @@ func TestValidateOpenRestyOption(t *testing.T) { } } } + +func TestValidateAgentOption(t *testing.T) { + if err := validateAgentOption("AgentWebsocketUpgradeEnabled", "true"); err != nil { + t.Fatalf("expected websocket upgrade option to accept true: %v", err) + } + if err := validateAgentOption("AgentWebsocketUpgradeEnabled", "false"); err != nil { + t.Fatalf("expected websocket upgrade option to accept false: %v", err) + } + if err := validateAgentOption("AgentWebsocketUpgradeEnabled", "on"); err == nil { + t.Fatal("expected websocket upgrade option to reject non-boolean value") + } +} diff --git a/openflare_server/model/option.go b/openflare_server/model/option.go index f005e904..46bd9a42 100644 --- a/openflare_server/model/option.go +++ b/openflare_server/model/option.go @@ -54,6 +54,7 @@ func InitOptionMap() { common.OptionMap["TurnstileSecretKey"] = "" common.OptionMap["AgentDiscoveryToken"] = "" common.OptionMap["AgentHeartbeatInterval"] = strconv.Itoa(common.AgentHeartbeatInterval) + common.OptionMap["AgentWebsocketUpgradeEnabled"] = strconv.FormatBool(common.AgentWebsocketUpgradeEnabled) common.OptionMap["NodeOfflineThreshold"] = strconv.Itoa(int(common.NodeOfflineThreshold.Milliseconds())) common.OptionMap["AgentUpdateRepo"] = common.AgentUpdateRepo common.OptionMap["GeoIPProvider"] = common.GeoIPProvider @@ -221,6 +222,8 @@ func updateOptionMap(key string, value string) { if v, err := strconv.Atoi(value); err == nil && v > 0 { common.AgentHeartbeatInterval = v } + case "AgentWebsocketUpgradeEnabled": + common.AgentWebsocketUpgradeEnabled = value == "true" case "NodeOfflineThreshold": if v, err := strconv.Atoi(value); err == nil && v > 0 { common.NodeOfflineThreshold = time.Duration(v) * time.Millisecond diff --git a/openflare_server/router/api-router.go b/openflare_server/router/api-router.go index 2cbce3ec..a1bd1cbb 100644 --- a/openflare_server/router/api-router.go +++ b/openflare_server/router/api-router.go @@ -197,6 +197,7 @@ func SetApiRouter(router *gin.Engine) { authorizedRoute := agentRoute.Group("/") authorizedRoute.Use(middleware.AgentAuth()) { + authorizedRoute.GET("/ws", controller.AgentWebSocket) authorizedRoute.POST("/nodes/heartbeat", controller.AgentHeartbeat) authorizedRoute.GET("/config-versions/active", controller.AgentGetActiveConfig) authorizedRoute.POST("/apply-logs", controller.AgentReportApplyLog) diff --git a/openflare_server/service/agent.go b/openflare_server/service/agent.go index a942a910..484e921c 100644 --- a/openflare_server/service/agent.go +++ b/openflare_server/service/agent.go @@ -89,13 +89,14 @@ type AgentConfigResponse struct { } type AgentSettings struct { - HeartbeatInterval int `json:"heartbeat_interval"` - AutoUpdate bool `json:"auto_update"` - UpdateRepo string `json:"update_repo"` - UpdateNow bool `json:"update_now"` - UpdateChannel string `json:"update_channel"` - UpdateTag string `json:"update_tag"` - RestartOpenrestyNow bool `json:"restart_openresty_now"` + HeartbeatInterval int `json:"heartbeat_interval"` + WebsocketUpgradeEnabled bool `json:"websocket_upgrade_enabled"` + AutoUpdate bool `json:"auto_update"` + UpdateRepo string `json:"update_repo"` + UpdateNow bool `json:"update_now"` + UpdateChannel string `json:"update_channel"` + UpdateTag string `json:"update_tag"` + RestartOpenrestyNow bool `json:"restart_openresty_now"` } type ActiveConfigMeta struct { @@ -130,7 +131,7 @@ type NodeView struct { OpenrestyMessage string `json:"openresty_message"` Status string `json:"status"` CurrentVersion string `json:"current_version"` - LastSeenAt time.Time `json:"last_seen_at"` + LastSeenAt any `json:"last_seen_at"` LastError string `json:"last_error"` LatestApplyResult string `json:"latest_apply_result"` LatestApplyMessage string `json:"latest_apply_message"` @@ -177,20 +178,32 @@ func HeartbeatNode(node *model.Node, payload AgentNodePayload) (*HeartbeatRespon return nil, err } return &HeartbeatResponse{ - Node: node, - AgentSettings: &AgentSettings{ - HeartbeatInterval: common.AgentHeartbeatInterval, - AutoUpdate: node.AutoUpdateEnabled, - UpdateRepo: common.AgentUpdateRepo, - UpdateNow: updateNow, - UpdateChannel: updateChannel.String(), - UpdateTag: updateTag, - RestartOpenrestyNow: restartOpenrestyNow, - }, - ActiveConfig: activeConfig, + Node: node, + AgentSettings: buildAgentSettings(node, updateNow, updateChannel.String(), updateTag, restartOpenrestyNow), + ActiveConfig: activeConfig, }, nil } +func buildAgentSettings(node *model.Node, updateNow bool, updateChannel string, updateTag string, restartOpenrestyNow bool) *AgentSettings { + autoUpdate := false + if node != nil { + autoUpdate = node.AutoUpdateEnabled + } + if strings.TrimSpace(updateChannel) == "" { + updateChannel = ReleaseChannelStable.String() + } + return &AgentSettings{ + HeartbeatInterval: common.AgentHeartbeatInterval, + WebsocketUpgradeEnabled: common.AgentWebsocketUpgradeEnabled, + AutoUpdate: autoUpdate, + UpdateRepo: common.AgentUpdateRepo, + UpdateNow: updateNow, + UpdateChannel: updateChannel, + UpdateTag: strings.TrimSpace(updateTag), + RestartOpenrestyNow: restartOpenrestyNow, + } +} + func GetActiveConfigMetaForAgent() (*ActiveConfigMeta, error) { version, err := model.GetActiveConfigVersion() if err != nil { @@ -438,6 +451,9 @@ func computeNodeStatus(node *model.Node) string { if node == nil { return NodeStatusOffline } + if IsAgentWSConnected(node.NodeID) { + return NodeStatusOnline + } if node.LastSeenAt.IsZero() { return NodeStatusPending } diff --git a/openflare_server/service/agent_ws.go b/openflare_server/service/agent_ws.go new file mode 100644 index 00000000..b299ab8a --- /dev/null +++ b/openflare_server/service/agent_ws.go @@ -0,0 +1,226 @@ +package service + +import ( + "encoding/json" + "log/slog" + "sync" +) + +const ( + AgentWSMessageTypeStatus = "status" + AgentWSMessageTypeSettings = "settings" + AgentWSMessageTypeActiveConfig = "active_config" + AgentWSMessageTypePing = "ping" + AgentWSMessageTypePong = "pong" + + AgentWSConnectedLastSeenValue = "__OPENFLARE_WS_CONNECTED__" +) + +type AgentWSInboundMessage struct { + Type string `json:"type"` + Payload json.RawMessage `json:"payload,omitempty"` +} + +type AgentWSOutboundMessage struct { + Type string `json:"type"` + Payload any `json:"payload,omitempty"` +} + +type AgentWSBroadcastResult struct { + Version string `json:"version"` + Checksum string `json:"checksum"` + ClientCount int `json:"client_count"` + SuccessCount int `json:"success_count"` + FailedNodes []string `json:"failed_nodes"` +} + +type AgentWSClient struct { + nodeID string + send chan AgentWSOutboundMessage + done chan struct{} + once sync.Once +} + +func (client *AgentWSClient) NodeID() string { + if client == nil { + return "" + } + return client.nodeID +} + +func (client *AgentWSClient) Messages() <-chan AgentWSOutboundMessage { + if client == nil { + return nil + } + return client.send +} + +func (client *AgentWSClient) Done() <-chan struct{} { + if client == nil { + return nil + } + return client.done +} + +func (client *AgentWSClient) Send(message AgentWSOutboundMessage) bool { + if client == nil { + return false + } + select { + case <-client.done: + return false + case client.send <- message: + return true + default: + return false + } +} + +func (client *AgentWSClient) Close() { + if client == nil { + return + } + client.once.Do(func() { + close(client.done) + }) +} + +type agentWSHub struct { + mu sync.RWMutex + clients map[string]*AgentWSClient +} + +var defaultAgentWSHub = &agentWSHub{ + clients: make(map[string]*AgentWSClient), +} + +func RegisterAgentWSClient(nodeID string) *AgentWSClient { + client := &AgentWSClient{ + nodeID: nodeID, + send: make(chan AgentWSOutboundMessage, 16), + done: make(chan struct{}), + } + defaultAgentWSHub.mu.Lock() + if existing := defaultAgentWSHub.clients[nodeID]; existing != nil { + slog.Debug("agent ws replacing existing connection", "node_id", nodeID) + existing.Close() + } + defaultAgentWSHub.clients[nodeID] = client + count := len(defaultAgentWSHub.clients) + defaultAgentWSHub.mu.Unlock() + slog.Debug("agent ws connection registered", "node_id", nodeID, "client_count", count) + return client +} + +func UnregisterAgentWSClient(client *AgentWSClient) { + if client == nil { + return + } + defaultAgentWSHub.mu.Lock() + if current := defaultAgentWSHub.clients[client.nodeID]; current == client { + delete(defaultAgentWSHub.clients, client.nodeID) + } + count := len(defaultAgentWSHub.clients) + defaultAgentWSHub.mu.Unlock() + client.Close() + slog.Debug("agent ws connection unregistered", "node_id", client.nodeID, "client_count", count) +} + +func IsAgentWSConnected(nodeID string) bool { + defaultAgentWSHub.mu.RLock() + client := defaultAgentWSHub.clients[nodeID] + defaultAgentWSHub.mu.RUnlock() + if client == nil { + return false + } + select { + case <-client.done: + return false + default: + return true + } +} + +func AgentWSClientCount() int { + defaultAgentWSHub.mu.RLock() + defer defaultAgentWSHub.mu.RUnlock() + return len(defaultAgentWSHub.clients) +} + +func SendAgentWSSettings(nodeID string, settings *AgentSettings) bool { + if settings == nil { + return false + } + return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{ + Type: AgentWSMessageTypeSettings, + Payload: settings, + }) +} + +func SendAgentWSActiveConfig(nodeID string, activeConfig *ActiveConfigMeta) bool { + if activeConfig == nil { + return false + } + return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{ + Type: AgentWSMessageTypeActiveConfig, + Payload: activeConfig, + }) +} + +func SendAgentWSPong(nodeID string) bool { + return sendAgentWSMessage(nodeID, AgentWSOutboundMessage{ + Type: AgentWSMessageTypePong, + }) +} + +func sendAgentWSMessage(nodeID string, message AgentWSOutboundMessage) bool { + defaultAgentWSHub.mu.RLock() + client := defaultAgentWSHub.clients[nodeID] + defaultAgentWSHub.mu.RUnlock() + if client == nil { + return false + } + ok := client.Send(message) + if !ok { + slog.Debug("agent ws send queued message failed", "node_id", nodeID, "type", message.Type) + } + return ok +} + +func BroadcastAgentWSActiveConfig(activeConfig *ActiveConfigMeta) AgentWSBroadcastResult { + result := AgentWSBroadcastResult{} + if activeConfig == nil { + slog.Debug("agent ws broadcast skipped because active config is nil") + return result + } + result.Version = activeConfig.Version + result.Checksum = activeConfig.Checksum + + defaultAgentWSHub.mu.RLock() + clients := make([]*AgentWSClient, 0, len(defaultAgentWSHub.clients)) + for _, client := range defaultAgentWSHub.clients { + clients = append(clients, client) + } + defaultAgentWSHub.mu.RUnlock() + + result.ClientCount = len(clients) + message := AgentWSOutboundMessage{ + Type: AgentWSMessageTypeActiveConfig, + Payload: activeConfig, + } + for _, client := range clients { + if client.Send(message) { + result.SuccessCount++ + continue + } + result.FailedNodes = append(result.FailedNodes, client.NodeID()) + } + slog.Debug("agent ws broadcast active config", + "version", result.Version, + "checksum", result.Checksum, + "client_count", result.ClientCount, + "success_count", result.SuccessCount, + "failed_nodes", result.FailedNodes, + ) + return result +} diff --git a/openflare_server/service/config_version.go b/openflare_server/service/config_version.go index 18697215..1a5c5644 100644 --- a/openflare_server/service/config_version.go +++ b/openflare_server/service/config_version.go @@ -385,6 +385,10 @@ func PublishConfigVersion(createdBy string, force bool) (*ReleaseResult, error) } return nil, err } + BroadcastAgentWSActiveConfig(&ActiveConfigMeta{ + Version: record.Version, + Checksum: record.Checksum, + }) return &ReleaseResult{ Version: record, Routes: bundle.Routes, @@ -409,6 +413,10 @@ func ActivateConfigVersion(id uint) (*model.ConfigVersion, error) { return nil, err } version.IsActive = true + BroadcastAgentWSActiveConfig(&ActiveConfigMeta{ + Version: version.Version, + Checksum: version.Checksum, + }) return version, nil } diff --git a/openflare_server/service/dashboard.go b/openflare_server/service/dashboard.go index f1acd476..e4155033 100644 --- a/openflare_server/service/dashboard.go +++ b/openflare_server/service/dashboard.go @@ -48,23 +48,23 @@ type DashboardTrends struct { } type DashboardNodeHealth struct { - ID uint `json:"id"` - NodeID string `json:"node_id"` - Name string `json:"name"` - GeoName string `json:"geo_name"` - GeoLatitude *float64 `json:"geo_latitude"` - GeoLongitude *float64 `json:"geo_longitude"` - Status string `json:"status"` - OpenrestyStatus string `json:"openresty_status"` - CurrentVersion string `json:"current_version"` - LastSeenAt time.Time `json:"last_seen_at"` - ActiveEventCount int `json:"active_event_count"` - CPUUsagePercent float64 `json:"cpu_usage_percent"` - MemoryUsagePercent float64 `json:"memory_usage_percent"` - StorageUsagePercent float64 `json:"storage_usage_percent"` - RequestCount int64 `json:"request_count"` - ErrorCount int64 `json:"error_count"` - UniqueVisitorCount int64 `json:"unique_visitor_count"` + ID uint `json:"id"` + NodeID string `json:"node_id"` + Name string `json:"name"` + GeoName string `json:"geo_name"` + GeoLatitude *float64 `json:"geo_latitude"` + GeoLongitude *float64 `json:"geo_longitude"` + Status string `json:"status"` + OpenrestyStatus string `json:"openresty_status"` + CurrentVersion string `json:"current_version"` + LastSeenAt any `json:"last_seen_at"` + ActiveEventCount int `json:"active_event_count"` + CPUUsagePercent float64 `json:"cpu_usage_percent"` + MemoryUsagePercent float64 `json:"memory_usage_percent"` + StorageUsagePercent float64 `json:"storage_usage_percent"` + RequestCount int64 `json:"request_count"` + ErrorCount int64 `json:"error_count"` + UniqueVisitorCount int64 `json:"unique_visitor_count"` } func GetDashboardOverview() (*DashboardOverviewView, error) { @@ -139,7 +139,7 @@ func GetDashboardOverview() (*DashboardOverviewView, error) { Status: computedStatus, OpenrestyStatus: node.OpenrestyStatus, CurrentVersion: node.CurrentVersion, - LastSeenAt: node.LastSeenAt, + LastSeenAt: nodeViewLastSeenAt(node), ActiveEventCount: len(nodeActiveEvents), } diff --git a/openflare_server/service/node.go b/openflare_server/service/node.go index 9290869f..c3074890 100644 --- a/openflare_server/service/node.go +++ b/openflare_server/service/node.go @@ -275,7 +275,7 @@ func buildNodeView(node *model.Node) *NodeView { OpenrestyMessage: strings.TrimSpace(node.OpenrestyMessage), Status: status, CurrentVersion: node.CurrentVersion, - LastSeenAt: node.LastSeenAt, + LastSeenAt: nodeViewLastSeenAt(node), LastError: node.LastError, CreatedAt: node.CreatedAt, UpdatedAt: node.UpdatedAt, @@ -288,6 +288,16 @@ func buildNodeView(node *model.Node) *NodeView { return view } +func nodeViewLastSeenAt(node *model.Node) any { + if node != nil && IsAgentWSConnected(node.NodeID) { + return AgentWSConnectedLastSeenValue + } + if node == nil { + return time.Time{} + } + return node.LastSeenAt +} + func normalizeNodeInput(input NodeInput) (string, string, string, *float64, *float64, bool, error) { name := strings.TrimSpace(input.Name) ip := strings.TrimSpace(input.IP) diff --git a/openflare_server/service/node_update_test.go b/openflare_server/service/node_update_test.go index 2f4396ce..cd416797 100644 --- a/openflare_server/service/node_update_test.go +++ b/openflare_server/service/node_update_test.go @@ -615,6 +615,43 @@ func TestListNodeViewsDoesNotPersistComputedStatus(t *testing.T) { } } +func TestAgentWSConnectionMarksNodeViewOnline(t *testing.T) { + node := &model.Node{ + ID: 99, + NodeID: "node-ws-view", + Name: "edge-ws", + IP: "10.0.0.9", + Status: NodeStatusOffline, + LastSeenAt: time.Now().Add(-common.NodeOfflineThreshold - time.Minute), + } + client := RegisterAgentWSClient(node.NodeID) + defer UnregisterAgentWSClient(client) + + view := buildNodeView(node) + if view.Status != NodeStatusOnline { + t.Fatalf("expected websocket-connected node to be online, got %s", view.Status) + } + if view.LastSeenAt != AgentWSConnectedLastSeenValue { + t.Fatalf("expected websocket special last_seen_at, got %#v", view.LastSeenAt) + } + + result := BroadcastAgentWSActiveConfig(&ActiveConfigMeta{ + Version: "20260529-001", + Checksum: "checksum-ws", + }) + if result.ClientCount != 1 || result.SuccessCount != 1 || len(result.FailedNodes) != 0 { + t.Fatalf("unexpected broadcast result: %+v", result) + } + select { + case message := <-client.Messages(): + if message.Type != AgentWSMessageTypeActiveConfig { + t.Fatalf("unexpected websocket message type: %s", message.Type) + } + case <-time.After(time.Second): + t.Fatal("expected websocket broadcast message") + } +} + func TestHeartbeatNodePersistsObservabilityPayload(t *testing.T) { setupServiceTestDB(t) withFakeAccessLogGeoProvider(t, &geoip.GeoInfo{ diff --git a/openflare_server/web/features/dashboard/components/dashboard-overview.tsx b/openflare_server/web/features/dashboard/components/dashboard-overview.tsx index 0e612283..1f4cd5d4 100644 --- a/openflare_server/web/features/dashboard/components/dashboard-overview.tsx +++ b/openflare_server/web/features/dashboard/components/dashboard-overview.tsx @@ -18,6 +18,7 @@ import { getNodeStatusVariant, getOpenrestyStatusLabel, getOpenrestyStatusVariant, + isWSConnectedLastSeen, } from '@/features/nodes/utils'; import { formatDateTime, formatRelativeTime } from '@/lib/utils/date'; import { @@ -133,7 +134,9 @@ function NodeHealthRow({ node }: { node: DashboardNodeHealth }) {

当前版本：{node.current_version || '未应用'}

最近心跳： - {node.last_seen_at + {isWSConnectedLastSeen(node.last_seen_at) + ? ' WS 已连接' + : node.last_seen_at ? ` ${formatRelativeTime(node.last_seen_at)} · ${formatDateTime(node.last_seen_at)}` : ' 暂无'}

diff --git a/openflare_server/web/features/nodes/components/nodes-page.tsx b/openflare_server/web/features/nodes/components/nodes-page.tsx index 88506c0a..76fce423 100644 --- a/openflare_server/web/features/nodes/components/nodes-page.tsx +++ b/openflare_server/web/features/nodes/components/nodes-page.tsx @@ -34,6 +34,7 @@ import { getOpenrestyStatusLabel, getOpenrestyStatusVariant, isMeaningfulTime, + isWSConnectedLastSeen, } from '@/features/nodes/utils'; const nodesQueryKey = ['nodes']; @@ -350,7 +351,9 @@ export function NodesPage() { - {isMeaningfulTime(node.last_seen_at) + {isWSConnectedLastSeen(node.last_seen_at) + ? 'WS 已连接' + : isMeaningfulTime(node.last_seen_at) ? `${formatRelativeTime( node.last_seen_at, )} · ${formatDateTime(node.last_seen_at)}` diff --git a/openflare_server/web/features/nodes/utils.ts b/openflare_server/web/features/nodes/utils.ts index fce1c5d2..0187776e 100644 --- a/openflare_server/web/features/nodes/utils.ts +++ b/openflare_server/web/features/nodes/utils.ts @@ -1,8 +1,18 @@ -import type { NodeItem } from '@/features/nodes/types'; - -export function isMeaningfulTime(value: string | null | undefined) { - return Boolean(value) && !String(value).startsWith('0001-01-01'); -} +import type { NodeItem } from '@/features/nodes/types'; + +export const WS_CONNECTED_LAST_SEEN = '__OPENFLARE_WS_CONNECTED__'; + +export function isWSConnectedLastSeen(value: string | null | undefined) { + return value === WS_CONNECTED_LAST_SEEN; +} + +export function isMeaningfulTime(value: string | null | undefined) { + return ( + Boolean(value) && + !isWSConnectedLastSeen(value) && + !String(value).startsWith('0001-01-01') + ); +} export function getNodeStatusVariant(status: NodeItem['status']) { if (status === 'online') { diff --git a/openflare_server/web/features/settings/components/settings-page.tsx b/openflare_server/web/features/settings/components/settings-page.tsx index c7e3b8b2..fc36ad79 100644 --- a/openflare_server/web/features/settings/components/settings-page.tsx +++ b/openflare_server/web/features/settings/components/settings-page.tsx @@ -87,6 +87,7 @@ const defaultSystemFields = { const defaultOperationFields = { AgentHeartbeatInterval: '10000', + AgentWebsocketUpgradeEnabled: true, NodeOfflineThreshold: '120000', AgentUpdateRepo: 'Rain-kl/OpenFlare', GeoIPProvider: 'ipinfo', @@ -379,6 +380,10 @@ export function SettingsPage() { setOperationFields({ AgentHeartbeatInterval: optionMap.AgentHeartbeatInterval ?? '10000', + AgentWebsocketUpgradeEnabled: toBoolean( + optionMap.AgentWebsocketUpgradeEnabled, + true, + ), NodeOfflineThreshold: optionMap.NodeOfflineThreshold ?? '120000', AgentUpdateRepo: optionMap.AgentUpdateRepo ?? 'Rain-kl/OpenFlare', GeoIPProvider: optionMap.GeoIPProvider ?? 'ipinfo', @@ -1059,6 +1064,12 @@ export function SettingsPage() { await saveOptionEntries( [ ['AgentHeartbeatInterval', String(heartbeat)], + [ + 'AgentWebsocketUpgradeEnabled', + String( + operationFields.AgentWebsocketUpgradeEnabled, + ), + ], ['NodeOfflineThreshold', String(offline)], [ 'AgentUpdateRepo', @@ -1115,6 +1126,17 @@ export function SettingsPage() { /> + + setOperationFields((previous) => ({ + ...previous, + AgentWebsocketUpgradeEnabled: checked, + })) + } + />