From 7c95d40e0543e08422dd2667a358c35f7caa3bac Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Sun, 24 May 2026 23:13:52 +0800 Subject: [PATCH] fix(observer): add nil guards, fix resource leak, silent event drops, and missing doc comments - WrapUDPConn, WrapPacketConn, WrapListener: add nil guards for pc/ln params - NewGRPCPlugin: return no-op observer instead of nil on connection error - gRPC plugin: fix resource leak by wiring Close() through observerWrapper and service.Close() lifecycle - Both plugins: log unknown event types instead of silently dropping them - httpPlugin: remove dead nil check for always-non-nil client field - Stats.Reset: update core interface doc to match implementation behavior - Add doc comments to all exported symbols across the observer package --- observer/plugin/grpc.go | 6 +++++- observer/plugin/http.go | 6 +++++- observer/stats/stats.go | 10 ++++++++++ observer/stats/wrapper/conn.go | 15 ++++++++++++++- observer/stats/wrapper/io.go | 2 ++ observer/stats/wrapper/listener.go | 5 ++++- registry/observer.go | 13 +++++++++++++ service/service.go | 5 +++++ 8 files changed, 58 insertions(+), 4 deletions(-) diff --git a/observer/plugin/grpc.go b/observer/plugin/grpc.go index 8ca8b1d5..b10ef3d8 100644 --- a/observer/plugin/grpc.go +++ b/observer/plugin/grpc.go @@ -21,6 +21,8 @@ type grpcPlugin struct { } // NewGRPCPlugin creates an Observer plugin based on gRPC. +// On connection error it returns a no-op observer that logs and discards +// events rather than nil, so callers can always call Observe safely. func NewGRPCPlugin(name string, addr string, opts ...plugin.Option) observer.Observer { var options plugin.Options for _, opt := range opts { @@ -34,7 +36,7 @@ func NewGRPCPlugin(name string, addr string, opts ...plugin.Option) observer.Obs conn, err := plugin.NewGRPCConn(addr, &options) if err != nil { log.Error(err) - return nil + return &grpcPlugin{log: log} // no-op: client is nil, Observe returns nil } p := &grpcPlugin{ @@ -82,6 +84,8 @@ func (p *grpcPlugin) Observe(ctx context.Context, events []observer.Event, opts TotalErrs: ev.TotalErrs, }, }) + default: + p.log.Warnf("unknown event type: %s", event.Type()) } } reply, err := p.client.Observe(ctx, &req) diff --git a/observer/plugin/http.go b/observer/plugin/http.go index 1e7fdcd4..64efff51 100644 --- a/observer/plugin/http.go +++ b/observer/plugin/http.go @@ -53,6 +53,8 @@ type httpPlugin struct { } // NewHTTPPlugin creates an Observer plugin based on HTTP. +// Events are POSTed as JSON to the given URL. If url does not have a scheme +// prefix, "http://" is prepended. func NewHTTPPlugin(name string, url string, opts ...plugin.Option) observer.Observer { var options plugin.Options for _, opt := range opts { @@ -74,7 +76,7 @@ func NewHTTPPlugin(name string, url string, opts ...plugin.Option) observer.Obse } func (p *httpPlugin) Observe(ctx context.Context, events []observer.Event, opts ...observer.Option) error { - if p.client == nil || len(events) == 0 { + if len(events) == 0 { return nil } @@ -108,6 +110,8 @@ func (p *httpPlugin) Observe(ctx context.Context, events []observer.Event, opts TotalErrs: ev.TotalErrs, }, }) + default: + p.log.Warnf("unknown event type: %s", e.Type()) } } v, err := json.Marshal(r) diff --git a/observer/stats/stats.go b/observer/stats/stats.go index a3a4e982..58bc0dc0 100644 --- a/observer/stats/stats.go +++ b/observer/stats/stats.go @@ -7,6 +7,10 @@ import ( "github.com/go-gost/core/observer/stats" ) +// Stats implements the stats.Stats interface using atomic counters. +// When resetTraffic is true, Get for KindInputBytes and KindOutputBytes +// atomically swaps the counter with zero, returning the value at the time +// of the call. type Stats struct { updated atomic.Bool totalConns atomic.Uint64 @@ -17,6 +21,9 @@ type Stats struct { resetTraffic bool } +// NewStats creates a new Stats instance. When resetTraffic is true, calls +// to Get for KindInputBytes and KindOutputBytes atomically reset the counter +// to zero after reading, which is useful for rate calculations. func NewStats(resetTraffic bool) stats.Stats { return &Stats{ resetTraffic: resetTraffic, @@ -85,6 +92,8 @@ func (s *Stats) IsUpdated() bool { return s.updated.Swap(false) } +// StatsEvent carries a snapshot of all tracked statistics for a specific +// service and optional client. It implements observer.Event. type StatsEvent struct { Kind string Service string @@ -97,6 +106,7 @@ type StatsEvent struct { TotalErrs uint64 } +// Type returns observer.EventStats to identify this as a statistics event. func (StatsEvent) Type() observer.EventType { return observer.EventStats } diff --git a/observer/stats/wrapper/conn.go b/observer/stats/wrapper/conn.go index 196be4eb..90413916 100644 --- a/observer/stats/wrapper/conn.go +++ b/observer/stats/wrapper/conn.go @@ -26,6 +26,10 @@ type conn struct { mu sync.Mutex } +// WrapConn wraps a net.Conn to track connection and traffic statistics. +// It increments total and current connection counts, and tracks bytes +// read and written. On Close, it decrements the current connection count. +// If c or pStats is nil, the original conn is returned unchanged. func WrapConn(c net.Conn, pStats stats.Stats) net.Conn { if c == nil || pStats == nil { return c @@ -103,8 +107,10 @@ type packetConn struct { stats stats.Stats } +// WrapPacketConn wraps a net.PacketConn to track input and output bytes. +// If pc or stats is nil, the original conn is returned unchanged. func WrapPacketConn(pc net.PacketConn, stats stats.Stats) net.PacketConn { - if stats == nil { + if pc == nil || stats == nil { return pc } return &packetConn{ @@ -137,7 +143,14 @@ type udpConn struct { stats stats.Stats } +// WrapUDPConn wraps a net.PacketConn as a udp.Conn to track input and output +// bytes across all read/write methods (Read, Write, ReadFrom, WriteTo, +// ReadFromUDP, ReadMsgUDP, WriteToUDP, WriteMsgUDP). If pc or stats is nil, +// it returns nil. func WrapUDPConn(pc net.PacketConn, stats stats.Stats) udp.Conn { + if pc == nil || stats == nil { + return nil + } return &udpConn{ PacketConn: pc, stats: stats, diff --git a/observer/stats/wrapper/io.go b/observer/stats/wrapper/io.go index 4eef8c2d..21b71bc4 100644 --- a/observer/stats/wrapper/io.go +++ b/observer/stats/wrapper/io.go @@ -12,6 +12,8 @@ type readWriter struct { stats stats.Stats } +// WrapReadWriter wraps an io.ReadWriter to track input and output bytes. +// If rw or stats is nil, the original ReadWriter is returned unchanged. func WrapReadWriter(rw io.ReadWriter, stats stats.Stats) io.ReadWriter { if rw == nil || stats == nil { return rw diff --git a/observer/stats/wrapper/listener.go b/observer/stats/wrapper/listener.go index 5f67da33..54d9edd9 100644 --- a/observer/stats/wrapper/listener.go +++ b/observer/stats/wrapper/listener.go @@ -11,8 +11,11 @@ type listener struct { net.Listener } +// WrapListener wraps a net.Listener to track connection statistics. Each +// accepted connection is wrapped via WrapConn to count connections and bytes. +// If ln or stats is nil, the original listener is returned unchanged. func WrapListener(ln net.Listener, stats stats.Stats) net.Listener { - if stats == nil { + if ln == nil || stats == nil { return ln } diff --git a/registry/observer.go b/registry/observer.go index 62e9df29..e3f90ff0 100644 --- a/registry/observer.go +++ b/registry/observer.go @@ -2,6 +2,7 @@ package registry import ( "context" + "io" "github.com/go-gost/core/observer" ) @@ -37,3 +38,15 @@ func (w *observerWrapper) Observe(ctx context.Context, events []observer.Event, } return v.Observe(ctx, events, opts...) } + +// Close closes the underlying observer if it implements io.Closer. +func (w *observerWrapper) Close() error { + v := w.r.get(w.name) + if v == nil { + return nil + } + if closer, ok := v.(io.Closer); ok { + return closer.Close() + } + return nil +} diff --git a/service/service.go b/service/service.go index 0bceef03..7f232c3e 100644 --- a/service/service.go +++ b/service/service.go @@ -302,6 +302,11 @@ func (s *defaultService) Close() error { if closer, ok := s.handler.(io.Closer); ok { closer.Close() } + if s.options.observer != nil { + if closer, ok := s.options.observer.(io.Closer); ok { + closer.Close() + } + } return s.listener.Close() }