From a5d7be8405cd07ac711c69e5600b48b0f4cd273e Mon Sep 17 00:00:00 2001 From: qaq <1937228092@qq.com> Date: Fri, 27 Jun 2025 11:56:59 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=B5=81=E9=87=8F=E7=BB=9F?= =?UTF-8?q?=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go-gost/go.mod | 3 +- go-gost/go.sum | 9 +- go-gost/main.go | 17 +- go-gost/program.go | 12 + go-gost/x/config/parsing/service/parse.go | 37 ++- go-gost/x/handler/forward/local/handler.go | 91 ++----- go-gost/x/observer/stats/stats.go | 15 +- go-gost/x/registry/observer.go | 3 +- go-gost/x/service/service.go | 66 +++-- go-gost/x/service/traffic_reporter.go | 226 ++++++++++++++++++ go-gost/x/socket/config.go | 58 ----- go-gost/x/socket/websocket_reporter.go | 132 +--------- go-gost/x/traffic/memory_manager.go | 118 --------- go-gost/x/traffic/traffic.go | 34 --- go-gost/x/traffic/traffic_reporter.go | 190 --------------- .../com/admin/common/dto/UserUpdateDto.java | 3 - .../common/task/CheckGostConfigAsync.java | 4 +- .../admin/common/utils/WebSocketServer.java | 11 +- .../com/admin/config/WebSocketConfig.java | 11 - .../com/admin/controller/FlowController.java | 21 +- .../admin/service/impl/NodeServiceImpl.java | 26 +- .../src/main/resources/application.yml | 1 - vue-frontend/src/views/Index.vue | 13 +- vue-frontend/src/views/User.vue | 15 +- 24 files changed, 431 insertions(+), 685 deletions(-) create mode 100644 go-gost/x/service/traffic_reporter.go delete mode 100644 go-gost/x/traffic/memory_manager.go delete mode 100644 go-gost/x/traffic/traffic.go delete mode 100644 go-gost/x/traffic/traffic_reporter.go diff --git a/go-gost/go.mod b/go-gost/go.mod index 9a6da8f..5e36bc1 100644 --- a/go-gost/go.mod +++ b/go-gost/go.mod @@ -103,7 +103,7 @@ require ( github.com/tklauser/numcpus v0.6.1 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect - github.com/vishvananda/netlink v1.1.0 // indirect + github.com/vishvananda/netlink v1.1.1-0.20211118161826-650dca95af54 // indirect github.com/vishvananda/netns v0.0.4 // indirect github.com/vulcand/predicate v1.2.0 // indirect github.com/xtaci/kcp-go/v5 v5.6.5 // indirect @@ -133,6 +133,7 @@ require ( gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect + gvisor.dev/gvisor v0.0.0-20230927004350-cbd86285d259 // indirect ) replace github.com/go-gost/x => ./x diff --git a/go-gost/go.sum b/go-gost/go.sum index 1c52ff4..d8b07b5 100644 --- a/go-gost/go.sum +++ b/go-gost/go.sum @@ -266,9 +266,9 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE= github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= -github.com/vishvananda/netlink v1.1.0 h1:1iyaYNBLmP6L0220aDnYQpo1QEV4t4hJ+xEEhhJH8j0= -github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE= -github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU= +github.com/vishvananda/netlink v1.1.1-0.20211118161826-650dca95af54 h1:8mhqcHPqTMhSPoslhGYihEgSfc77+7La1P6kiB6+9So= +github.com/vishvananda/netlink v1.1.1-0.20211118161826-650dca95af54/go.mod h1:twkDnbuQxJYemMlGd4JFIcuhgX83tXhKS2B/PRMpOho= +github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0= github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8= github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM= github.com/vulcand/predicate v1.2.0 h1:uFsW1gcnnR7R+QTID+FVcs0sSYlIGntoGOTb3rQJt50= @@ -340,9 +340,10 @@ golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200217220822-9197077df867/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200728102440-3e129f6d46b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= diff --git a/go-gost/main.go b/go-gost/main.go index 687d5b2..b5909f6 100644 --- a/go-gost/main.go +++ b/go-gost/main.go @@ -14,8 +14,8 @@ import ( "github.com/go-gost/core/logger" xlogger "github.com/go-gost/x/logger" + "github.com/go-gost/x/service" "github.com/go-gost/x/socket" - "github.com/go-gost/x/traffic" "github.com/judwhite/go-svc" ) @@ -109,28 +109,21 @@ func main() { // 加载配置文件 config, err := LoadConfig("config.json") if err != nil { - fmt.Printf("❌ 配置加载失败: %v\n", err) + fmt.Println("❌ 配置加载失败: %v\n", err) fmt.Println("请确保当前目录存在 config.json 文件") os.Exit(1) } - fmt.Printf("✅ 配置加载成功 - addr: %s", config.Addr) + fmt.Println("✅ 配置加载成功 - addr: %s", config.Addr) log := xlogger.NewLogger() logger.SetDefault(log) - trafficMgr := traffic.GetGlobalManager() - defer trafficMgr.Close() - - fmt.Println("✅ 使用内存流量管理器") - logger.Default().Info("Using memory traffic manager") - - traffic.SetHTTPReportURL(config.Addr, config.Secret) - traffic.StartTrafficReporter(trafficMgr) - wsReporter := socket.StartWebSocketReporterWithConfig(config.Addr, config.Secret) defer wsReporter.Stop() + service.SetHTTPReportURL(config.Addr, config.Secret) + p := &program{} if err := svc.Run(p); err != nil { diff --git a/go-gost/program.go b/go-gost/program.go index 13be148..6bb3b4f 100644 --- a/go-gost/program.go +++ b/go-gost/program.go @@ -8,6 +8,7 @@ import ( "os/signal" "strings" "syscall" + "time" "github.com/go-gost/core/auth" "github.com/go-gost/core/logger" @@ -21,6 +22,7 @@ import ( xmetrics "github.com/go-gost/x/metrics" metrics "github.com/go-gost/x/metrics/service" "github.com/go-gost/x/registry" + xservice "github.com/go-gost/x/service" "github.com/judwhite/go-svc" ) @@ -73,6 +75,16 @@ func (p *program) Start() error { p.cancel = cancel go p.reload(ctx) + // 延迟启动配置定时上报器（等待30秒让WebSocket连接稳定） + go func() { + select { + case <-time.After(10 * time.Second): + xservice.StartConfigReporter(ctx) + case <-ctx.Done(): + return + } + }() + return nil } diff --git a/go-gost/x/config/parsing/service/parse.go b/go-gost/x/config/parsing/service/parse.go index 4ff061f..afe5325 100644 --- a/go-gost/x/config/parsing/service/parse.go +++ b/go-gost/x/config/parsing/service/parse.go @@ -12,6 +12,7 @@ import ( "github.com/go-gost/core/hop" "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" + "github.com/go-gost/core/observer" "github.com/go-gost/core/observer/stats" "github.com/go-gost/core/recorder" "github.com/go-gost/core/selector" @@ -112,6 +113,9 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { var limiterCleanupInterval time.Duration var limiterScope string + enableStats := true + observerPeriod = 5 * time.Second + if cfg.Metadata != nil { md := metadata.NewMetadata(cfg.Metadata) ppv = mdutil.GetInt(md, parsing.MDKeyProxyProtocol) @@ -129,10 +133,13 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { postDown = mdutil.GetStrings(md, parsing.MDKeyPostDown) ignoreChain = mdutil.GetBool(md, parsing.MDKeyIgnoreChain) - if mdutil.GetBool(md, parsing.MDKeyEnableStats) { - pStats = xstats.NewStats(mdutil.GetBool(md, parsing.MDKeyObserverResetTraffic)) + if md.IsExists(parsing.MDKeyEnableStats) { + enableStats = mdutil.GetBool(md, parsing.MDKeyEnableStats) + } + + if period := mdutil.GetDuration(md, parsing.MDKeyObserverPeriod, "observePeriod"); period > 0 { + observerPeriod = period } - observerPeriod = mdutil.GetDuration(md, parsing.MDKeyObserverPeriod, "observePeriod") netnsIn = mdutil.GetString(md, parsing.MDKeyNetns) netnsOut = mdutil.GetString(md, parsing.MDKeyNetnsOut) @@ -144,6 +151,17 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { limiterScope = mdutil.GetString(md, parsing.MDKeyLimiterScope) } + if enableStats { + resetTraffic := true + if cfg.Metadata != nil { + md := metadata.NewMetadata(cfg.Metadata) + if md.IsExists(parsing.MDKeyObserverResetTraffic) { + resetTraffic = mdutil.GetBool(md, parsing.MDKeyObserverResetTraffic) + } + } + pStats = xstats.NewStats(resetTraffic) + } + listenerLogger := serviceLogger.WithFields(map[string]any{ "kind": "listener", }) @@ -329,6 +347,17 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { return nil, err } + var observer observer.Observer + // 如果服务名以_tls结尾，则不启用观察器 + if strings.HasSuffix(cfg.Name, "_tls") { + observer = nil + fmt.Println("服务名以_tls结尾，跳过观察器启用") + } else if cfg.Observer != "" { + observer = registry.ObserverRegistry().Get(cfg.Observer) + } else if pStats != nil { + observer = registry.ObserverRegistry().Get("console") + } + s := xservice.NewService(cfg.Name, ln, h, xservice.AdmissionOption(xadmission.AdmissionGroup(admissions...)), xservice.PreUpOption(preUp), @@ -337,7 +366,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { xservice.PostDownOption(postDown), xservice.RecordersOption(recorders...), xservice.StatsOption(pStats), - xservice.ObserverOption(registry.ObserverRegistry().Get(cfg.Observer)), + xservice.ObserverOption(observer), xservice.ObserverPeriodOption(observerPeriod), xservice.LoggerOption(serviceLogger), ) diff --git a/go-gost/x/handler/forward/local/handler.go b/go-gost/x/handler/forward/local/handler.go index 5d18880..cff7e81 100644 --- a/go-gost/x/handler/forward/local/handler.go +++ b/go-gost/x/handler/forward/local/handler.go @@ -24,7 +24,6 @@ import ( stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" - "github.com/go-gost/x/traffic" ) func init() { @@ -34,12 +33,11 @@ func init() { } type forwardHandler struct { - hop hop.Hop - md metadata - options handler.Options - recorder recorder.RecorderObject - certPool tls_util.CertPool - trafficManager traffic.Manager + hop hop.Hop + md metadata + options handler.Options + recorder recorder.RecorderObject + certPool tls_util.CertPool } func NewHandler(opts ...handler.Option) handler.Handler { @@ -49,8 +47,7 @@ func NewHandler(opts ...handler.Option) handler.Handler { } return &forwardHandler{ - options: options, - trafficManager: traffic.GetGlobalManager(), + options: options, } } @@ -103,75 +100,23 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand ro.ClientIP = h } - log := h.options.Logger.WithFields(map[string]any{ - "remote": conn.RemoteAddr().String(), - "local": conn.LocalAddr().String(), - "sid": ro.SID, - "client": ro.ClientIP, - }) - network := "tcp" if _, ok := conn.(net.PacketConn); ok { network = "udp" } ro.Network = network - connStats := xstats.NewStats(false) // false表示不在Get时清零 - ccStats := xstats.NewStats(false) // false表示不在Get时清零 - conn = stats_wrapper.WrapConn(conn, connStats) - - // 启动定期上报流量的goroutine - trafficCtx, trafficCancel := context.WithCancel(context.Background()) - defer trafficCancel() - - if h.trafficManager != nil { - go func() { - ticker := time.NewTicker(5 * time.Second) - defer ticker.Stop() - - // 记录上次的流量值 - var lastConnOutput, lastConnInput, lastCCOutput, lastCCInput uint64 - - for { - select { - case <-ticker.C: - // 获取当前流量统计（不清零） - connOutput := connStats.Get(stats.KindInputBytes) - connInput := connStats.Get(stats.KindOutputBytes) - ccOutput := ccStats.Get(stats.KindOutputBytes) - ccInput := ccStats.Get(stats.KindInputBytes) - - // 计算增量 - connOutputDelta := connOutput - lastConnOutput - connInputDelta := connInput - lastConnInput - ccOutputDelta := ccOutput - lastCCOutput - ccInputDelta := ccInput - lastCCInput - - if connInputDelta > 0 || connOutputDelta > 0 { - h.trafficManager.RecordTraffic(ctx, ro.Service+":conn", int64(connInputDelta), int64(connOutputDelta)) - } - if ccOutputDelta > 0 || ccInputDelta > 0 { - h.trafficManager.RecordTraffic(ctx, ro.Service+":cc", int64(ccOutputDelta), int64(ccInputDelta)) - } - - // 更新上次的值 - lastConnOutput = connOutput - lastConnInput = connInput - lastCCOutput = ccOutput - lastCCInput = ccInput - case <-trafficCtx.Done(): - return - } - } - }() - } + pStats := xstats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) defer func() { if err != nil { ro.Err = err.Error() } + ro.InputBytes = pStats.Get(stats.KindInputBytes) + ro.OutputBytes = pStats.Get(stats.KindOutputBytes) ro.Duration = time.Since(start) - // 流量统计已经在定期上报中处理，这里不需要再次记录 + }() if !h.checkRateLimit(conn.RemoteAddr()) { @@ -196,9 +141,6 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand var buf bytes.Buffer cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", address) ro.Route = buf.String() - - cc = stats_wrapper.WrapConn(cc, ccStats) - return cc, err } sniffer := &forwarder.Sniffer{ @@ -223,7 +165,6 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand forwarder.WithBypass(h.options.Bypass), forwarder.WithHTTPKeepalive(h.md.httpKeepalive), forwarder.WithRecorderObject(ro), - forwarder.WithLog(log), ) case sniffing.ProtoTLS: return sniffer.HandleTLS(ctx, conn, @@ -231,14 +172,15 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand forwarder.WithHop(h.hop), forwarder.WithBypass(h.options.Bypass), forwarder.WithRecorderObject(ro), - forwarder.WithLog(log), ) } } target := &chain.Node{} if h.hop != nil { - target = h.hop.Select(ctx, hop.ProtocolSelectOption(proto)) + target = h.hop.Select(ctx, + hop.ProtocolSelectOption(proto), + ) } if target == nil { err := errors.New("node not available") @@ -264,6 +206,8 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), network, addr) ro.Route = buf.String() if err != nil { + // TODO: the router itself may be failed due to the failed node in the router, + // the dead marker may be a wrong operation. if marker := target.Marker(); marker != nil { marker.Mark() } @@ -272,9 +216,6 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand if marker := target.Marker(); marker != nil { marker.Reset() } - - cc = stats_wrapper.WrapConn(cc, ccStats) - defer cc.Close() xnet.Transport(conn, cc) diff --git a/go-gost/x/observer/stats/stats.go b/go-gost/x/observer/stats/stats.go index a3a4e98..6fabbae 100644 --- a/go-gost/x/observer/stats/stats.go +++ b/go-gost/x/observer/stats/stats.go @@ -57,21 +57,20 @@ func (s *Stats) Get(kind stats.Kind) uint64 { case stats.KindCurrentConns: return s.currentConns.Load() case stats.KindInputBytes: - if s.resetTraffic { - return s.inputBytes.Swap(0) - } - return s.inputBytes.Load() + return s.inputBytes.Load() // 只获取，不自动清零 case stats.KindOutputBytes: - if s.resetTraffic { - return s.outputBytes.Swap(0) - } - return s.outputBytes.Load() + return s.outputBytes.Load() // 只获取，不自动清零 case stats.KindTotalErrs: return s.totalErrs.Load() } return 0 } +func (s *Stats) ResetTraffic(reportedInputBytes, reportedOutputBytes uint64) { + s.inputBytes.Store(reportedInputBytes) + s.outputBytes.Store(reportedOutputBytes) +} + func (s *Stats) Reset() { s.updated.Store(false) s.totalConns.Store(0) diff --git a/go-gost/x/registry/observer.go b/go-gost/x/registry/observer.go index 62e9df2..8008d48 100644 --- a/go-gost/x/registry/observer.go +++ b/go-gost/x/registry/observer.go @@ -35,5 +35,6 @@ func (w *observerWrapper) Observe(ctx context.Context, events []observer.Event, if v == nil { return nil } - return v.Observe(ctx, events, opts...) + err := v.Observe(ctx, events, opts...) + return err } diff --git a/go-gost/x/service/service.go b/go-gost/x/service/service.go index 10807df..94a22ea 100644 --- a/go-gost/x/service/service.go +++ b/go-gost/x/service/service.go @@ -338,31 +338,65 @@ func (s *defaultService) observeStats(ctx context.Context) { for { select { case <-ticker.C: + + // First, try to send any pending events if len(events) > 0 { if err := s.options.observer.Observe(ctx, events); err == nil { events = nil } - break + continue } st := s.status.Stats() - if st == nil || !st.IsUpdated() { - break + if st == nil { + continue } - evs := []observer.Event{ - xstats.StatsEvent{ - Kind: "service", - Service: s.name, - TotalConns: st.Get(stats.KindTotalConns), - CurrentConns: st.Get(stats.KindCurrentConns), - InputBytes: st.Get(stats.KindInputBytes), - OutputBytes: st.Get(stats.KindOutputBytes), - TotalErrs: st.Get(stats.KindTotalErrs), - }, - } - if err := s.options.observer.Observe(ctx, evs); err != nil { - events = evs + isUpdated := st.IsUpdated() + if isUpdated { + inputBytes := st.Get(stats.KindInputBytes) + outputBytes := st.Get(stats.KindOutputBytes) + + evs := []observer.Event{ + xstats.StatsEvent{ + Kind: "service", + Service: s.name, + TotalConns: st.Get(stats.KindTotalConns), + CurrentConns: st.Get(stats.KindCurrentConns), + InputBytes: inputBytes, + OutputBytes: outputBytes, + TotalErrs: st.Get(stats.KindTotalErrs), + }, + } + if outputBytes > 0 || inputBytes > 0 { + reportItems := []TrafficReportItem{ + { + N: s.name, + T: "cc", + U: int64(outputBytes), + D: int64(inputBytes), + }, + { + N: s.name, + T: "conn", + U: int64(inputBytes), + D: int64(outputBytes), + }, + } + success, err := sendTrafficReport(ctx, reportItems) + if err != nil { + fmt.Printf("发送流量报告失败: %v", err) + } else if success { + if xstats, ok := st.(*xstats.Stats); ok { + xstats.ResetTraffic(st.Get(stats.KindInputBytes)-inputBytes, st.Get(stats.KindOutputBytes)-outputBytes) + } + } + } + + if err := s.options.observer.Observe(ctx, evs); err != nil { + fmt.Printf("发送观察器事件失败: %v", err) + events = evs + } } case <-ctx.Done(): diff --git a/go-gost/x/service/traffic_reporter.go b/go-gost/x/service/traffic_reporter.go new file mode 100644 index 0000000..cbe1642 --- /dev/null +++ b/go-gost/x/service/traffic_reporter.go @@ -0,0 +1,226 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "github.com/go-gost/core/observer/stats" + "github.com/go-gost/x/config" + "github.com/go-gost/x/registry" +) + +var httpReportURL string +var configReportURL string + +// TrafficReportItem 流量报告项（压缩格式） +type TrafficReportItem struct { + N string `json:"n"` // 服务名（name缩写） + T string `json:"t"` // 连接类型：conn, cc（type缩写） + U int64 `json:"u"` // 上行流量（up缩写） + D int64 `json:"d"` // 下行流量（down缩写） +} + +func SetHTTPReportURL(addr string, secret string) { + httpReportURL = "http://" + addr + "/flow/upload?secret=" + secret + configReportURL = "http://" + addr + "/flow/config?secret=" + secret +} + +// sendTrafficReport 发送流量报告到HTTP接口 +func sendTrafficReport(ctx context.Context, reportItems []TrafficReportItem) (bool, error) { + jsonData, err := json.Marshal(reportItems) + if err != nil { + return false, fmt.Errorf("序列化报告数据失败: %v", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", httpReportURL, bytes.NewBuffer(jsonData)) + if err != nil { + return false, fmt.Errorf("创建HTTP请求失败: %v", err) + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "GOST-Traffic-Reporter/1.0") + + client := &http.Client{ + Timeout: 5 * time.Second, + } + + resp, err := client.Do(req) + if err != nil { + return false, fmt.Errorf("发送HTTP请求失败: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return false, fmt.Errorf("HTTP响应错误: %d %s", resp.StatusCode, resp.Status) + } + + // 读取响应内容 + var responseBytes bytes.Buffer + _, err = responseBytes.ReadFrom(resp.Body) + if err != nil { + return false, fmt.Errorf("读取响应内容失败: %v", err) + } + + responseText := strings.TrimSpace(responseBytes.String()) + + // 检查响应是否为"ok" + if responseText == "ok" { + return true, nil + } else { + return false, fmt.Errorf("服务器响应: %s (期望: ok)", responseText) + } +} + +// sendConfigReport 发送配置报告到HTTP接口 +func sendConfigReport(ctx context.Context) (bool, error) { + if configReportURL == "" { + return false, fmt.Errorf("配置上报URL未设置") + } + + // 获取配置数据 + configData, err := getConfigData() + if err != nil { + return false, fmt.Errorf("获取配置数据失败: %v", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", configReportURL, bytes.NewBuffer(configData)) + if err != nil { + return false, fmt.Errorf("创建HTTP请求失败: %v", err) + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "GOST-Config-Reporter/1.0") + + client := &http.Client{ + Timeout: 10 * time.Second, // 配置上报可以稍长一些 + } + + resp, err := client.Do(req) + if err != nil { + return false, fmt.Errorf("发送HTTP请求失败: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return false, fmt.Errorf("HTTP响应错误: %d %s", resp.StatusCode, resp.Status) + } + + // 读取响应内容 + var responseBytes bytes.Buffer + _, err = responseBytes.ReadFrom(resp.Body) + if err != nil { + return false, fmt.Errorf("读取响应内容失败: %v", err) + } + + responseText := strings.TrimSpace(responseBytes.String()) + + // 检查响应是否为"ok" + if responseText == "ok" { + return true, nil + } else { + return false, fmt.Errorf("服务器响应: %s (期望: ok)", responseText) + } +} + +// StartConfigReporter 启动配置定时上报器（每10分钟上报一次） +func StartConfigReporter(ctx context.Context) { + if configReportURL == "" { + fmt.Printf("⚠️ 配置上报URL未设置，跳过定时上报\n") + return + } + + fmt.Printf("🚀 配置定时上报器已启动，每10分钟上报一次（WebSocket连接稳定后启动）\n") + + // 创建10分钟定时器 + ticker := time.NewTicker(10 * time.Minute) + defer ticker.Stop() + + // 立即执行一次配置上报 + go func() { + success, err := sendConfigReport(ctx) + if err != nil { + fmt.Printf("❌ 初始配置上报失败: %v\n", err) + } else if success { + fmt.Printf("✅ 初始配置上报成功\n") + } + }() + + // 定时上报循环 + for { + select { + case <-ticker.C: + go func() { + success, err := sendConfigReport(ctx) + if err != nil { + fmt.Printf("❌ 定时配置上报失败: %v\n", err) + } else if success { + fmt.Printf("✅ 定时配置上报成功\n") + } + }() + + case <-ctx.Done(): + fmt.Printf("⏹️ 配置定时上报器已停止\n") + return + } + } +} + +// serviceStatus 接口定义 +type serviceStatus interface { + Status() *Status +} + +// getConfigResponse 配置响应结构 +type getConfigResponse struct { + Config *config.Config `json:"config"` +} + +// getConfigData 获取配置数据（避免循环依赖） +func getConfigData() ([]byte, error) { + config.OnUpdate(func(c *config.Config) error { + for _, svc := range c.Services { + if svc == nil { + continue + } + s := registry.ServiceRegistry().Get(svc.Name) + ss, ok := s.(serviceStatus) + if ok && ss != nil { + status := ss.Status() + svc.Status = &config.ServiceStatus{ + CreateTime: status.CreateTime().Unix(), + State: string(status.State()), + } + if st := status.Stats(); st != nil { + svc.Status.Stats = &config.ServiceStats{ + TotalConns: st.Get(stats.KindTotalConns), + CurrentConns: st.Get(stats.KindCurrentConns), + TotalErrs: st.Get(stats.KindTotalErrs), + InputBytes: st.Get(stats.KindInputBytes), + OutputBytes: st.Get(stats.KindOutputBytes), + } + } + for _, ev := range status.Events() { + if !ev.Time.IsZero() { + svc.Status.Events = append(svc.Status.Events, config.ServiceEvent{ + Time: ev.Time.Unix(), + Msg: ev.Message, + }) + } + } + } + } + return nil + }) + + var resp getConfigResponse + resp.Config = config.Global() + + buf := &bytes.Buffer{} + resp.Config.Write(buf, "json") + return buf.Bytes(), nil +} diff --git a/go-gost/x/socket/config.go b/go-gost/x/socket/config.go index e3d37db..1fa1758 100644 --- a/go-gost/x/socket/config.go +++ b/go-gost/x/socket/config.go @@ -1,13 +1,9 @@ package socket import ( - "bytes" "os" - "github.com/go-gost/core/observer/stats" "github.com/go-gost/x/config" - "github.com/go-gost/x/registry" - "github.com/go-gost/x/service" ) func saveConfig() { @@ -27,57 +23,3 @@ func saveConfig() { return } - -type serviceStatus interface { - Status() *service.Status -} - -type getConfigResponse struct { - Config *config.Config -} - -func getConfig() ([]byte, error) { - - config.OnUpdate(func(c *config.Config) error { - for _, svc := range c.Services { - if svc == nil { - continue - } - s := registry.ServiceRegistry().Get(svc.Name) - ss, ok := s.(serviceStatus) - if ok && ss != nil { - status := ss.Status() - svc.Status = &config.ServiceStatus{ - CreateTime: status.CreateTime().Unix(), - State: string(status.State()), - } - if st := status.Stats(); st != nil { - svc.Status.Stats = &config.ServiceStats{ - TotalConns: st.Get(stats.KindTotalConns), - CurrentConns: st.Get(stats.KindCurrentConns), - TotalErrs: st.Get(stats.KindTotalErrs), - InputBytes: st.Get(stats.KindInputBytes), - OutputBytes: st.Get(stats.KindOutputBytes), - } - } - for _, ev := range status.Events() { - if !ev.Time.IsZero() { - svc.Status.Events = append(svc.Status.Events, config.ServiceEvent{ - Time: ev.Time.Unix(), - Msg: ev.Message, - }) - } - } - } - } - return nil - }) - - var resp getConfigResponse - resp.Config = config.Global() - - buf := &bytes.Buffer{} - - resp.Config.Write(buf, "json") - return buf.Bytes(), nil -} diff --git a/go-gost/x/socket/websocket_reporter.go b/go-gost/x/socket/websocket_reporter.go index 2db3e2c..f66544b 100644 --- a/go-gost/x/socket/websocket_reporter.go +++ b/go-gost/x/socket/websocket_reporter.go @@ -160,9 +160,6 @@ func (w *WebSocketReporter) connect() error { w.conn = conn w.connected = true - // 设置最大消息大小为 10MB (默认是 1024 * 1024) - w.conn.SetReadLimit(10 * 1024 * 1024) - // 设置关闭处理器来检测连接状态 w.conn.SetCloseHandler(func(code int, text string) error { w.connected = false @@ -185,9 +182,6 @@ func (w *WebSocketReporter) handleConnection() { // 启动消息接收goroutine go w.receiveMessages() - // 启动配置上报goroutine - go w.reportConfig() - // 主发送循环 ticker := time.NewTicker(w.pingInterval) defer ticker.Stop() @@ -347,14 +341,14 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt // routeCommand 路由命令到对应的处理函数 func (w *WebSocketReporter) routeCommand(cmd CommandMessage) { - jsonBytes, errs := json.MarshalIndent(cmd, "", " ") - if errs != nil { - fmt.Println("Error marshaling JSON:", errs) - return - } - - // 打印 JSON 字符串 - fmt.Println("🔔 收到命令: ", string(jsonBytes)) + //jsonBytes, errs := json.MarshalIndent(cmd, "", " ") + //if errs != nil { + // fmt.Println("Error marshaling JSON:", errs) + // return + //} + // + //// 打印 JSON 字符串 + //fmt.Println("🔔 收到命令: ", string(jsonBytes)) var err error var response CommandResponse @@ -674,116 +668,6 @@ func (w *WebSocketReporter) handleCall(data interface{}) error { return nil } -// reportConfig 定时上报配置信息 -func (w *WebSocketReporter) reportConfig() { - // 立即发送一次配置 - w.sendConfigReport() - - // 启动定时器 - ticker := time.NewTicker(w.configInterval) - defer ticker.Stop() - - for { - select { - case <-w.ctx.Done(): - return - case <-ticker.C: - if w.connected { - w.sendConfigReport() - } - } - } -} - -// sendConfigReport 发送配置报告 -func (w *WebSocketReporter) sendConfigReport() { - if w.conn == nil || !w.connected { - return - } - - // 获取配置数据 - configData, err := getConfig() - if err != nil { - fmt.Printf("❌ 获取配置失败: %v\n", err) - return - } - - // 检查数据大小，如果超过1MB则压缩 - if len(configData) > 1024*1024 { - fmt.Printf("📦 配置数据较大 (%.2f MB)，进行压缩处理\n", float64(len(configData))/(1024*1024)) - - // 压缩配置数据 - var compressedBuf bytes.Buffer - gzipWriter := gzip.NewWriter(&compressedBuf) - if _, err := gzipWriter.Write(configData); err != nil { - fmt.Printf("❌ 压缩配置数据失败: %v\n", err) - return - } - gzipWriter.Close() - - // 构建压缩后的配置报告消息 - configMsg := struct { - Type string `json:"type"` - Compressed bool `json:"compressed"` - Data []byte `json:"data"` - }{ - Type: "config_report", - Compressed: true, - Data: compressedBuf.Bytes(), - } - - // 转换为JSON - jsonData, err := json.Marshal(configMsg) - if err != nil { - fmt.Printf("❌ 序列化压缩配置报告失败: %v\n", err) - return - } - - fmt.Printf("✅ 压缩后大小: %.2f MB -> %.2f MB (压缩率: %.1f%%)\n", - float64(len(configData))/(1024*1024), - float64(len(jsonData))/(1024*1024), - 100.0-float64(len(jsonData))/float64(len(configData))*100) - - // 设置写入超时 - w.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) - - if err := w.conn.WriteMessage(websocket.TextMessage, jsonData); err != nil { - fmt.Printf("❌ 发送压缩配置报告失败: %v\n", err) - w.connected = false - return - } - } else { - // 数据较小，直接发送 - // 构建配置报告消息 - configMsg := struct { - Type string `json:"type"` - Compressed bool `json:"compressed"` - Data interface{} `json:"data"` - }{ - Type: "config_report", - Compressed: false, - Data: json.RawMessage(configData), - } - - // 转换为JSON - jsonData, err := json.Marshal(configMsg) - if err != nil { - fmt.Printf("❌ 序列化配置报告失败: %v\n", err) - return - } - - // 设置写入超时 - w.conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) - - if err := w.conn.WriteMessage(websocket.TextMessage, jsonData); err != nil { - fmt.Printf("❌ 发送配置报告失败: %v\n", err) - w.connected = false - return - } - } - -} - // sendResponse 发送响应消息到服务端 func (w *WebSocketReporter) sendResponse(response CommandResponse) { if w.conn == nil || !w.connected { diff --git a/go-gost/x/traffic/memory_manager.go b/go-gost/x/traffic/memory_manager.go deleted file mode 100644 index b5a1584..0000000 --- a/go-gost/x/traffic/memory_manager.go +++ /dev/null @@ -1,118 +0,0 @@ -package traffic - -import ( - "context" - "sync" - "sync/atomic" -) - -// MemoryManager 内存流量管理器 -type MemoryManager struct { - mu sync.RWMutex - stats map[string]*trafficStats -} - -// trafficStats 流量统计 -type trafficStats struct { - upload atomic.Int64 - download atomic.Int64 -} - -// NewMemoryManager 创建内存流量管理器 -func NewMemoryManager() *MemoryManager { - return &MemoryManager{ - stats: make(map[string]*trafficStats), - } -} - -// RecordTraffic 记录流量 -func (m *MemoryManager) RecordTraffic(ctx context.Context, service string, upload, download int64) error { - m.mu.RLock() - stats, exists := m.stats[service] - m.mu.RUnlock() - - if !exists { - m.mu.Lock() - stats, exists = m.stats[service] - if !exists { - stats = &trafficStats{} - m.stats[service] = stats - } - m.mu.Unlock() - } - - if upload > 0 { - stats.upload.Add(upload) - } - if download > 0 { - stats.download.Add(download) - } - - return nil -} - -// GetAllServicesStats 获取所有服务的流量统计 -func (m *MemoryManager) GetAllServicesStats(ctx context.Context) (map[string]map[string]int64, error) { - m.mu.RLock() - defer m.mu.RUnlock() - - result := make(map[string]map[string]int64) - for service, stats := range m.stats { - result[service] = map[string]int64{ - "upload": stats.upload.Load(), - "download": stats.download.Load(), - } - } - - return result, nil -} - -// ClearAllTrafficStats 清零所有流量统计 -func (m *MemoryManager) ClearAllTrafficStats(ctx context.Context) error { - m.mu.RLock() - defer m.mu.RUnlock() - - for _, stats := range m.stats { - stats.upload.Store(0) - stats.download.Store(0) - } - - return nil -} - -// SubtractTrafficStats 从指定服务的流量统计中减去给定的值 -func (m *MemoryManager) SubtractTrafficStats(ctx context.Context, stats map[string]map[string]int64) error { - m.mu.RLock() - defer m.mu.RUnlock() - - for service, serviceStats := range stats { - if managerStats, exists := m.stats[service]; exists { - if upload, ok := serviceStats["upload"]; ok && upload > 0 { - managerStats.upload.Add(-upload) - // 确保不会变成负数 - if managerStats.upload.Load() < 0 { - managerStats.upload.Store(0) - } - } - if download, ok := serviceStats["download"]; ok && download > 0 { - managerStats.download.Add(-download) - // 确保不会变成负数 - if managerStats.download.Load() < 0 { - managerStats.download.Store(0) - } - } - } - } - - return nil -} - -// Close 关闭管理器（内存管理器无需特殊清理） -func (m *MemoryManager) Close() error { - return nil -} - -// TestConnection 测试连接（内存管理器总是返回成功） -func (m *MemoryManager) TestConnection(ctx context.Context) error { - return nil -} diff --git a/go-gost/x/traffic/traffic.go b/go-gost/x/traffic/traffic.go deleted file mode 100644 index 49a470e..0000000 --- a/go-gost/x/traffic/traffic.go +++ /dev/null @@ -1,34 +0,0 @@ -package traffic - -import ( - "context" - "sync" -) - -// Manager 流量管理器接口 -type Manager interface { - RecordTraffic(ctx context.Context, service string, upload, download int64) error - GetAllServicesStats(ctx context.Context) (map[string]map[string]int64, error) - ClearAllTrafficStats(ctx context.Context) error - SubtractTrafficStats(ctx context.Context, stats map[string]map[string]int64) error - Close() error - TestConnection(ctx context.Context) error -} - -var ( - globalManager Manager - once sync.Once -) - -// GetGlobalManager 获取全局流量管理器实例 -func GetGlobalManager() Manager { - once.Do(func() { - globalManager = NewMemoryManager() - }) - return globalManager -} - -// SetGlobalManager 设置全局流量管理器（用于测试） -func SetGlobalManager(m Manager) { - globalManager = m -} diff --git a/go-gost/x/traffic/traffic_reporter.go b/go-gost/x/traffic/traffic_reporter.go deleted file mode 100644 index 37b6fa7..0000000 --- a/go-gost/x/traffic/traffic_reporter.go +++ /dev/null @@ -1,190 +0,0 @@ -package traffic - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "net/http" - "strings" - "time" -) - -// 全局变量存储HTTP地址 -var httpReportURL string - -// TrafficReportItem 流量报告项（压缩格式） -type TrafficReportItem struct { - N string `json:"n"` // 服务名（name缩写） - T string `json:"t"` // 连接类型：conn, cc（type缩写） - U int64 `json:"u"` // 上行流量（up缩写） - D int64 `json:"d"` // 下行流量（down缩写） -} - -// SetHTTPReportURL 设置HTTP报告地址 -func SetHTTPReportURL(addr string, secret string) { - httpReportURL = "http://" + addr + "/flow/upload?secret=" + secret -} - -// StartTrafficReporter 启动流量报告任务 -func StartTrafficReporter(trafficMgr Manager) { - // 检查是否设置了HTTP地址 - if httpReportURL == "" { - fmt.Println("❌ HTTP报告地址未设置，无法启动流量报告任务") - return - } - - ticker := time.NewTicker(5 * time.Second) - - go func() { - defer ticker.Stop() - - for range ticker.C { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - - // 先获取流量统计（不清零） - stats, err := trafficMgr.GetAllServicesStats(ctx) - if err != nil { - fmt.Printf("获取流量统计失败: %v\n", err) - cancel() - continue - } - - if len(stats) == 0 { - cancel() - continue - } - - // 构建报告数据为数组格式，分别处理不同类型的流量 - var reportItems []TrafficReportItem - totalServices := 0 - totalTraffic := int64(0) - - // 解析服务和类型，分组处理 - serviceGroups := make(map[string]map[string]map[string]int64) // service -> type -> direction -> bytes - - for serviceKey, serviceStats := range stats { - upload := serviceStats["upload"] - download := serviceStats["download"] - - // 解析服务名和类型 (format: service:type) - var serviceName, serviceType string - parts := strings.Split(serviceKey, ":") - if len(parts) >= 2 { - serviceName = parts[0] - serviceType = parts[1] - } else { - serviceName = serviceKey - serviceType = "unknown" - } - - if serviceGroups[serviceName] == nil { - serviceGroups[serviceName] = make(map[string]map[string]int64) - } - if serviceGroups[serviceName][serviceType] == nil { - serviceGroups[serviceName][serviceType] = make(map[string]int64) - } - - serviceGroups[serviceName][serviceType]["upload"] = upload - serviceGroups[serviceName][serviceType]["download"] = download - - totalTraffic += upload + download - } - - // 为每个服务的每种类型创建报告项 - for serviceName, types := range serviceGroups { - for serviceType, trafficData := range types { - // 过滤掉total类型，只保留conn和cc - if serviceType == "total" { - continue - } - - upload := trafficData["upload"] - download := trafficData["download"] - - // 只有当有流量时才报告 - if upload > 0 || download > 0 { - reportItems = append(reportItems, TrafficReportItem{ - N: serviceName, - T: serviceType, - U: upload, - D: download, - }) - totalServices++ - } - } - } - - // 只有当有流量数据时才发送 - if len(reportItems) > 0 && totalTraffic > 0 { - // 发送到HTTP接口 - success, err := sendTrafficReport(ctx, reportItems) - if err != nil { - fmt.Printf("发送流量报告失败: %v\n", err) - } else if success { - // 只有收到"ok"响应才减去已上报的流量 - err = trafficMgr.SubtractTrafficStats(ctx, stats) - if err != nil { - fmt.Printf("减去已上报流量失败: %v\n", err) - } else { - fmt.Printf("✅ 流量报告已发送并减去已上报流量: %d个记录, 总流量: %d bytes\n", - totalServices, totalTraffic) - } - } else { - fmt.Printf("⚠️ 服务器未确认(非ok响应)，保留流量数据: %d个记录, 总流量: %d bytes\n", - totalServices, totalTraffic) - } - } - - cancel() - } - }() - - fmt.Printf("🚀 流量报告任务已启动 (5秒间隔)，目标地址: %s\n", httpReportURL) -} - -// sendTrafficReport 发送流量报告到HTTP接口 -func sendTrafficReport(ctx context.Context, reportItems []TrafficReportItem) (bool, error) { - jsonData, err := json.Marshal(reportItems) - if err != nil { - return false, fmt.Errorf("序列化报告数据失败: %v", err) - } - - req, err := http.NewRequestWithContext(ctx, "POST", httpReportURL, bytes.NewBuffer(jsonData)) - if err != nil { - return false, fmt.Errorf("创建HTTP请求失败: %v", err) - } - - req.Header.Set("Content-Type", "application/json") - req.Header.Set("User-Agent", "GOST-Traffic-Reporter/1.0") - - client := &http.Client{ - Timeout: 5 * time.Second, - } - - resp, err := client.Do(req) - if err != nil { - return false, fmt.Errorf("发送HTTP请求失败: %v", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return false, fmt.Errorf("HTTP响应错误: %d %s", resp.StatusCode, resp.Status) - } - - // 读取响应内容 - var responseBytes bytes.Buffer - _, err = responseBytes.ReadFrom(resp.Body) - if err != nil { - return false, fmt.Errorf("读取响应内容失败: %v", err) - } - - responseText := strings.TrimSpace(responseBytes.String()) - - // 检查响应是否为"ok" - if responseText == "ok" { - return true, nil - } else { - return false, fmt.Errorf("服务器响应: %s (期望: ok)", responseText) - } -} diff --git a/springboot-backend/src/main/java/com/admin/common/dto/UserUpdateDto.java b/springboot-backend/src/main/java/com/admin/common/dto/UserUpdateDto.java index 4849673..7a47f05 100644 --- a/springboot-backend/src/main/java/com/admin/common/dto/UserUpdateDto.java +++ b/springboot-backend/src/main/java/com/admin/common/dto/UserUpdateDto.java @@ -12,9 +12,6 @@ public class UserUpdateDto { @NotNull(message = "用户ID不能为空") private Long id; - @NotBlank(message = "姓名不能为空") - private String name; - @NotBlank(message = "用户名不能为空") private String user; diff --git a/springboot-backend/src/main/java/com/admin/common/task/CheckGostConfigAsync.java b/springboot-backend/src/main/java/com/admin/common/task/CheckGostConfigAsync.java index fce9bb7..3e13962 100644 --- a/springboot-backend/src/main/java/com/admin/common/task/CheckGostConfigAsync.java +++ b/springboot-backend/src/main/java/com/admin/common/task/CheckGostConfigAsync.java @@ -2,6 +2,7 @@ package com.admin.common.task; import com.admin.common.dto.ConfigItem; import com.admin.common.dto.GostConfigDto; +import com.admin.common.dto.GostDto; import com.admin.common.utils.GostUtil; import com.admin.entity.Forward; import com.admin.entity.Node; @@ -74,7 +75,8 @@ public class CheckGostConfigAsync { Forward forward = forwardService.getById(forwardId); if (forward == null) { log.warn("删除孤立的服务: {} (节点: {})", service.getName(), node.getId()); - GostUtil.DeleteService(node.getId(), forwardId+"_"+userId+"_"+userTunnelId); + GostDto gostDto = GostUtil.DeleteService(node.getId(), forwardId + "_" + userId + "_" + userTunnelId); + System.out.println(gostDto); } } if (Objects.equals(type, "tls")) { diff --git a/springboot-backend/src/main/java/com/admin/common/utils/WebSocketServer.java b/springboot-backend/src/main/java/com/admin/common/utils/WebSocketServer.java index 703d153..f121f13 100644 --- a/springboot-backend/src/main/java/com/admin/common/utils/WebSocketServer.java +++ b/springboot-backend/src/main/java/com/admin/common/utils/WebSocketServer.java @@ -30,9 +30,6 @@ public class WebSocketServer extends TextWebSocketHandler { @Resource NodeService nodeService; - @Resource - CheckGostConfigAsync checkGostConfigAsync; - // 存储所有活跃的 WebSocket 连接（ private static final CopyOnWriteArraySet activeSessions = new CopyOnWriteArraySet<>(); @@ -57,13 +54,7 @@ public class WebSocketServer extends TextWebSocketHandler { if (message.getPayload().contains("memory_usage")){ // 先发送确认消息 sendToUser(session, "{\"type\":\"call\"}"); - } else if (message.getPayload().contains("config_report")) { - log.info("收到消息: {}", message.getPayload()); - JSONObject jsonObject = JSONObject.parseObject(message.getPayload()); - String string = jsonObject.getString("data"); - GostConfigDto gostConfigDto = JSONObject.parseObject(string, GostConfigDto.class); - checkGostConfigAsync.cleanNodeConfigs(id, gostConfigDto); - } else if (message.getPayload().contains("requestId")) { + }else if (message.getPayload().contains("requestId")) { log.info("收到消息: {}", message.getPayload()); // 处理命令响应消息 try { diff --git a/springboot-backend/src/main/java/com/admin/config/WebSocketConfig.java b/springboot-backend/src/main/java/com/admin/config/WebSocketConfig.java index 5bec7e7..51035f0 100644 --- a/springboot-backend/src/main/java/com/admin/config/WebSocketConfig.java +++ b/springboot-backend/src/main/java/com/admin/config/WebSocketConfig.java @@ -23,17 +23,6 @@ public class WebSocketConfig implements WebSocketConfigurer { @Resource private WebSocketInterceptor webSocketInterceptor; - @Bean - public ServletContextInitializer websocketBufferConfig() { - return new ServletContextInitializer() { - @Override - public void onStartup(ServletContext servletContext) throws ServletException { - servletContext.setInitParameter("org.apache.tomcat.websocket.textBufferSize", String.valueOf(10 * 1024 * 1024)); - servletContext.setInitParameter("org.apache.tomcat.websocket.binaryBufferSize", String.valueOf(10 * 1024 * 1024)); - } - }; - } - @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry webSocketHandlerRegistry) { webSocketHandlerRegistry diff --git a/springboot-backend/src/main/java/com/admin/controller/FlowController.java b/springboot-backend/src/main/java/com/admin/controller/FlowController.java index a192e07..569d473 100644 --- a/springboot-backend/src/main/java/com/admin/controller/FlowController.java +++ b/springboot-backend/src/main/java/com/admin/controller/FlowController.java @@ -2,16 +2,16 @@ package com.admin.controller; import com.admin.common.aop.LogAnnotation; import com.admin.common.dto.FlowDto; +import com.admin.common.dto.GostConfigDto; import com.admin.common.lang.R; +import com.admin.common.task.CheckGostConfigAsync; import com.admin.common.utils.GostUtil; import com.admin.entity.*; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper; -import org.springframework.web.bind.annotation.CrossOrigin; -import org.springframework.web.bind.annotation.RequestBody; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.bind.annotation.*; +import javax.annotation.Resource; import java.math.BigDecimal; import java.util.List; import java.util.Objects; @@ -56,6 +56,19 @@ public class FlowController extends BaseController { private static final ConcurrentHashMap FORWARD_LOCKS = new ConcurrentHashMap<>(); + @Resource + CheckGostConfigAsync checkGostConfigAsync; + + @PostMapping("/config") + @LogAnnotation + public String config(@RequestBody GostConfigDto gostConfigDto, String secret) { + Node node = nodeService.getOne(new QueryWrapper().eq("secret", secret)); + if (node == null) return SUCCESS_RESPONSE; + checkGostConfigAsync.cleanNodeConfigs(node.getId().toString(), gostConfigDto); + return SUCCESS_RESPONSE; + } + + @RequestMapping("/test") @LogAnnotation public String test() { diff --git a/springboot-backend/src/main/java/com/admin/service/impl/NodeServiceImpl.java b/springboot-backend/src/main/java/com/admin/service/impl/NodeServiceImpl.java index 666191d..d84d9ec 100644 --- a/springboot-backend/src/main/java/com/admin/service/impl/NodeServiceImpl.java +++ b/springboot-backend/src/main/java/com/admin/service/impl/NodeServiceImpl.java @@ -11,10 +11,12 @@ import com.admin.entity.Tunnel; import com.admin.mapper.NodeMapper; import com.admin.mapper.TunnelMapper; import com.admin.service.NodeService; +import com.admin.service.TunnelService; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import org.springframework.beans.BeanUtils; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import javax.annotation.Resource; @@ -58,6 +60,10 @@ public class NodeServiceImpl extends ServiceImpl implements No @Resource private TunnelMapper tunnelMapper; + @Resource + @Lazy + private TunnelService tunnelService; + @Value("${server-addr}") private String serverAddr; @@ -107,7 +113,25 @@ public class NodeServiceImpl extends ServiceImpl implements No // 2. 构建更新对象并执行更新 Node updateNode = buildUpdateNode(nodeUpdateDto); boolean result = this.updateById(updateNode); - + + // 更新隧道入口ip + List inNodeId = tunnelService.list(new QueryWrapper().eq("in_node_id", updateNode.getId())); + if (!inNodeId.isEmpty()) { + for (Tunnel tunnel : inNodeId) { + tunnel.setInIp(updateNode.getIp()); + } + tunnelService.updateBatchById(inNodeId); + } + + // 更新服务器出口ip + List outNodeId = tunnelService.list(new QueryWrapper().eq("out_node_id", updateNode.getId())); + if (!outNodeId.isEmpty()) { + for (Tunnel tunnel : outNodeId) { + tunnel.setOutIp(updateNode.getServerIp()); + } + tunnelService.updateBatchById(outNodeId); + } + return result ? R.ok(SUCCESS_UPDATE_MSG) : R.err(ERROR_UPDATE_MSG); } diff --git a/springboot-backend/src/main/resources/application.yml b/springboot-backend/src/main/resources/application.yml index 351eb1a..3176ded 100644 --- a/springboot-backend/src/main/resources/application.yml +++ b/springboot-backend/src/main/resources/application.yml @@ -26,7 +26,6 @@ server: uri-encoding: UTF-8 max-thread: 800 max-connections: 2000 - max-swallow-size: 10MB shutdown: graceful mybatis-plus: diff --git a/vue-frontend/src/views/Index.vue b/vue-frontend/src/views/Index.vue index ae7743e..430b5e5 100644 --- a/vue-frontend/src/views/Index.vue +++ b/vue-frontend/src/views/Index.vue @@ -92,7 +92,7 @@
@@ -185,7 +185,7 @@
@@ -1085,12 +1085,11 @@ export default { /* 隧道权限空状态卡片样式 */ .tunnel-card.empty-card { grid-column: 1 / -1; - justify-self: center; - max-width: 400px; - min-height: 160px; + width: 100%; display: flex; align-items: center; justify-content: center; + padding: 20px; } /* 转发详情的空状态样式由EmptyState组件内置处理 */ @@ -1172,9 +1171,9 @@ export default { } .tunnel-card.empty-card { - max-width: 100%; - min-height: 140px; + width: 100%; margin-bottom: 12px; + padding: 16px; } .tunnel-header { diff --git a/vue-frontend/src/views/User.vue b/vue-frontend/src/views/User.vue index c38ef3d..b5bf34f 100644 --- a/vue-frontend/src/views/User.vue +++ b/vue-frontend/src/views/User.vue @@ -1652,13 +1652,24 @@ export default { /* 移动端分页优化 */ .mobile-pagination { - justify-content: center; + display: flex; + justify-content: space-between; + align-items: center; + } + + .mobile-pagination >>> .btn-prev { + flex-shrink: 0; } .mobile-pagination >>> .el-pager { display: flex; - flex-wrap: wrap; + flex: 1; justify-content: center; + margin: 0 8px; + } + + .mobile-pagination >>> .btn-next { + flex-shrink: 0; } .mobile-pagination >>> .el-pager li {