diff --git a/api/config.go b/api/config.go index 996bd313..cd58b177 100644 --- a/api/config.go +++ b/api/config.go @@ -17,6 +17,35 @@ type serviceStatus interface { Status() *service.Status } +func fillServiceStatus(svc *config.ServiceConfig) { + 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, + }) + } + } + } +} + // swagger:parameters getConfigRequest type getConfigRequest struct { // output format, one of yaml|json, default is json. @@ -49,32 +78,7 @@ func getConfig(ctx *gin.Context) { 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, - }) - } - } - } + fillServiceStatus(svc) } return nil }) diff --git a/api/config_service.go b/api/config_service.go index 17004f36..cfc3940f 100644 --- a/api/config_service.go +++ b/api/config_service.go @@ -43,6 +43,13 @@ func getServiceList(ctx *gin.Context) { list := config.Global().Services + for _, svc := range list { + if svc == nil { + continue + } + fillServiceStatus(svc) + } + var resp getServiceListResponse resp.Data = serviceList{ Count: len(list), @@ -89,6 +96,7 @@ func getService(ctx *gin.Context) { continue } if service.Name == req.Service { + fillServiceStatus(service) resp.Data = service } } diff --git a/handler/http/handler.go b/handler/http/handler.go index f9bc1209..0c09bfe1 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -559,18 +559,21 @@ func (h *httpHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: - // Retry previously failed events first to preserve ordering. + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - break } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/http2/handler.go b/handler/http2/handler.go index cb2fae5f..62aaabe9 100644 --- a/handler/http2/handler.go +++ b/handler/http2/handler.go @@ -194,17 +194,21 @@ func (h *http2Handler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - break } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/masque/handler.go b/handler/masque/handler.go index ce6eb3b9..f01c570b 100644 --- a/handler/masque/handler.go +++ b/handler/masque/handler.go @@ -576,17 +576,21 @@ func (h *masqueHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - break } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = append(events, evs...) + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/relay/observe.go b/handler/relay/observe.go index b2c9803a..0781c10a 100644 --- a/handler/relay/observe.go +++ b/handler/relay/observe.go @@ -33,17 +33,21 @@ func (h *relayHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - // fallthrough: still collect fresh events for this tick } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/router/associate.go b/handler/router/associate.go index 581f646d..864645e6 100644 --- a/handler/router/associate.go +++ b/handler/router/associate.go @@ -52,7 +52,9 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw if !rid.Equal(routerID) { resp.Status = relay.StatusHostUnreachable - resp.WriteTo(conn) + if _, werr := resp.WriteTo(conn); werr != nil { + log.Error(werr) + } err := fmt.Errorf("no route to host %s", host) log.Error(err) return err @@ -62,7 +64,9 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw uuid, err := uuid.NewRandom() if err != nil { resp.Status = relay.StatusInternalServerError - resp.WriteTo(conn) + if _, werr := resp.WriteTo(conn); werr != nil { + log.Error(werr) + } return } connectorID := relay.NewConnectorID(uuid[:]) @@ -72,7 +76,9 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw ID: connectorID, }, ) - resp.WriteTo(conn) + if _, werr := resp.WriteTo(conn); werr != nil { + log.Error(werr) + } conn = &packetConn{conn} @@ -203,7 +209,9 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID if c := h.pool.Get(routerID, route.Gateway); c != nil { if w := c.Writer(); w != nil { - w.Write(data) + if _, werr := w.Write(data); werr != nil { + log.Error(werr) + } } return nil } @@ -231,7 +239,9 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID req.WriteTo(&buf) buf.Write(data) - h.epConn.WriteTo(buf.Bytes(), raddr) + if _, werr := h.epConn.WriteTo(buf.Bytes(), raddr); werr != nil { + log.Error(werr) + } return nil } diff --git a/handler/router/entrypoint.go b/handler/router/entrypoint.go index 8b794a4b..fd779f64 100644 --- a/handler/router/entrypoint.go +++ b/handler/router/entrypoint.go @@ -48,7 +48,9 @@ func (h *routerHandler) handleEntrypoint(log logger.Logger) error { if c := h.pool.Get(routerID, gateway); c != nil { if w := c.Writer(); w != nil { - w.Write(buf[nn:]) + if _, werr := w.Write(buf[nn:]); werr != nil { + log.Error(werr) + } } } diff --git a/handler/router/handler.go b/handler/router/handler.go index d0f0f214..f63efb2c 100644 --- a/handler/router/handler.go +++ b/handler/router/handler.go @@ -273,17 +273,22 @@ func (h *routerHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - break } + // Collect and send fresh events. evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + if len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/router/router.go b/handler/router/router.go index 24dd1bc5..f4a74621 100644 --- a/handler/router/router.go +++ b/handler/router/router.go @@ -1,10 +1,8 @@ package router import ( - "context" "io" "sync" - "time" "github.com/go-gost/core/logger" "github.com/go-gost/relay" @@ -76,7 +74,6 @@ type Router struct { node string id relay.TunnelID connectors map[string][]*Connector - t time.Time close chan struct{} mu sync.RWMutex } @@ -86,7 +83,6 @@ func NewRouter(node string, rid relay.TunnelID) *Router { node: node, id: rid, connectors: make(map[string][]*Connector), - t: time.Now(), close: make(chan struct{}), } return r @@ -139,8 +135,8 @@ func (r *Router) GetConnector(host string) *Connector { } func (r *Router) DelConnector(host string, cid relay.ConnectorID) { - r.mu.RLock() - defer r.mu.RUnlock() + r.mu.Lock() + defer r.mu.Unlock() connectors := r.connectors[host] for i, c := range connectors { @@ -174,7 +170,6 @@ type ConnectorPool struct { node string routers map[relay.TunnelID]*Router mu sync.RWMutex - cancel context.CancelFunc } func NewConnectorPool(node string) *ConnectorPool { @@ -219,8 +214,8 @@ func (p *ConnectorPool) Del(rid relay.TunnelID, host string, cid relay.Connector return } - p.mu.RLock() - defer p.mu.RUnlock() + p.mu.Lock() + defer p.mu.Unlock() r := p.routers[rid] if r == nil { @@ -235,10 +230,6 @@ func (p *ConnectorPool) Close() error { return nil } - if p.cancel != nil { - p.cancel() - } - p.mu.Lock() defer p.mu.Unlock() diff --git a/handler/socks/v4/handler.go b/handler/socks/v4/handler.go index 203c8ca7..8fb9efdf 100644 --- a/handler/socks/v4/handler.go +++ b/handler/socks/v4/handler.go @@ -365,17 +365,21 @@ func (h *socks4Handler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - break } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/socks/v5/handler.go b/handler/socks/v5/handler.go index 8931b546..028cba47 100644 --- a/handler/socks/v5/handler.go +++ b/handler/socks/v5/handler.go @@ -230,17 +230,21 @@ func (h *socks5Handler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - break } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/tungo/handler.go b/handler/tungo/handler.go index b3f4d92d..1d6c39c8 100644 --- a/handler/tungo/handler.go +++ b/handler/tungo/handler.go @@ -187,17 +187,21 @@ func (h *tungoHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } - break } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index 96d64295..996b049e 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -364,21 +364,21 @@ func (h *tunnelHandler) observeStats(ctx context.Context) { for { select { case <-ticker.C: + // Try to flush any buffered events from a previous failed attempt. if len(events) > 0 { - if err := h.options.Observer.Observe(ctx, events); err == nil { - events = nil - // Retry succeeded — also flush new events this tick. - // Fall through without break to reach the normal path. - } else { - // Retry still failing — skip new events, try again next tick. - break + if err := h.options.Observer.Observe(ctx, events); err != nil { + continue } } - evs := h.stats.Events() - if err := h.options.Observer.Observe(ctx, evs); err != nil { - events = evs + // Collect and send fresh events. + if evs := h.stats.Events(); len(evs) > 0 { + if err := h.options.Observer.Observe(ctx, evs); err != nil { + events = evs + continue + } } + events = nil case <-ctx.Done(): return