add traffic reset for stats

config/parsing/service/parse.go

@@ -31,6 +31,7 @@ import (
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	"github.com/go-gost/x/metadata"
 	mdutil "github.com/go-gost/x/metadata/util"
+	xstats "github.com/go-gost/x/observer/stats"
 	"github.com/go-gost/x/registry"
 	xservice "github.com/go-gost/x/service"
 	"github.com/vishvananda/netns"
@@ -101,8 +102,8 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
 	ifce := cfg.Interface
 	var preUp, preDown, postUp, postDown []string
 	var ignoreChain bool
-	var pStats *stats.Stats
+	var pStats stats.Stats
-	var observePeriod time.Duration
+	var observerPeriod time.Duration
 	var netnsIn, netnsOut string
 	var dialTimeout time.Duration
 	if cfg.Metadata != nil {
@@ -123,9 +124,10 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
 		ignoreChain = mdutil.GetBool(md, parsing.MDKeyIgnoreChain)
 
 		if mdutil.GetBool(md, parsing.MDKeyEnableStats) {
-			pStats = &stats.Stats{}
+			pStats = xstats.NewStats(mdutil.GetBool(md, "observer.resetTraffic"))
 		}
-		observePeriod = mdutil.GetDuration(md, "observePeriod")
+		observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period")
+
 		netnsIn = mdutil.GetString(md, "netns")
 		netnsOut = mdutil.GetString(md, "netns.out")
 		dialTimeout = mdutil.GetDuration(md, "dialTimeout")
@@ -318,7 +320,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) {
 		xservice.RecordersOption(recorders...),
 		xservice.StatsOption(pStats),
 		xservice.ObserverOption(registry.ObserverRegistry().Get(cfg.Observer)),
-		xservice.ObservePeriodOption(observePeriod),
+		xservice.ObserverPeriodOption(observerPeriod),
 		xservice.LoggerOption(serviceLogger),
 	)
 

dialer/http3/dialer.go

@@ -70,7 +70,7 @@ func (d *http3Dialer) Dial(ctx context.Context, addr string, opts ...dialer.Dial
 			Host: host,
 			Client: &http.Client{
 				// Timeout:   60 * time.Second,
-				Transport: &http3.RoundTripper{
+				Transport: &http3.Transport{
 					TLSClientConfig: d.options.TLSConfig,
 					Dial: func(ctx context.Context, adr string, tlsCfg *tls.Config, cfg *quic.Config) (quic.EarlyConnection, error) {
 						// d.options.Logger.Infof("dial: %s/%s, %s", addr, network, host)

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d
 	github.com/gin-contrib/cors v1.7.2
 	github.com/gin-gonic/gin v1.10.0
-	github.com/go-gost/core v0.1.11
+	github.com/go-gost/core v0.2.0
 	github.com/go-gost/gosocks4 v0.0.1
 	github.com/go-gost/gosocks5 v0.4.2
 	github.com/go-gost/plugin v0.1.1
@@ -45,7 +45,7 @@ require (
 	github.com/zalando/go-keyring v0.2.4
 	golang.org/x/crypto v0.31.0
 	golang.org/x/exp v0.0.0-20241210194714-1829a127f884
-	golang.org/x/net v0.32.0
+	golang.org/x/net v0.33.0
 	golang.org/x/sys v0.28.0
 	golang.org/x/text v0.21.0
 	golang.org/x/time v0.5.0

go.sum

@@ -51,8 +51,8 @@ github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE
 github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
 github.com/gin-gonic/gin v1.10.0 h1:nTuyha1TYqgedzytsKYqna+DfLos46nTv2ygFy86HFU=
 github.com/gin-gonic/gin v1.10.0/go.mod h1:4PMNQiOhvDRa013RKVbsiNwoyezlm2rm0uX/T7kzp5Y=
-github.com/go-gost/core v0.1.11 h1:7zmRx0stWrClUJM/IT/ltRQJC9wUc8f6TAj1fbBiSPs=
+github.com/go-gost/core v0.2.0 h1:hZf94DAv93iwKXXzGXBmspE5bExYcBJeLl6inL9mG6Y=
-github.com/go-gost/core v0.1.11/go.mod h1:WGI43jOka7FAsSAwi/fSMaqxdR+E339ycb4NBGlFr6A=
+github.com/go-gost/core v0.2.0/go.mod h1:WGI43jOka7FAsSAwi/fSMaqxdR+E339ycb4NBGlFr6A=
 github.com/go-gost/gosocks4 v0.0.1 h1:+k1sec8HlELuQV7rWftIkmy8UijzUt2I6t+iMPlGB2s=
 github.com/go-gost/gosocks4 v0.0.1/go.mod h1:3B6L47HbU/qugDg4JnoFPHgJXE43Inz8Bah1QaN9qCc=
 github.com/go-gost/gosocks5 v0.4.2 h1:IianxHTkACPqCwiOAT3MHoMdSUl+SEPSRu1ikawC1Pc=
@@ -305,8 +305,8 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.32.0 h1:ZqPmj8Kzc+Y6e0+skZsuACbx+wzMgo5MQsJh9Qd6aYI=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
-golang.org/x/net v0.32.0/go.mod h1:CwU0IoeOlnQQWJ6ioyFrfRuomB8GKF6KbYXZVyeXNfs=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

handler/dns/handler.go

@@ -22,6 +22,7 @@ import (
 	xhop "github.com/go-gost/x/hop"
 	resolver_util "github.com/go-gost/x/internal/util/resolver"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -163,7 +164,7 @@ func (h *dnsHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
 
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/forward/local/handler.go

@@ -21,6 +21,7 @@ import (
 	"github.com/go-gost/x/internal/util/sniffing"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -114,7 +115,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand
 	}
 	ro.Network = network
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/forward/remote/handler.go

@@ -24,6 +24,7 @@ import (
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
 	mdutil "github.com/go-gost/x/metadata/util"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -116,7 +117,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand
 	}
 	ro.Network = network
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/http/handler.go

@@ -26,6 +26,7 @@ import (
 	"github.com/go-gost/core/limiter/traffic"
 	"github.com/go-gost/core/logger"
 	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer"
 	"github.com/go-gost/core/observer/stats"
 	"github.com/go-gost/core/recorder"
 	xbypass "github.com/go-gost/x/bypass"
@@ -41,6 +42,7 @@ import (
 	ws_util "github.com/go-gost/x/internal/util/ws"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
 	traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -83,7 +85,7 @@ func (h *httpHandler) Init(md md.Metadata) error {
 	h.cancel = cancel
 
 	if h.options.Observer != nil {
-		h.stats = stats_util.NewHandlerStats(h.options.Service)
+		h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
 		go h.observeStats(ctx)
 	}
 
@@ -143,7 +145,7 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {
@@ -420,7 +422,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt
 }
 
 func (h *httpHandler) handleProxy(ctx context.Context, conn net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error {
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	ro.Time = time.Time{}
@@ -452,7 +454,7 @@ func (h *httpHandler) handleProxy(ctx context.Context, conn net.Conn, req *http.
 	}
 }
 
-func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats *stats.Stats, log logger.Logger) (close bool, err error) {
+func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) {
 	close = true
 
 	ro2 := &xrecorder.HandlerRecorderObject{}
@@ -949,13 +951,26 @@ func (h *httpHandler) observeStats(ctx context.Context) {
 		return
 	}
 
-	ticker := time.NewTicker(h.md.observePeriod)
+	var events []observer.Event
+
+	ticker := time.NewTicker(h.md.observerPeriod)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			h.options.Observer.Observe(ctx, h.stats.Events())
+			if len(events) > 0 {
+				if err := h.options.Observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
+				break
+			}
+
+			evs := h.stats.Events()
+			if err := h.options.Observer.Observe(ctx, evs); err != nil {
+				events = evs
+			}
+
 		case <-ctx.Done():
 			return
 		}

handler/http/metadata.go

@@ -29,9 +29,11 @@ type metadata struct {
 	hash                   string
 	authBasicRealm         string
 	proxyAgent             string
-	observePeriod          time.Duration
 	limiterRefreshInterval time.Duration
 
+	observerPeriod       time.Duration
+	observerResetTraffic bool
+
 	sniffing                    bool
 	sniffingTimeout             time.Duration
 	sniffingWebsocket           bool
@@ -73,14 +75,16 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error {
 	h.md.hash = mdutil.GetString(md, "hash")
 	h.md.authBasicRealm = mdutil.GetString(md, "authBasicRealm")
 
-	h.md.observePeriod = mdutil.GetDuration(md, "observePeriod", "observer.observePeriod")
+	h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
-	if h.md.observePeriod == 0 {
+	if h.md.observerPeriod == 0 {
-		h.md.observePeriod = 5 * time.Second
+		h.md.observerPeriod = 5 * time.Second
 	}
-	if h.md.observePeriod < time.Second {
+	if h.md.observerPeriod < time.Second {
-		h.md.observePeriod = time.Second
+		h.md.observerPeriod = time.Second
 	}
 
+	h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
+
 	h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
 	if h.md.limiterRefreshInterval == 0 {
 		h.md.limiterRefreshInterval = 30 * time.Second

handler/http2/handler.go

@@ -23,6 +23,7 @@ import (
 	"github.com/go-gost/core/limiter/traffic"
 	"github.com/go-gost/core/logger"
 	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer"
 	"github.com/go-gost/core/observer/stats"
 	"github.com/go-gost/core/recorder"
 	xbypass "github.com/go-gost/x/bypass"
@@ -72,7 +73,7 @@ func (h *http2Handler) Init(md md.Metadata) error {
 	h.cancel = cancel
 
 	if h.options.Observer != nil {
-		h.stats = stats_util.NewHandlerStats(h.options.Service)
+		h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
 		go h.observeStats(ctx)
 	}
 
@@ -483,13 +484,26 @@ func (h *http2Handler) observeStats(ctx context.Context) {
 		return
 	}
 
-	ticker := time.NewTicker(h.md.observePeriod)
+	var events []observer.Event
+
+	ticker := time.NewTicker(h.md.observerPeriod)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			h.options.Observer.Observe(ctx, h.stats.Events())
+			if len(events) > 0 {
+				if err := h.options.Observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
+				break
+			}
+
+			evs := h.stats.Events()
+			if err := h.options.Observer.Observe(ctx, evs); err != nil {
+				events = evs
+			}
+
 		case <-ctx.Done():
 			return
 		}

handler/http2/metadata.go

@@ -18,7 +18,8 @@ type metadata struct {
 	header                 http.Header
 	hash                   string
 	authBasicRealm         string
-	observePeriod          time.Duration
+	observerPeriod         time.Duration
+	observerResetTraffic   bool
 	limiterRefreshInterval time.Duration
 }
 
@@ -43,13 +44,14 @@ func (h *http2Handler) parseMetadata(md mdata.Metadata) error {
 	h.md.hash = mdutil.GetString(md, "hash")
 	h.md.authBasicRealm = mdutil.GetString(md, "authBasicRealm")
 
-	h.md.observePeriod = mdutil.GetDuration(md, "observePeriod", "observer.observePeriod")
+	h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
-	if h.md.observePeriod == 0 {
+	if h.md.observerPeriod == 0 {
-		h.md.observePeriod = 5 * time.Second
+		h.md.observerPeriod = 5 * time.Second
 	}
-	if h.md.observePeriod < time.Second {
+	if h.md.observerPeriod < time.Second {
-		h.md.observePeriod = time.Second
+		h.md.observerPeriod = time.Second
 	}
+	h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
 
 	h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
 	if h.md.limiterRefreshInterval == 0 {

handler/redirect/tcp/handler.go

@@ -17,6 +17,7 @@ import (
 	xbypass "github.com/go-gost/x/bypass"
 	ctxvalue "github.com/go-gost/x/ctx"
 	xnet "github.com/go-gost/x/internal/net"
+	xstats "github.com/go-gost/x/observer/stats"
 	"github.com/go-gost/x/internal/util/sniffing"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
@@ -90,7 +91,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/redirect/udp/handler.go

@@ -15,6 +15,7 @@ import (
 	ctxvalue "github.com/go-gost/x/ctx"
 	xnet "github.com/go-gost/x/internal/net"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -78,7 +79,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han
 
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/relay/handler.go

@@ -11,6 +11,7 @@ import (
 	"github.com/go-gost/core/hop"
 	"github.com/go-gost/core/limiter/traffic"
 	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer"
 	"github.com/go-gost/core/observer/stats"
 	"github.com/go-gost/core/recorder"
 	"github.com/go-gost/relay"
@@ -19,6 +20,7 @@ import (
 	stats_util "github.com/go-gost/x/internal/util/stats"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -65,7 +67,7 @@ func (h *relayHandler) Init(md md.Metadata) (err error) {
 	h.cancel = cancel
 
 	if h.options.Observer != nil {
-		h.stats = stats_util.NewHandlerStats(h.options.Service)
+		h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
 		go h.observeStats(ctx)
 	}
 
@@ -122,7 +124,7 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle
 
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {
@@ -253,13 +255,26 @@ func (h *relayHandler) observeStats(ctx context.Context) {
 		return
 	}
 
-	ticker := time.NewTicker(h.md.observePeriod)
+	var events []observer.Event
+
+	ticker := time.NewTicker(h.md.observerPeriod)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			h.options.Observer.Observe(ctx, h.stats.Events())
+			if len(events) > 0 {
+				if err := h.options.Observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
+				break
+			}
+
+			evs := h.stats.Events()
+			if err := h.options.Observer.Observe(ctx, evs); err != nil {
+				events = evs
+			}
+
 		case <-ctx.Done():
 			return
 		}

handler/relay/metadata.go

@@ -21,9 +21,11 @@ type metadata struct {
 	noDelay                bool
 	hash                   string
 	muxCfg                 *mux.Config
-	observePeriod          time.Duration
 	limiterRefreshInterval time.Duration
 
+	observerPeriod       time.Duration
+	observerResetTraffic bool
+
 	sniffing                    bool
 	sniffingTimeout             time.Duration
 	sniffingWebsocket           bool
@@ -62,14 +64,16 @@ func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) {
 		MaxStreamBuffer:   mdutil.GetInt(md, "mux.maxStreamBuffer"),
 	}
 
-	h.md.observePeriod = mdutil.GetDuration(md, "observePeriod", "observer.observePeriod")
+	h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
-	if h.md.observePeriod == 0 {
+	if h.md.observerPeriod == 0 {
-		h.md.observePeriod = 5 * time.Second
+		h.md.observerPeriod = 5 * time.Second
 	}
-	if h.md.observePeriod < time.Second {
+	if h.md.observerPeriod < time.Second {
-		h.md.observePeriod = time.Second
+		h.md.observerPeriod = time.Second
 	}
 
+	h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
+
 	h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
 	if h.md.limiterRefreshInterval == 0 {
 		h.md.limiterRefreshInterval = 30 * time.Second

handler/sni/handler.go

@@ -18,6 +18,7 @@ import (
 	"github.com/go-gost/x/internal/util/sniffing"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -96,7 +97,7 @@ func (h *sniHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/socks/v4/handler.go

@@ -14,6 +14,7 @@ import (
 	"github.com/go-gost/core/limiter/traffic"
 	"github.com/go-gost/core/logger"
 	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer"
 	"github.com/go-gost/core/observer/stats"
 	"github.com/go-gost/core/recorder"
 	"github.com/go-gost/gosocks4"
@@ -25,6 +26,7 @@ import (
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
 	traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -70,7 +72,7 @@ func (h *socks4Handler) Init(md md.Metadata) (err error) {
 	h.cancel = cancel
 
 	if h.options.Observer != nil {
-		h.stats = stats_util.NewHandlerStats(h.options.Service)
+		h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
 		go h.observeStats(ctx)
 	}
 
@@ -122,7 +124,7 @@ func (h *socks4Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {
@@ -338,13 +340,26 @@ func (h *socks4Handler) observeStats(ctx context.Context) {
 		return
 	}
 
-	ticker := time.NewTicker(h.md.observePeriod)
+	var events []observer.Event
+
+	ticker := time.NewTicker(h.md.observerPeriod)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			h.options.Observer.Observe(ctx, h.stats.Events())
+			if len(events) > 0 {
+				if err := h.options.Observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
+				break
+			}
+
+			evs := h.stats.Events()
+			if err := h.options.Observer.Observe(ctx, evs); err != nil {
+				events = evs
+			}
+
 		case <-ctx.Done():
 			return
 		}

handler/socks/v4/metadata.go

@@ -15,9 +15,11 @@ import (
 type metadata struct {
 	readTimeout            time.Duration
 	hash                   string
-	observePeriod          time.Duration
 	limiterRefreshInterval time.Duration
 
+	observerPeriod       time.Duration
+	observerResetTraffic bool
+
 	sniffing                    bool
 	sniffingTimeout             time.Duration
 	sniffingWebsocket           bool
@@ -37,14 +39,16 @@ func (h *socks4Handler) parseMetadata(md mdata.Metadata) (err error) {
 
 	h.md.hash = mdutil.GetString(md, "hash")
 
-	h.md.observePeriod = mdutil.GetDuration(md, "observePeriod", "observer.observePeriod")
+	h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
-	if h.md.observePeriod == 0 {
+	if h.md.observerPeriod == 0 {
-		h.md.observePeriod = 5 * time.Second
+		h.md.observerPeriod = 5 * time.Second
 	}
-	if h.md.observePeriod < time.Second {
+	if h.md.observerPeriod < time.Second {
-		h.md.observePeriod = time.Second
+		h.md.observerPeriod = time.Second
 	}
 
+	h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
+
 	h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
 	if h.md.limiterRefreshInterval == 0 {
 		h.md.limiterRefreshInterval = 30 * time.Second

handler/socks/v5/handler.go

@@ -9,6 +9,7 @@ import (
 	"github.com/go-gost/core/handler"
 	"github.com/go-gost/core/limiter/traffic"
 	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer"
 	"github.com/go-gost/core/observer/stats"
 	"github.com/go-gost/core/recorder"
 	"github.com/go-gost/gosocks5"
@@ -18,6 +19,7 @@ import (
 	stats_util "github.com/go-gost/x/internal/util/stats"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -70,7 +72,7 @@ func (h *socks5Handler) Init(md md.Metadata) (err error) {
 	h.cancel = cancel
 
 	if h.options.Observer != nil {
-		h.stats = stats_util.NewHandlerStats(h.options.Service)
+		h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
 		go h.observeStats(ctx)
 	}
 
@@ -122,7 +124,7 @@ func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {
@@ -216,13 +218,26 @@ func (h *socks5Handler) observeStats(ctx context.Context) {
 		return
 	}
 
-	ticker := time.NewTicker(h.md.observePeriod)
+	var events []observer.Event
+
+	ticker := time.NewTicker(h.md.observerPeriod)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			h.options.Observer.Observe(ctx, h.stats.Events())
+			if len(events) > 0 {
+				if err := h.options.Observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
+				break
+			}
+
+			evs := h.stats.Events()
+			if err := h.options.Observer.Observe(ctx, evs); err != nil {
+				events = evs
+			}
+
 		case <-ctx.Done():
 			return
 		}

handler/socks/v5/metadata.go

@@ -23,9 +23,11 @@ type metadata struct {
 	compatibilityMode      bool
 	hash                   string
 	muxCfg                 *mux.Config
-	observePeriod          time.Duration
 	limiterRefreshInterval time.Duration
 
+	observerPeriod       time.Duration
+	observerResetTraffic bool
+
 	sniffing                    bool
 	sniffingTimeout             time.Duration
 	sniffingWebsocket           bool
@@ -66,13 +68,14 @@ func (h *socks5Handler) parseMetadata(md mdata.Metadata) (err error) {
 		MaxStreamBuffer:   mdutil.GetInt(md, "mux.maxStreamBuffer"),
 	}
 
-	h.md.observePeriod = mdutil.GetDuration(md, "observePeriod", "observer.observePeriod")
+	h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
-	if h.md.observePeriod == 0 {
+	if h.md.observerPeriod == 0 {
-		h.md.observePeriod = 5 * time.Second
+		h.md.observerPeriod = 5 * time.Second
 	}
-	if h.md.observePeriod < time.Second {
+	if h.md.observerPeriod < time.Second {
-		h.md.observePeriod = time.Second
+		h.md.observerPeriod = time.Second
 	}
+	h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
 
 	h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
 	if h.md.limiterRefreshInterval == 0 {

handler/ss/handler.go

@@ -20,6 +20,7 @@ import (
 	"github.com/go-gost/x/internal/util/ss"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -106,7 +107,7 @@ func (h *ssHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.H
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/sshd/handler.go

@@ -24,6 +24,7 @@ import (
 	sshd_util "github.com/go-gost/x/internal/util/sshd"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	rate_limiter "github.com/go-gost/x/limiter/rate"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -98,7 +99,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

handler/tunnel/bind.go

@@ -65,7 +65,7 @@ func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network,
 		return
 	}
 
-	var stats *stats.Stats
+	var stats stats.Stats
 	if h.stats != nil {
 		stats = h.stats.Stats(tunnelID.String())
 	}

handler/tunnel/entrypoint.go

@@ -38,6 +38,7 @@ import (
 	ws_util "github.com/go-gost/x/internal/util/ws"
 	climiter "github.com/go-gost/x/limiter/conn/wrapper"
 	metrics "github.com/go-gost/x/metrics/wrapper"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"golang.org/x/net/http/httpguts"
@@ -86,7 +87,7 @@ func (ep *entrypoint) Handle(ctx context.Context, conn net.Conn) (err error) {
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {
@@ -195,7 +196,7 @@ func (ep *entrypoint) dial(ctx context.Context, network, addr string) (conn net.
 }
 
 func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) {
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	br := bufio.NewReader(conn)
@@ -250,7 +251,7 @@ func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecord
 	}
 }
 
-func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats *stats.Stats, log logger.Logger) (err error) {
+func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (err error) {
 	ro2 := &xrecorder.HandlerRecorderObject{}
 	*ro2 = *ro
 	ro = ro2

handler/tunnel/handler.go

@@ -14,6 +14,7 @@ import (
 	"github.com/go-gost/core/listener"
 	"github.com/go-gost/core/logger"
 	md "github.com/go-gost/core/metadata"
+	"github.com/go-gost/core/observer"
 	"github.com/go-gost/core/service"
 	"github.com/go-gost/relay"
 	ctxvalue "github.com/go-gost/x/ctx"
@@ -116,7 +117,7 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) {
 	h.cancel = cancel
 
 	if h.options.Observer != nil {
-		h.stats = stats_util.NewHandlerStats(h.options.Service)
+		h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
 		go h.observeStats(ctx)
 	}
 
@@ -321,13 +322,26 @@ func (h *tunnelHandler) observeStats(ctx context.Context) {
 		return
 	}
 
-	ticker := time.NewTicker(h.md.observePeriod)
+	var events []observer.Event
+
+	ticker := time.NewTicker(h.md.observerPeriod)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			h.options.Observer.Observe(ctx, h.stats.Events())
+			if len(events) > 0 {
+				if err := h.options.Observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
+				break
+			}
+
+			evs := h.stats.Events()
+			if err := h.options.Observer.Observe(ctx, evs); err != nil {
+				events = evs
+			}
+
 		case <-ctx.Done():
 			return
 		}

handler/tunnel/metadata.go

@@ -36,8 +36,10 @@ type metadata struct {
 	ingress                ingress.Ingress
 	sd                     sd.SD
 	muxCfg                 *mux.Config
-	observePeriod          time.Duration
 	limiterRefreshInterval time.Duration
+
+	observerPeriod       time.Duration
+	observerResetTraffic bool
 }
 
 func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
@@ -105,13 +107,14 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
 		h.md.muxCfg.MaxStreamBuffer = 1048576
 	}
 
-	h.md.observePeriod = mdutil.GetDuration(md, "observePeriod", "observer.observePeriod")
+	h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
-	if h.md.observePeriod == 0 {
+	if h.md.observerPeriod == 0 {
-		h.md.observePeriod = 5 * time.Second
+		h.md.observerPeriod = 5 * time.Second
 	}
-	if h.md.observePeriod < time.Second {
+	if h.md.observerPeriod < time.Second {
-		h.md.observePeriod = time.Second
+		h.md.observerPeriod = time.Second
 	}
+	h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
 
 	h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
 	if h.md.limiterRefreshInterval == 0 {

handler/tunnel/tunnel.go

@@ -27,7 +27,7 @@ const (
 type ConnectorOptions struct {
 	service string
 	sd      sd.SD
-	stats   *stats.Stats
+	stats   stats.Stats
 	limiter traffic.TrafficLimiter
 }
 

handler/unix/handler.go

@@ -21,6 +21,7 @@ import (
 	xnet "github.com/go-gost/x/internal/net"
 	"github.com/go-gost/x/internal/util/sniffing"
 	tls_util "github.com/go-gost/x/internal/util/tls"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"github.com/go-gost/x/registry"
@@ -95,7 +96,7 @@ func (h *unixHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler
 	})
 	log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	defer func() {

internal/util/forwarder/sniffer.go

@@ -34,6 +34,7 @@ import (
 	"github.com/go-gost/x/internal/util/sniffing"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	ws_util "github.com/go-gost/x/internal/util/ws"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"golang.org/x/net/http/httpguts"
@@ -123,7 +124,7 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO
 		opt(&ho)
 	}
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	br := bufio.NewReader(conn)
@@ -349,7 +350,7 @@ func (h *Sniffer) serveH2(ctx context.Context, conn net.Conn, ho *HandleOptions)
 	return nil
 }
 
-func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node *chain.Node, req *http.Request, pStats *stats.Stats, ho *HandleOptions) (close bool, err error) {
+func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node *chain.Node, req *http.Request, pStats stats.Stats, ho *HandleOptions) (close bool, err error) {
 	close = true
 
 	log := ho.Log

internal/util/sniffing/sniffer.go

@@ -30,6 +30,7 @@ import (
 	xhttp "github.com/go-gost/x/internal/net/http"
 	tls_util "github.com/go-gost/x/internal/util/tls"
 	ws_util "github.com/go-gost/x/internal/util/ws"
+	xstats "github.com/go-gost/x/observer/stats"
 	stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
 	xrecorder "github.com/go-gost/x/recorder"
 	"golang.org/x/net/http/httpguts"
@@ -114,7 +115,7 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO
 		opt(&ho)
 	}
 
-	pStats := stats.Stats{}
+	pStats := xstats.Stats{}
 	conn = stats_wrapper.WrapConn(conn, &pStats)
 
 	br := bufio.NewReader(conn)
@@ -262,7 +263,7 @@ func (h *Sniffer) serveH2(ctx context.Context, conn net.Conn, ho *HandleOptions)
 	return nil
 }
 
-func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats *stats.Stats, log logger.Logger) (close bool, err error) {
+func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (close bool, err error) {
 	close = true
 
 	ro2 := &xrecorder.HandlerRecorderObject{}

internal/util/stats/stats.go

@@ -5,22 +5,25 @@ import (
 
 	"github.com/go-gost/core/observer"
 	"github.com/go-gost/core/observer/stats"
+	xstats "github.com/go-gost/x/observer/stats"
 )
 
 type HandlerStats struct {
 	service      string
-	stats   map[string]*stats.Stats
+	stats        map[string]stats.Stats
+	resetTraffic bool
 	mu           sync.RWMutex
 }
 
-func NewHandlerStats(service string) *HandlerStats {
+func NewHandlerStats(service string, resetTraffic bool) *HandlerStats {
 	return &HandlerStats{
 		service:      service,
-		stats:   make(map[string]*stats.Stats),
+		stats:        make(map[string]stats.Stats),
+		resetTraffic: resetTraffic,
 	}
 }
 
-func (p *HandlerStats) Stats(client string) *stats.Stats {
+func (p *HandlerStats) Stats(client string) stats.Stats {
 	p.mu.RLock()
 	pstats := p.stats[client]
 	p.mu.RUnlock()
@@ -32,7 +35,7 @@ func (p *HandlerStats) Stats(client string) *stats.Stats {
 	defer p.mu.Unlock()
 	pstats = p.stats[client]
 	if pstats == nil {
-		pstats = &stats.Stats{}
+		pstats = xstats.NewStats(p.resetTraffic)
 	}
 	p.stats[client] = pstats
 
@@ -47,7 +50,7 @@ func (p *HandlerStats) Events() (events []observer.Event) {
 		if !v.IsUpdated() {
 			continue
 		}
-		events = append(events, stats.StatsEvent{
+		events = append(events, xstats.StatsEvent{
 			Kind:         "handler",
 			Service:      p.service,
 			Client:       k,

observer/plugin/grpc.go

@@ -2,13 +2,14 @@ package observer
 
 import (
 	"context"
+	"errors"
 	"io"
 
 	"github.com/go-gost/core/logger"
 	"github.com/go-gost/core/observer"
-	"github.com/go-gost/core/observer/stats"
 	"github.com/go-gost/plugin/observer/proto"
 	"github.com/go-gost/x/internal/plugin"
+	xstats "github.com/go-gost/x/observer/stats"
 	"github.com/go-gost/x/service"
 	"google.golang.org/grpc"
 )
@@ -67,7 +68,7 @@ func (p *grpcPlugin) Observe(ctx context.Context, events []observer.Event, opts
 				},
 			})
 		case observer.EventStats:
-			ev := event.(stats.StatsEvent)
+			ev := event.(xstats.StatsEvent)
 			req.Events = append(req.Events, &proto.Event{
 				Kind:    ev.Kind,
 				Service: ev.Service,
@@ -83,11 +84,14 @@ func (p *grpcPlugin) Observe(ctx context.Context, events []observer.Event, opts
 			})
 		}
 	}
-	_, err := p.client.Observe(ctx, &req)
+	reply, err := p.client.Observe(ctx, &req)
 	if err != nil {
 		p.log.Error(err)
 		return err
 	}
+	if reply == nil || !reply.Ok {
+		return errors.New("observe failed")
+	}
 	return nil
 }
 

observer/plugin/http.go

@@ -4,14 +4,14 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"fmt"
+	"errors"
 	"net/http"
 	"strings"
 
 	"github.com/go-gost/core/logger"
 	"github.com/go-gost/core/observer"
-	"github.com/go-gost/core/observer/stats"
 	"github.com/go-gost/x/internal/plugin"
+	xstats "github.com/go-gost/x/observer/stats"
 	"github.com/go-gost/x/service"
 )
 
@@ -41,7 +41,7 @@ type statusEvent struct {
 	Msg   string `json:"msg"`
 }
 
-type observeResponse struct {
+type httpPluginResponse struct {
 	OK bool `json:"ok"`
 }
 
@@ -94,7 +94,7 @@ func (p *httpPlugin) Observe(ctx context.Context, events []observer.Event, opts
 				},
 			})
 		case observer.EventStats:
-			ev := e.(stats.StatsEvent)
+			ev := e.(xstats.StatsEvent)
 			r.Events = append(r.Events, event{
 				Kind:    ev.Kind,
 				Service: ev.Service,
@@ -131,7 +131,16 @@ func (p *httpPlugin) Observe(ctx context.Context, events []observer.Event, opts
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf(resp.Status)
+		return errors.New(resp.Status)
+	}
+
+	res := httpPluginResponse{}
+	if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
+		return err
+	}
+
+	if !res.OK {
+		return errors.New("observe failed")
 	}
 
 	return nil

observer/stats/stats.go

@@ -0,0 +1,102 @@
+package stats
+
+import (
+	"sync/atomic"
+
+	"github.com/go-gost/core/observer"
+	"github.com/go-gost/core/observer/stats"
+)
+
+type Stats struct {
+	updated      atomic.Bool
+	totalConns   atomic.Uint64
+	currentConns atomic.Uint64
+	inputBytes   atomic.Uint64
+	outputBytes  atomic.Uint64
+	totalErrs    atomic.Uint64
+	resetTraffic bool
+}
+
+func NewStats(resetTraffic bool) stats.Stats {
+	return &Stats{
+		resetTraffic: resetTraffic,
+	}
+}
+
+func (s *Stats) Add(kind stats.Kind, n int64) {
+	if s == nil {
+		return
+	}
+	switch kind {
+	case stats.KindTotalConns:
+		if n > 0 {
+			s.totalConns.Add(uint64(n))
+		}
+	case stats.KindCurrentConns:
+		s.currentConns.Add(uint64(n))
+	case stats.KindInputBytes:
+		s.inputBytes.Add(uint64(n))
+	case stats.KindOutputBytes:
+		s.outputBytes.Add(uint64(n))
+	case stats.KindTotalErrs:
+		if n > 0 {
+			s.totalErrs.Add(uint64(n))
+		}
+	}
+	s.updated.Store(true)
+}
+
+func (s *Stats) Get(kind stats.Kind) uint64 {
+	if s == nil {
+		return 0
+	}
+
+	switch kind {
+	case stats.KindTotalConns:
+		return s.totalConns.Load()
+	case stats.KindCurrentConns:
+		return s.currentConns.Load()
+	case stats.KindInputBytes:
+		if s.resetTraffic {
+			return s.inputBytes.Swap(0)
+		}
+		return s.inputBytes.Load()
+	case stats.KindOutputBytes:
+		if s.resetTraffic {
+			return s.outputBytes.Swap(0)
+		}
+		return s.outputBytes.Load()
+	case stats.KindTotalErrs:
+		return s.totalErrs.Load()
+	}
+	return 0
+}
+
+func (s *Stats) Reset() {
+	s.updated.Store(false)
+	s.totalConns.Store(0)
+	s.currentConns.Store(0)
+	s.inputBytes.Store(0)
+	s.outputBytes.Store(0)
+	s.totalErrs.Store(0)
+}
+
+func (s *Stats) IsUpdated() bool {
+	return s.updated.Swap(false)
+}
+
+type StatsEvent struct {
+	Kind    string
+	Service string
+	Client  string
+
+	TotalConns   uint64
+	CurrentConns uint64
+	InputBytes   uint64
+	OutputBytes  uint64
+	TotalErrs    uint64
+}
+
+func (StatsEvent) Type() observer.EventType {
+	return observer.EventStats
+}

observer/stats/wrapper/conn.go

@@ -19,12 +19,12 @@ var (
 
 type conn struct {
 	net.Conn
-	stats  *stats.Stats
+	stats  stats.Stats
 	closed chan struct{}
 	mu     sync.Mutex
 }
 
-func WrapConn(c net.Conn, pStats *stats.Stats) net.Conn {
+func WrapConn(c net.Conn, pStats stats.Stats) net.Conn {
 	if pStats == nil {
 		return c
 	}
@@ -84,10 +84,10 @@ func (c *conn) Metadata() metadata.Metadata {
 
 type packetConn struct {
 	net.PacketConn
-	stats *stats.Stats
+	stats stats.Stats
 }
 
-func WrapPacketConn(pc net.PacketConn, stats *stats.Stats) net.PacketConn {
+func WrapPacketConn(pc net.PacketConn, stats stats.Stats) net.PacketConn {
 	if stats == nil {
 		return pc
 	}
@@ -118,10 +118,10 @@ func (c *packetConn) Metadata() metadata.Metadata {
 
 type udpConn struct {
 	net.PacketConn
-	stats *stats.Stats
+	stats stats.Stats
 }
 
-func WrapUDPConn(pc net.PacketConn, stats *stats.Stats) udp.Conn {
+func WrapUDPConn(pc net.PacketConn, stats stats.Stats) udp.Conn {
 	return &udpConn{
 		PacketConn: pc,
 		stats:      stats,

observer/stats/wrapper/io.go

@@ -9,10 +9,10 @@ import (
 // readWriter is an io.ReadWriter with Stats.
 type readWriter struct {
 	io.ReadWriter
-	stats *stats.Stats
+	stats stats.Stats
 }
 
-func WrapReadWriter(rw io.ReadWriter, stats *stats.Stats) io.ReadWriter {
+func WrapReadWriter(rw io.ReadWriter, stats stats.Stats) io.ReadWriter {
 	if stats == nil {
 		return rw
 	}

observer/stats/wrapper/listener.go

@@ -7,11 +7,11 @@ import (
 )
 
 type listener struct {
-	stats *stats.Stats
+	stats stats.Stats
 	net.Listener
 }
 
-func WrapListener(ln net.Listener, stats *stats.Stats) net.Listener {
+func WrapListener(ln net.Listener, stats stats.Stats) net.Listener {
 	if stats == nil {
 		return ln
 	}

recorder/plugin/grpc.go

@@ -3,6 +3,7 @@ package recorder
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 
 	"github.com/go-gost/core/logger"
@@ -56,7 +57,7 @@ func (p *grpcPlugin) Record(ctx context.Context, b []byte, opts ...recorder.Reco
 
 	md, _ := json.Marshal(options.Metadata)
 
-	_, err := p.client.Record(ctx,
+	reply, err := p.client.Record(ctx,
 		&proto.RecordRequest{
 			Data:     b,
 			Metadata: md,
@@ -65,6 +66,9 @@ func (p *grpcPlugin) Record(ctx context.Context, b []byte, opts ...recorder.Reco
 		p.log.Error(err)
 		return err
 	}
+	if reply == nil || !reply.Ok {
+		return errors.New("record failed")
+	}
 	return nil
 }
 

recorder/plugin/http.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"net/http"
 
 	"github.com/go-gost/core/logger"
@@ -84,7 +83,7 @@ func (p *httpPlugin) Record(ctx context.Context, b []byte, opts ...recorder.Reco
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("%s", resp.Status)
+		return errors.New(resp.Status)
 	}
 
 	res := httpPluginResponse{}

service/service.go

@@ -21,6 +21,7 @@ import (
 	ctxvalue "github.com/go-gost/x/ctx"
 	xnet "github.com/go-gost/x/internal/net"
 	xmetrics "github.com/go-gost/x/metrics"
+	xstats "github.com/go-gost/x/observer/stats"
 	"github.com/rs/xid"
 )
 
@@ -31,9 +32,9 @@ type options struct {
 	postUp         []string
 	preDown        []string
 	postDown       []string
-	stats         *stats.Stats
+	stats          stats.Stats
 	observer       observer.Observer
-	observePeriod time.Duration
+	observerPeriod time.Duration
 	logger         logger.Logger
 }
 
@@ -75,7 +76,7 @@ func PostDownOption(cmds []string) Option {
 	}
 }
 
-func StatsOption(stats *stats.Stats) Option {
+func StatsOption(stats stats.Stats) Option {
 	return func(opts *options) {
 		opts.stats = stats
 	}
@@ -87,9 +88,9 @@ func ObserverOption(observer observer.Observer) Option {
 	}
 }
 
-func ObservePeriodOption(period time.Duration) Option {
+func ObserverPeriodOption(period time.Duration) Option {
 	return func(opts *options) {
-		opts.observePeriod = period
+		opts.observerPeriod = period
 	}
 }
 
@@ -249,7 +250,9 @@ func (s *defaultService) Serve() error {
 					metrics.Labels{"service": s.name, "client": clientIP}); v != nil {
 					v.Inc()
 				}
-				s.status.stats.Add(stats.KindTotalErrs, 1)
+				if sts := s.status.stats; sts != nil {
+					sts.Add(stats.KindTotalErrs, 1)
+				}
 			}
 		}()
 	}
@@ -307,10 +310,15 @@ func (s *defaultService) observeStats(ctx context.Context) {
 		return
 	}
 
-	d := s.options.observePeriod
+	d := s.options.observerPeriod
-	if d < time.Millisecond {
+	if d == 0 {
 		d = 5 * time.Second
 	}
+	if d < time.Second {
+		d = 1 * time.Second
+	}
+
+	var events []observer.Event
 
 	ticker := time.NewTicker(d)
 	defer ticker.Stop()
@@ -318,20 +326,33 @@ func (s *defaultService) observeStats(ctx context.Context) {
 	for {
 		select {
 		case <-ticker.C:
-			st := s.status.Stats()
+			if len(events) > 0 {
-			if !st.IsUpdated() {
+				if err := s.options.observer.Observe(ctx, events); err == nil {
+					events = nil
+				}
 				break
 			}
-			s.options.observer.Observe(ctx, []observer.Event{
+
-				stats.StatsEvent{
+			st := s.status.Stats()
+			if st == nil || !st.IsUpdated() {
+				break
+			}
+
+			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
+			}
+
 		case <-ctx.Done():
 			return
 		}

service/status.go

@@ -29,7 +29,7 @@ type Status struct {
 	createTime time.Time
 	state      State
 	events     []Event
-	stats      *stats.Stats
+	stats      stats.Stats
 	mu         sync.RWMutex
 }
 
@@ -71,7 +71,7 @@ func (p *Status) addEvent(event Event) {
 	p.events = append(p.events, event)
 }
 
-func (p *Status) Stats() *stats.Stats {
+func (p *Status) Stats() stats.Stats {
 	if p == nil {
 		return nil
 	}