From 197cef8b7ff9a3c834dc418b3f813f9843ee98e3 Mon Sep 17 00:00:00 2001 From: lzhj <526494747@qq.com> Date: Sun, 7 Jun 2026 17:40:56 +0800 Subject: [PATCH] feat: tunnel state reporting to control plane Co-Authored-By: Claude Opus 4.7 --- .codegraph/.gitignore | 16 ++ .../config_control_plane_compat_test.go | 147 ++++++++++++++++++ server/client_host_list.go | 1 + server/engine_startup.go | 4 +- server/server.go | 5 + web/api/tunnels.go | 3 +- web/routers/node_ws.go | 90 +++++++++++ web/routers/state.go | 12 ++ web/service/resources.go | 50 ++++++ 9 files changed, 326 insertions(+), 2 deletions(-) create mode 100644 .codegraph/.gitignore create mode 100644 lib/servercfg/config_control_plane_compat_test.go diff --git a/.codegraph/.gitignore b/.codegraph/.gitignore new file mode 100644 index 00000000..9de0f169 --- /dev/null +++ b/.codegraph/.gitignore @@ -0,0 +1,16 @@ +# CodeGraph data files +# These are local to each machine and should not be committed + +# Database +*.db +*.db-wal +*.db-shm + +# Cache +cache/ + +# Logs +*.log + +# Hook markers +.dirty diff --git a/lib/servercfg/config_control_plane_compat_test.go b/lib/servercfg/config_control_plane_compat_test.go new file mode 100644 index 00000000..24fb638b --- /dev/null +++ b/lib/servercfg/config_control_plane_compat_test.go @@ -0,0 +1,147 @@ +package servercfg + +import ( + "strings" + "testing" +) + +// TestControlPlaneFlatConfigParsing verifies that the flat INI format +// produced by nps_enhanced/renderPushModeConfig is correctly parsed. +func TestControlPlaneFlatConfigParsing(t *testing.T) { + resetTestState(t) + + // This is the exact format produced by nps_enhanced renderPushModeConfig + // (flat keys in the default section, no [common] block). + path := writeConfig(t, "nps.conf", strings.Join([]string{ + "platform_ids=nps-enhanced-42", + "platform_tokens=test-platform-token-abc", + "platform_scopes=full", + "platform_connect_modes=reverse", + "platform_reverse_ws_urls=ws://ctrl.example.com:18081/node/ws", + "platform_reverse_enabled=true", + "platform_reverse_heartbeat_seconds=30", + "platform_callback_urls=https://ctrl.example.com:18081/node/callback", + "platform_callback_enabled=true", + "platform_callback_signing_keys=test-signing-key-xyz", + "# Legacy native keys (ignored by data-plane)", + "run_mode=node", + "platform_url=https://ctrl.example.com:18081", + "platform_token=test-platform-token-abc", + "callback_signing_key=test-signing-key-xyz", + "traffic_report=true", + }, "\n")+"\n") + + if err := Load(path); err != nil { + t.Fatalf("Load() error = %v", err) + } + + cfg := Current() + if cfg == nil { + t.Fatal("Current() returned nil") + } + + platforms := cfg.Runtime.ManagementPlatforms + if len(platforms) != 1 { + t.Fatalf("expected 1 management platform, got %d", len(platforms)) + } + + p := platforms[0] + if p.PlatformID != "nps-enhanced-42" { + t.Errorf("PlatformID = %q, want nps-enhanced-42", p.PlatformID) + } + if p.Token != "test-platform-token-abc" { + t.Errorf("Token = %q, want test-platform-token-abc", p.Token) + } + if p.ControlScope != "full" { + t.Errorf("ControlScope = %q, want full", p.ControlScope) + } + if p.ConnectMode != "reverse" { + t.Errorf("ConnectMode = %q, want reverse", p.ConnectMode) + } + if p.ReverseWSURL != "ws://ctrl.example.com:18081/node/ws" { + t.Errorf("ReverseWSURL = %q, want ws://ctrl.example.com:18081/node/ws", p.ReverseWSURL) + } + if !p.ReverseEnabled { + t.Error("ReverseEnabled = false, want true") + } + if p.ReverseHeartbeatSeconds != 30 { + t.Errorf("ReverseHeartbeatSeconds = %d, want 30", p.ReverseHeartbeatSeconds) + } + if p.CallbackURL != "https://ctrl.example.com:18081/node/callback" { + t.Errorf("CallbackURL = %q, want https://ctrl.example.com:18081/node/callback", p.CallbackURL) + } + if !p.CallbackEnabled { + t.Error("CallbackEnabled = false, want true") + } + if p.CallbackSigningKey != "test-signing-key-xyz" { + t.Errorf("CallbackSigningKey = %q, want test-signing-key-xyz", p.CallbackSigningKey) + } +} + +// TestControlPlaneMergedConfigParsing verifies that a merged config +// (push block inserted at top, before [common]) is correctly parsed. +func TestControlPlaneMergedConfigParsing(t *testing.T) { + resetTestState(t) + + path := writeConfig(t, "nps.conf", strings.Join([]string{ + "# --- nps-enhanced push mode config START ---", + "platform_ids=nps-enhanced-1", + "platform_tokens=tok-1", + "platform_scopes=full", + "platform_connect_modes=reverse", + "platform_reverse_ws_urls=ws://ctrl.example.com:18081/node/ws", + "platform_reverse_enabled=true", + "platform_reverse_heartbeat_seconds=30", + "platform_callback_urls=https://ctrl.example.com:18081/node/callback", + "platform_callback_enabled=true", + "platform_callback_signing_keys=key-1", + "# --- nps-enhanced push mode config END ---", + "", + "[common]", + "web_port=8080", + "run_mode=server", + }, "\n")+"\n") + + if err := Load(path); err != nil { + t.Fatalf("Load() error = %v", err) + } + + cfg := Current() + platforms := cfg.Runtime.ManagementPlatforms + if len(platforms) != 1 { + t.Fatalf("expected 1 platform, got %d", len(platforms)) + } + if platforms[0].PlatformID != "nps-enhanced-1" { + t.Errorf("PlatformID = %q, want nps-enhanced-1", platforms[0].PlatformID) + } + if platforms[0].ConnectMode != "reverse" { + t.Errorf("ConnectMode = %q, want reverse", platforms[0].ConnectMode) + } +} + +// TestCommonSectionPlatformKeysIgnored verifies that platform keys inside +// [common] are NOT visible to the data-plane parser (they become +// common_platform_ids after flattening). This confirms why mergePushConfig +// must place push keys in the default section (before any [section]). +func TestCommonSectionPlatformKeysIgnored(t *testing.T) { + resetTestState(t) + + path := writeConfig(t, "nps.conf", strings.Join([]string{ + "[common]", + "platform_ids=old-master", + "platform_tokens=old-tok", + "platform_connect_modes=reverse", + "platform_reverse_ws_urls=ws://old.example/ws", + "platform_reverse_enabled=true", + }, "\n")+"\n") + + if err := Load(path); err != nil { + t.Fatalf("Load() error = %v", err) + } + + cfg := Current() + platforms := cfg.Runtime.ManagementPlatforms + if len(platforms) != 0 { + t.Fatalf("expected 0 platforms (keys inside [common] are invisible), got %d", len(platforms)) + } +} diff --git a/server/client_host_list.go b/server/client_host_list.go index f43ddec5..6f400759 100644 --- a/server/client_host_list.go +++ b/server/client_host_list.go @@ -562,6 +562,7 @@ func snapshotTunnelForList(tunnel *file.Tunnel) *file.Tunnel { ServiceTraffic: cloneTrafficStatsForList(tunnel.ServiceTraffic), ServiceMeter: tunnel.ServiceMeter.Clone(), NowConn: tunnel.NowConn, + MaxConn: tunnel.MaxConn, Password: tunnel.Password, Remark: tunnel.Remark, TargetAddr: tunnel.TargetAddr, diff --git a/server/engine_startup.go b/server/engine_startup.go index f7ad1a7e..1a7377e5 100644 --- a/server/engine_startup.go +++ b/server/engine_startup.go @@ -184,7 +184,9 @@ func wrapBackgroundLoopStart(start func()) func() { } var once sync.Once return func() { - once.Do(start) + once.Do(func() { + go start() + }) } } diff --git a/server/server.go b/server/server.go index b6fd25a2..bfe07af2 100644 --- a/server/server.go +++ b/server/server.go @@ -26,6 +26,11 @@ var ( Bridge *bridge.Bridge RunList sync.Map //map[int]interface{} HttpProxyCache = index.NewAnyIntIndex() + + // ManagementEventHook is called when tunnel/client/host resources are mutated. + // It is set by web/routers/state.go during node initialization. + // The hook receives: eventName, resource, action, fields. + ManagementEventHook func(eventName, resource, action string, fields map[string]interface{}) ) func ClearProxyCache() { diff --git a/web/api/tunnels.go b/web/api/tunnels.go index 320d371e..b6ee1a6d 100644 --- a/web/api/tunnels.go +++ b/web/api/tunnels.go @@ -187,7 +187,8 @@ func (a *App) sanitizeNodeTunnel(actor *Actor, scope webservice.NodeAccessScope, } func (a *App) finishNodeTunnelMutation(c Context, action, eventName string, id int, tunnel *file.Tunnel, overrides map[string]interface{}) { - a.emitNodeResourceMutationEvent(c, eventName, "tunnel", action, nodeResourceMutationFields(id, tunnelEventFields(tunnel), overrides)) + // Event emission is now handled exclusively by the service layer (web/service/resources.go). + // a.emitNodeResourceMutationEvent(c, eventName, "tunnel", action, nodeResourceMutationFields(id, tunnelEventFields(tunnel), overrides)) a.respondCompletedNodeMutation(c, "tunnel", action, id, a.nodeTunnelResource(tunnel)) } diff --git a/web/routers/node_ws.go b/web/routers/node_ws.go index cff59cce..c9a85c9a 100644 --- a/web/routers/node_ws.go +++ b/web/routers/node_ws.go @@ -18,6 +18,7 @@ import ( "github.com/djylb/nps/lib/logs" "github.com/djylb/nps/lib/servercfg" + "github.com/djylb/nps/server" webapi "github.com/djylb/nps/web/api" webservice "github.com/djylb/nps/web/service" "github.com/gin-gonic/gin" @@ -2018,6 +2019,25 @@ func (m *nodeReverseManager) serveConn(platform servercfg.ManagementPlatformConf ) }(platform.ReverseHeartbeatSeconds) + // Periodic full tunnel state reporting (60s) + backgroundWG.Add(1) + go func() { + defer backgroundWG.Done() + ticker := time.NewTicker(60 * time.Second) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + report := buildTunnelFullReportFrame(m.state, platform.PlatformID) + if report != nil { + _ = writeFrame(*report) + } + } + } + }() + reverseHost := reverseWSHost(platform.ReverseWSURL) base := nodeWSDispatchBase{ Context: m.ctx, @@ -2246,3 +2266,73 @@ func (m *nodeReverseManager) runtimeStatus() webservice.ManagementPlatformRuntim } return webservice.NewInMemoryManagementPlatformRuntimeStatusStore() } + + +// buildTunnelFullReportFrame collects all tunnels and builds a WS event frame. +func buildTunnelFullReportFrame(state *State, platformID string) *nodeWSFrame { + if state == nil { + return nil + } + allTunnels, _ := server.GetTunnel(0, 99999, "", 0, "", "", "") + if len(allTunnels) == 0 { + return nil + } + tunnelSnapshots := make([]map[string]interface{}, 0, len(allTunnels)) + for _, t := range allTunnels { + targetAddr := "" + if t.Target != nil { + targetAddr = t.Target.TargetStr + } + snap := map[string]interface{}{ + "id": t.Id, + "port": t.Port, + "mode": t.Mode, + "status": t.Status, + "run_status": t.RunStatus, + "remark": t.Remark, + "target": targetAddr, + "target_type": t.TargetType, + "now_conn": t.NowConn, + "max_conn": t.MaxConn, + "expire_at": t.ExpireAt, + "rate_limit": t.RateLimit, + } + if t.Client != nil { + snap["client_id"] = t.Client.Id + } + if t.Flow != nil { + snap["flow_in"] = t.Flow.InletFlow + snap["flow_out"] = t.Flow.ExportFlow + snap["flow_limit"] = t.Flow.FlowLimit + } + if t.ServiceTraffic != nil { + snap["service_traffic_in"] = t.ServiceTraffic.InletBytes + snap["service_traffic_out"] = t.ServiceTraffic.ExportBytes + } + tunnelSnapshots = append(tunnelSnapshots, snap) + } + + event := webapi.Event{ + Name: "tunnel.full_report", + Resource: "tunnel", + Action: "report", + Fields: map[string]interface{}{ + "platform_id": platformID, + "tunnels": tunnelSnapshots, + "reported_at": time.Now().Unix(), + "count": len(tunnelSnapshots), + }, + } + body, err := json.Marshal(event) + if err != nil { + logs.Warn("[rev-ws] failed to marshal tunnel full report: %v", err) + return nil + } + frame := &nodeWSFrame{ + Type: "event", + ID: "tunnel-report-" + strconv.FormatInt(time.Now().UnixNano(), 10), + Timestamp: time.Now().Unix(), + Body: body, + } + return frame +} diff --git a/web/routers/state.go b/web/routers/state.go index 0a8ccfe3..eb9ec04e 100644 --- a/web/routers/state.go +++ b/web/routers/state.go @@ -16,6 +16,7 @@ import ( "github.com/djylb/nps/lib/common" "github.com/djylb/nps/lib/logs" "github.com/djylb/nps/lib/servercfg" + "github.com/djylb/nps/server" webapi "github.com/djylb/nps/web/api" webservice "github.com/djylb/nps/web/service" ) @@ -606,6 +607,17 @@ func NewStateWithApp(app *webapi.App) *State { hooks.webhooks = state.NodeWebhooks app.Hooks = hooks } + // Register the global management event hook so web/service can emit events. + // Must be done after state is fully initialized to avoid nil pointer dereferences + // if an event is emitted concurrently during startup. + server.ManagementEventHook = func(eventName, resource, action string, fields map[string]interface{}) { + emitNodeManagementEvent(state, state.BaseContext(), webapi.Event{ + Name: eventName, + Resource: resource, + Action: action, + Fields: fields, + }) + } return state } diff --git a/web/service/resources.go b/web/service/resources.go index 5a5d8707..9b4ee88e 100644 --- a/web/service/resources.go +++ b/web/service/resources.go @@ -9,6 +9,7 @@ import ( "github.com/djylb/nps/lib/common" "github.com/djylb/nps/lib/crypt" "github.com/djylb/nps/lib/file" + "github.com/djylb/nps/server" ) type IndexService interface { @@ -615,6 +616,7 @@ func (s DefaultIndexService) AddTunnel(input AddTunnelInput) (TunnelMutation, er return TunnelMutation{}, normalizedErr } + emitTunnelEvent("tunnel.created", "create", tunnel) return newRuntimeTunnelMutationResult(s.runtime(), tunnel), nil } @@ -733,6 +735,7 @@ func (s DefaultIndexService) EditTunnel(input EditTunnelInput) (TunnelMutation, return TunnelMutation{}, mapTunnelNotFound(normalizeRuntimeError(err)) } + emitTunnelEvent("tunnel.updated", "update", working) return newRuntimeTunnelMutationResult(s.runtime(), working), nil } @@ -754,6 +757,7 @@ func (s DefaultIndexService) StopTunnel(id int, mode string) (TunnelMutation, er } return TunnelMutation{}, err } + emitTunnelEvent("tunnel.stopped", "stop", tunnel) return newRuntimeTunnelMutationResult(s.runtime(), tunnel), nil } @@ -778,6 +782,7 @@ func (s DefaultIndexService) DeleteTunnel(id int) (TunnelMutation, error) { return TunnelMutation{ID: id}, nil } deleted.NowConn = 0 + emitTunnelEvent("tunnel.deleted", "delete", deleted) return newRuntimeTunnelMutationResult(s.runtime(), deleted), nil } @@ -799,6 +804,7 @@ func (s DefaultIndexService) StartTunnel(id int, mode string) (TunnelMutation, e } return TunnelMutation{}, err } + emitTunnelEvent("tunnel.started", "start", tunnel) return newRuntimeTunnelMutationResult(s.runtime(), tunnel), nil } @@ -851,6 +857,7 @@ func changeTunnelStatusWithRepo(repo IndexRepository, runtime IndexRuntime, id i if err := repo.SaveTunnel(working); err != nil { return TunnelMutation{}, mapTunnelNotFound(err) } + emitTunnelEvent("tunnel.updated", "update", working) return newRuntimeTunnelMutationResult(runtime, working), nil } @@ -1546,3 +1553,46 @@ func (s DefaultIndexService) quotaStore() QuotaStore { } return DefaultQuotaStore{} } + +func emitTunnelEvent(eventName, action string, tunnel *file.Tunnel) { + hook := server.ManagementEventHook + if hook == nil || tunnel == nil { + return + } + targetAddr := "" + if tunnel.Target != nil { + tunnel.Target.RLock() + targetAddr = tunnel.Target.TargetStr + tunnel.Target.RUnlock() + } + fields := map[string]interface{}{ + "id": tunnel.Id, + "port": tunnel.Port, + "mode": tunnel.Mode, + "status": tunnel.Status, + "run_status": tunnel.RunStatus, + "remark": tunnel.Remark, + "target": targetAddr, + "target_type": tunnel.TargetType, + "now_conn": tunnel.NowConn, + "max_conn": tunnel.MaxConn, + "expire_at": tunnel.ExpireAt, + "rate_limit": tunnel.RateLimit, + } + if tunnel.Client != nil { + fields["client_id"] = tunnel.Client.Id + } + if tunnel.Flow != nil { + tunnel.Flow.RLock() + fields["flow_in"] = tunnel.Flow.InletFlow + fields["flow_out"] = tunnel.Flow.ExportFlow + fields["flow_limit"] = tunnel.Flow.FlowLimit + tunnel.Flow.RUnlock() + } + if tunnel.ServiceTraffic != nil { + in, out, _ := tunnel.ServiceTraffic.Snapshot() + fields["service_traffic_in"] = in + fields["service_traffic_out"] = out + } + hook(eventName, "tunnel", action, fields) +}