From 1c07cf6394ae8004f9eece3f76d639a161b6451a Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Mon, 25 May 2026 23:47:47 +0800 Subject: [PATCH] fix(service): default nil logger, fix Events() allocation, collect Close errors - Default nil logger to Nop() in NewService to prevent panics - Fix Events() returning 20-element slice padded with zero values - Collect and join all errors in Close() instead of discarding - Remove unnecessary variable shadow in execCmds - Add doc comments to all 22 exported symbols - Add 34 tests covering status, serve loop, close, and edge cases --- service/service.go | 51 ++- service/service_test.go | 668 ++++++++++++++++++++++++++++++++++++++++ service/status.go | 28 +- service/status_test.go | 218 +++++++++++++ 4 files changed, 952 insertions(+), 13 deletions(-) create mode 100644 service/service_test.go create mode 100644 service/status_test.go diff --git a/service/service.go b/service/service.go index 7f232c3e..5decaa38 100644 --- a/service/service.go +++ b/service/service.go @@ -1,3 +1,6 @@ +// Package service implements the core [service.Service] interface, binding a +// [listener.Listener] and a [handler.Handler] into a runnable proxy service +// that accepts inbound connections and forwards traffic through a proxy chain. package service import ( @@ -22,6 +25,7 @@ import ( "github.com/go-gost/core/recorder" "github.com/go-gost/core/service" xctx "github.com/go-gost/x/ctx" + xlogger "github.com/go-gost/x/logger" xmetrics "github.com/go-gost/x/metrics" xstats "github.com/go-gost/x/observer/stats" "github.com/google/shlex" @@ -41,62 +45,76 @@ type options struct { logger logger.Logger } +// Option is a functional option for configuring a service. type Option func(opts *options) +// AdmissionOption sets the admission controller that filters connections by +// source address. func AdmissionOption(admission admission.Admission) Option { return func(opts *options) { opts.admission = admission } } +// RecordersOption sets the traffic recorders for the service. func RecordersOption(recorders ...recorder.RecorderObject) Option { return func(opts *options) { opts.recorders = recorders } } +// PreUpOption sets shell commands to run before the service starts. func PreUpOption(cmds []string) Option { return func(opts *options) { opts.preUp = cmds } } +// PreDownOption sets shell commands to run before the service stops. func PreDownOption(cmds []string) Option { return func(opts *options) { opts.preDown = cmds } } +// PostUpOption sets shell commands to run after the service starts listening. func PostUpOption(cmds []string) Option { return func(opts *options) { opts.postUp = cmds } } +// PostDownOption sets shell commands to run after the service stops. func PostDownOption(cmds []string) Option { return func(opts *options) { opts.postDown = cmds } } +// StatsOption sets the traffic statistics collector for the service. func StatsOption(stats stats.Stats) Option { return func(opts *options) { opts.stats = stats } } +// ObserverOption sets the observer that receives service state and stats events. func ObserverOption(observer observer.Observer) Option { return func(opts *options) { opts.observer = observer } } +// ObserverPeriodOption sets the interval at which stats are reported to the +// observer. Defaults to 5 seconds with a minimum of 1 second. func ObserverPeriodOption(period time.Duration) Option { return func(opts *options) { opts.observerPeriod = period } } +// LoggerOption sets the logger for the service. If not provided, a no-op +// logger is used. func LoggerOption(logger logger.Logger) Option { return func(opts *options) { opts.logger = logger @@ -111,11 +129,17 @@ type defaultService struct { options options } +// NewService creates a new service that binds the given listener and handler. +// The service is registered in the running state and pre-up commands are +// executed immediately. Call [Serve] to start accepting connections. func NewService(name string, ln listener.Listener, h handler.Handler, opts ...Option) service.Service { var options options for _, opt := range opts { opt(&options) } + if options.logger == nil { + options.logger = xlogger.Nop() + } s := &defaultService{ name: name, listener: ln, @@ -134,10 +158,14 @@ func NewService(name string, ln listener.Listener, h handler.Handler, opts ...Op return s } +// Addr returns the network address the service is listening on. func (s *defaultService) Addr() net.Addr { return s.listener.Addr() } +// Serve starts the accept loop. It blocks until the listener is closed or a +// fatal accept error occurs. Each accepted connection is handled in its own +// goroutine by the service's handler. func (s *defaultService) Serve() error { s.execCmds("post-up", s.options.postUp) s.setState(StateReady) @@ -291,28 +319,39 @@ func (s *defaultService) Serve() error { } } +// Status returns the runtime status of the service. func (s *defaultService) Status() *Status { return s.status } +// Close shuts down the service by closing the handler, observer, and listener +// in order. All errors are collected and returned as a joined error. func (s *defaultService) Close() error { s.execCmds("pre-down", s.options.preDown) defer s.execCmds("post-down", s.options.postDown) + var errs []error if closer, ok := s.handler.(io.Closer); ok { - closer.Close() + if err := closer.Close(); err != nil { + errs = append(errs, err) + } } if s.options.observer != nil { if closer, ok := s.options.observer.(io.Closer); ok { - closer.Close() + if err := closer.Close(); err != nil { + errs = append(errs, err) + } } } - return s.listener.Close() + if err := s.listener.Close(); err != nil { + errs = append(errs, err) + } + return errors.Join(errs...) } func (s *defaultService) execCmds(phase string, cmds []string) { - for _, cmd := range cmds { - cmd := strings.TrimSpace(cmd) + for _, raw := range cmds { + cmd := strings.TrimSpace(raw) if cmd == "" { continue } @@ -411,6 +450,7 @@ func (s *defaultService) observeStats(ctx context.Context) { } } +// ServiceEvent is an observer event representing a service state change. type ServiceEvent struct { Kind string Service string @@ -418,6 +458,7 @@ type ServiceEvent struct { Msg string } +// Type returns the observer event type for service events. func (ServiceEvent) Type() observer.EventType { return observer.EventStatus } diff --git a/service/service_test.go b/service/service_test.go new file mode 100644 index 00000000..6db0fc54 --- /dev/null +++ b/service/service_test.go @@ -0,0 +1,668 @@ +package service + +import ( + "context" + "errors" + "io" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/go-gost/core/admission" + "github.com/go-gost/core/handler" + "github.com/go-gost/core/listener" + "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" + xctx "github.com/go-gost/x/ctx" +) + +// --- Mocks --- + +type mockListener struct { + addr net.Addr + connCh chan net.Conn // connections to deliver via Accept + errCh chan error // errors to deliver via Accept + closed atomic.Bool + closeMu sync.Mutex + closeCh chan struct{} +} + +func newMockListener() *mockListener { + return &mockListener{ + addr: &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}, + connCh: make(chan net.Conn, 10), + errCh: make(chan error, 10), + closeCh: make(chan struct{}), + } +} + +func (l *mockListener) Init(metadata.Metadata) error { return nil } + +func (l *mockListener) Accept() (net.Conn, error) { + select { + case c := <-l.connCh: + return c, nil + case e := <-l.errCh: + return nil, e + case <-l.closeCh: + return nil, net.ErrClosed + } +} + +func (l *mockListener) Addr() net.Addr { return l.addr } + +func (l *mockListener) Close() error { + l.closeMu.Lock() + defer l.closeMu.Unlock() + if l.closed.CompareAndSwap(false, true) { + close(l.closeCh) + } + return nil +} + +type mockHandler struct { + handleFn func(ctx context.Context, conn net.Conn) error + closed atomic.Bool +} + +func newMockHandler(fn func(ctx context.Context, conn net.Conn) error) *mockHandler { + return &mockHandler{handleFn: fn} +} + +func (h *mockHandler) Init(metadata.Metadata) error { return nil } + +func (h *mockHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { + if h.handleFn != nil { + return h.handleFn(ctx, conn) + } + return nil +} + +func (h *mockHandler) Close() error { + h.closed.Store(true) + return nil +} + +type mockConn struct { + net.Conn + remoteAddr net.Addr + localAddr net.Addr + ctx context.Context + closed atomic.Bool +} + +func (c *mockConn) RemoteAddr() net.Addr { return c.remoteAddr } +func (c *mockConn) LocalAddr() net.Addr { return c.localAddr } +func (c *mockConn) Close() error { c.closed.Store(true); return nil } + +type mockAdmission struct { + allow bool + calls atomic.Int32 +} + +func (a *mockAdmission) Admit(ctx context.Context, network, addr string, opts ...admission.Option) bool { + a.calls.Add(1) + return a.allow +} + +type mockObserver struct { + events []observer.Event + mu sync.Mutex + err error + closed atomic.Bool +} + +func (o *mockObserver) Observe(ctx context.Context, events []observer.Event, opts ...observer.Option) error { + o.mu.Lock() + defer o.mu.Unlock() + o.events = append(o.events, events...) + return o.err +} + +func (o *mockObserver) Close() error { + o.closed.Store(true) + return nil +} + +func (o *mockObserver) getEvents() []observer.Event { + o.mu.Lock() + defer o.mu.Unlock() + out := make([]observer.Event, len(o.events)) + copy(out, o.events) + return out +} + +type mockRecorder struct { + recorded [][]byte + mu sync.Mutex + err error +} + +func (r *mockRecorder) Record(ctx context.Context, b []byte, opts ...recorder.RecordOption) error { + r.mu.Lock() + defer r.mu.Unlock() + cp := make([]byte, len(b)) + copy(cp, b) + r.recorded = append(r.recorded, cp) + return r.err +} + +func (r *mockRecorder) getRecorded() [][]byte { + r.mu.Lock() + defer r.mu.Unlock() + return r.recorded +} + +// --- NewService tests --- + +func TestNewServiceDefaultsNopLogger(t *testing.T) { + ln := newMockListener() + h := newMockHandler(nil) + svc := NewService("test", ln, h).(*defaultService) + // Should not panic when logging + svc.options.logger.Info("test") +} + +func TestNewServiceSetsRunningState(t *testing.T) { + ln := newMockListener() + h := newMockHandler(nil) + svc := NewService("test", ln, h).(*defaultService) + if svc.Status().State() != StateRunning { + t.Errorf("State() = %q, want %q", svc.Status().State(), StateRunning) + } +} + +func TestNewServiceRecordsCreateTime(t *testing.T) { + before := time.Now() + ln := newMockListener() + h := newMockHandler(nil) + svc := NewService("test", ln, h).(*defaultService) + after := time.Now() + + ct := svc.Status().CreateTime() + if ct.Before(before) || ct.After(after) { + t.Errorf("CreateTime = %v, want between %v and %v", ct, before, after) + } +} + +// --- Addr tests --- + +func TestAddr(t *testing.T) { + addr := &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 8080} + ln := newMockListener() + ln.addr = addr + svc := NewService("test", ln, newMockHandler(nil)).(*defaultService) + if got := svc.Addr(); got.String() != addr.String() { + t.Errorf("Addr() = %v, want %v", got, addr) + } +} + +// --- Serve tests --- + +func TestServeAcceptsConnection(t *testing.T) { + ln := newMockListener() + var handled atomic.Int32 + h := newMockHandler(func(ctx context.Context, conn net.Conn) error { + handled.Add(1) + return nil + }) + svc := NewService("test", ln, h).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("192.168.1.1"), Port: 1234}, + localAddr: ln.addr, + } + + // Give handler goroutine time to run + time.Sleep(100 * time.Millisecond) + + ln.Close() + <-done + + if handled.Load() != 1 { + t.Errorf("handled = %d, want 1", handled.Load()) + } +} + +func TestServeSetsReadyState(t *testing.T) { + ln := newMockListener() + svc := NewService("test", ln, newMockHandler(nil)).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + time.Sleep(50 * time.Millisecond) + if got := svc.Status().State(); got != StateReady { + t.Errorf("State() = %q, want %q", got, StateReady) + } + + ln.Close() + <-done +} + +func TestServeReturnsOnListenerClose(t *testing.T) { + ln := newMockListener() + svc := NewService("test", ln, newMockHandler(nil)).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.Close() + err := <-done + if !errors.Is(err, net.ErrClosed) { + t.Errorf("Serve() = %v, want net.ErrClosed", err) + } + + if svc.Status().State() != StateClosed { + t.Errorf("State() = %q, want %q", svc.Status().State(), StateClosed) + } +} + +func TestServeTemporaryErrorRetry(t *testing.T) { + ln := newMockListener() + svc := NewService("test", ln, newMockHandler(nil)).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + // Send a temporary error, then close + ln.errCh <- &tempErr{} + time.Sleep(50 * time.Millisecond) + + if svc.Status().State() != StateFailed { + t.Errorf("State() = %q, want %q", svc.Status().State(), StateFailed) + } + + ln.Close() + <-done +} + +type tempErr struct{} + +func (tempErr) Error() string { return "temporary error" } +func (tempErr) Temporary() bool { return true } +func (tempErr) Timeout() bool { return false } + +func TestServeAdmissionDeny(t *testing.T) { + ln := newMockListener() + var handled atomic.Int32 + h := newMockHandler(func(ctx context.Context, conn net.Conn) error { + handled.Add(1) + return nil + }) + adm := &mockAdmission{allow: false} + svc := NewService("test", ln, h, AdmissionOption(adm)).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999}, + localAddr: ln.addr, + } + + time.Sleep(100 * time.Millisecond) + ln.Close() + <-done + + if handled.Load() != 0 { + t.Error("handler should not be called for denied connection") + } + if adm.calls.Load() != 1 { + t.Errorf("admission calls = %d, want 1", adm.calls.Load()) + } +} + +func TestServeAdmissionAllow(t *testing.T) { + ln := newMockListener() + var handled atomic.Int32 + h := newMockHandler(func(ctx context.Context, conn net.Conn) error { + handled.Add(1) + return nil + }) + adm := &mockAdmission{allow: true} + svc := NewService("test", ln, h, AdmissionOption(adm)).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999}, + localAddr: ln.addr, + } + + time.Sleep(100 * time.Millisecond) + ln.Close() + <-done + + if handled.Load() != 1 { + t.Errorf("handled = %d, want 1", handled.Load()) + } +} + +func TestServeHandlerErrorIncrementsStats(t *testing.T) { + ln := newMockListener() + h := newMockHandler(func(ctx context.Context, conn net.Conn) error { + return errors.New("handler error") + }) + ms := newMockStats() + svc := NewService("test", ln, h, StatsOption(ms)).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999}, + localAddr: ln.addr, + } + + time.Sleep(100 * time.Millisecond) + ln.Close() + <-done + + if got := ms.Get(stats.KindTotalErrs); got != 1 { + t.Errorf("TotalErrs = %d, want 1", got) + } +} + +func TestServeContextFromConn(t *testing.T) { + ln := newMockListener() + var gotSid string + h := newMockHandler(func(ctx context.Context, conn net.Conn) error { + if s := xctx.SidFromContext(ctx); s != "" { + gotSid = string(s) + } + return nil + }) + svc := NewService("test", ln, h).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999}, + localAddr: ln.addr, + } + + time.Sleep(100 * time.Millisecond) + ln.Close() + <-done + + if gotSid == "" { + t.Error("expected session ID in context, got empty") + } +} + +func TestServeCtxConnContext(t *testing.T) { + ln := newMockListener() + wantCtx := context.Background() + var gotCtx context.Context + h := newMockHandler(func(ctx context.Context, conn net.Conn) error { + gotCtx = ctx + return nil + }) + svc := NewService("test", ln, h).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &ctxConn{ + Conn: &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 9999}, + localAddr: ln.addr, + }, + ctx: wantCtx, + } + + time.Sleep(100 * time.Millisecond) + ln.Close() + <-done + + if gotCtx == nil { + t.Fatal("handler context is nil") + } + // The ctx from the conn should be used as the base + if s := xctx.SidFromContext(gotCtx); s == "" { + t.Error("expected session ID to be added to conn context") + } +} + +type ctxConn struct { + net.Conn + ctx context.Context +} + +func (c *ctxConn) Context() context.Context { return c.ctx } + +func TestServeRecordsClientAddress(t *testing.T) { + ln := newMockListener() + rec := &mockRecorder{} + h := newMockHandler(nil) + + svc := NewService("test", ln, h, + RecordersOption(recorder.RecorderObject{ + Recorder: rec, + Record: recorder.RecorderServiceClientAddress, + }), + ).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("192.168.1.100"), Port: 54321}, + localAddr: ln.addr, + } + + time.Sleep(100 * time.Millisecond) + ln.Close() + <-done + + recorded := rec.getRecorded() + if len(recorded) != 1 { + t.Fatalf("recorded = %d entries, want 1", len(recorded)) + } + if string(recorded[0]) != "192.168.1.100" { + t.Errorf("recorded = %q, want %q", string(recorded[0]), "192.168.1.100") + } +} + +func TestServeRecordsClientAddressError(t *testing.T) { + ln := newMockListener() + rec := &mockRecorder{err: errors.New("record fail")} + h := newMockHandler(nil) + + svc := NewService("test", ln, h, + RecordersOption(recorder.RecorderObject{ + Recorder: rec, + Record: recorder.RecorderServiceClientAddress, + }), + ).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("192.168.1.100"), Port: 54321}, + localAddr: ln.addr, + } + + time.Sleep(100 * time.Millisecond) + ln.Close() + <-done + // Should not panic; error is just logged +} + +func TestServeObserverStats(t *testing.T) { + ln := newMockListener() + obs := &mockObserver{} + ms := newMockStats() + h := newMockHandler(nil) + + svc := NewService("test", ln, h, + ObserverOption(obs), + StatsOption(ms), + ObserverPeriodOption(50*time.Millisecond), + ).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + // Add some stats to trigger observation + ms.Add(stats.KindTotalConns, 10) + + time.Sleep(200 * time.Millisecond) + ln.Close() + <-done + + events := obs.getEvents() + if len(events) == 0 { + t.Error("expected observer to receive events") + } +} + +// --- Close tests --- + +func TestCloseClosesListener(t *testing.T) { + ln := newMockListener() + svc := NewService("test", ln, newMockHandler(nil)) + + if err := svc.Close(); err != nil { + t.Errorf("Close() = %v", err) + } + if !ln.closed.Load() { + t.Error("listener was not closed") + } +} + +func TestCloseClosesHandler(t *testing.T) { + ln := newMockListener() + h := newMockHandler(nil) + svc := NewService("test", ln, h).(*defaultService) + + if err := svc.Close(); err != nil { + t.Errorf("Close() = %v", err) + } + if !h.closed.Load() { + t.Error("handler was not closed") + } +} + +func TestCloseClosesObserver(t *testing.T) { + ln := newMockListener() + obs := &mockObserver{} + svc := NewService("test", ln, newMockHandler(nil), ObserverOption(obs)).(*defaultService) + + if err := svc.Close(); err != nil { + t.Errorf("Close() = %v", err) + } + if !obs.closed.Load() { + t.Error("observer was not closed") + } +} + +func TestCloseJoinsErrors(t *testing.T) { + ln := newMockListener() + h := &closeErrHandler{err: errors.New("handler close error")} + svc := NewService("test", ln, h).(*defaultService) + + err := svc.Close() + if err == nil { + t.Fatal("Close() = nil, want error") + } + if err.Error() != "handler close error" { + t.Errorf("Close() = %v, want 'handler close error'", err) + } +} + +type closeErrHandler struct { + err error +} + +func (h *closeErrHandler) Init(metadata.Metadata) error { return nil } +func (h *closeErrHandler) Handle(context.Context, net.Conn, ...handler.HandleOption) error { + return nil +} +func (h *closeErrHandler) Close() error { return h.err } + +// --- Status tests --- + +func TestStatusReturnsStatus(t *testing.T) { + ln := newMockListener() + svc := NewService("test", ln, newMockHandler(nil)).(*defaultService) + st := svc.Status() + if st == nil { + t.Fatal("Status() = nil") + } + if st.State() != StateRunning { + t.Errorf("State() = %q, want %q", st.State(), StateRunning) + } +} + +// --- ServiceEvent tests --- + +func TestServiceEventType(t *testing.T) { + ev := ServiceEvent{Kind: "service", Service: "test", State: StateReady} + if ev.Type() != observer.EventStatus { + t.Errorf("Type() = %v, want %v", ev.Type(), observer.EventStatus) + } +} + +// --- Integration: multiple connections --- + +func TestServeMultipleConnections(t *testing.T) { + ln := newMockListener() + var handled atomic.Int32 + h := newMockHandler(func(ctx context.Context, conn net.Conn) error { + handled.Add(1) + return nil + }) + svc := NewService("test", ln, h).(*defaultService) + + done := make(chan error, 1) + go func() { done <- svc.Serve() }() + + for i := 0; i < 5; i++ { + ln.connCh <- &mockConn{ + remoteAddr: &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 1000 + i}, + localAddr: ln.addr, + } + } + + time.Sleep(200 * time.Millisecond) + ln.Close() + <-done + + if got := handled.Load(); got != 5 { + t.Errorf("handled = %d, want 5", got) + } +} + +// --- execCmd tests --- + +func TestExecCmdInvalidCommand(t *testing.T) { + svc := NewService("test", newMockListener(), newMockHandler(nil)).(*defaultService) + err := svc.execCmd("") + if err == nil { + t.Error("execCmd('') = nil, want error") + } +} + +// --- Interface assertions --- + +var _ io.Closer = (*closeErrHandler)(nil) +var _ listener.Listener = (*mockListener)(nil) +var _ handler.Handler = (*mockHandler)(nil) +var _ handler.Handler = (*closeErrHandler)(nil) +var _ admission.Admission = (*mockAdmission)(nil) +var _ observer.Observer = (*mockObserver)(nil) +var _ stats.Stats = (*mockStats)(nil) +var _ recorder.Recorder = (*mockRecorder)(nil) +var _ xctx.Context = (*ctxConn)(nil) diff --git a/service/status.go b/service/status.go index eb504acc..ce8b933b 100644 --- a/service/status.go +++ b/service/status.go @@ -7,24 +7,31 @@ import ( "github.com/go-gost/core/observer/stats" ) -const ( - MaxEventSize = 20 -) +// MaxEventSize is the maximum number of events retained in a Status event log. +// When the log is full, the oldest event is discarded. +const MaxEventSize = 20 +// State represents the operational state of a service. type State string const ( + // StateRunning indicates the service has been created but is not yet accepting. StateRunning State = "running" - StateReady State = "ready" - StateFailed State = "failed" - StateClosed State = "closed" + // StateReady indicates the service is actively accepting connections. + StateReady State = "ready" + // StateFailed indicates the service accept loop encountered a temporary error. + StateFailed State = "failed" + // StateClosed indicates the service has been shut down. + StateClosed State = "closed" ) +// Event records a state change or notable occurrence in a service's lifetime. type Event struct { Time time.Time Message string } +// Status tracks the runtime state, event log, and traffic statistics of a service. type Status struct { createTime time.Time state State @@ -33,10 +40,12 @@ type Status struct { mu sync.RWMutex } +// CreateTime returns the time at which the service was created. func (p *Status) CreateTime() time.Time { return p.createTime } +// State returns the current service state. func (p *Status) State() State { p.mu.RLock() defer p.mu.RUnlock() @@ -49,12 +58,13 @@ func (p *Status) setState(state State) { p.state = state } +// Events returns a snapshot of the recent event log. The returned slice is safe +// to modify without affecting the internal state. func (p *Status) Events() []Event { - events := make([]Event, MaxEventSize) - p.mu.RLock() defer p.mu.RUnlock() + events := make([]Event, len(p.events)) copy(events, p.events) return events } @@ -71,6 +81,8 @@ func (p *Status) addEvent(event Event) { p.events = append(p.events, event) } +// Stats returns the traffic statistics for the service. It safely handles nil +// receivers by returning nil. func (p *Status) Stats() stats.Stats { if p == nil { return nil diff --git a/service/status_test.go b/service/status_test.go new file mode 100644 index 00000000..044246b9 --- /dev/null +++ b/service/status_test.go @@ -0,0 +1,218 @@ +package service + +import ( + "sync" + "testing" + "time" + + "github.com/go-gost/core/observer/stats" +) + +func TestStatusCreateAndState(t *testing.T) { + before := time.Now() + st := &Status{ + createTime: time.Now(), + events: make([]Event, 0, MaxEventSize), + } + after := time.Now() + + if st.CreateTime().Before(before) || st.CreateTime().After(after) { + t.Errorf("CreateTime = %v, want between %v and %v", st.CreateTime(), before, after) + } + + // Initial state is zero value + if st.State() != State("") { + t.Errorf("State() = %q, want empty", st.State()) + } + + st.setState(StateReady) + if got := st.State(); got != StateReady { + t.Errorf("State() = %q, want %q", got, StateReady) + } + + st.setState(StateFailed) + if got := st.State(); got != StateFailed { + t.Errorf("State() = %q, want %q", got, StateFailed) + } + + st.setState(StateClosed) + if got := st.State(); got != StateClosed { + t.Errorf("State() = %q, want %q", got, StateClosed) + } +} + +func TestStatusStateConcurrency(t *testing.T) { + st := &Status{ + events: make([]Event, 0, MaxEventSize), + } + + states := []State{StateRunning, StateReady, StateFailed, StateClosed} + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + st.setState(states[i%len(states)]) + _ = st.State() + }(i) + } + wg.Wait() +} + +func TestStatusEventsEmpty(t *testing.T) { + st := &Status{ + events: make([]Event, 0, MaxEventSize), + } + + events := st.Events() + if len(events) != 0 { + t.Errorf("Events() = %d events, want 0", len(events)) + } +} + +func TestStatusEventsAddAndRetrieve(t *testing.T) { + st := &Status{ + events: make([]Event, 0, MaxEventSize), + } + + want := []Event{ + {Time: time.Now(), Message: "event1"}, + {Time: time.Now(), Message: "event2"}, + {Time: time.Now(), Message: "event3"}, + } + for _, e := range want { + st.addEvent(e) + } + + events := st.Events() + if len(events) != len(want) { + t.Fatalf("Events() = %d events, want %d", len(events), len(want)) + } + for i, e := range events { + if e.Message != want[i].Message { + t.Errorf("Events()[%d].Message = %q, want %q", i, e.Message, want[i].Message) + } + } +} + +func TestStatusEventsReturnsCopy(t *testing.T) { + st := &Status{ + events: make([]Event, 0, MaxEventSize), + } + st.addEvent(Event{Message: "original"}) + + events := st.Events() + events[0] = Event{Message: "modified"} + + original := st.Events() + if original[0].Message != "original" { + t.Error("Events() should return a copy, internal state was mutated") + } +} + +func TestStatusEventsRingBuffer(t *testing.T) { + st := &Status{ + events: make([]Event, 0, MaxEventSize), + } + + // Add MaxEventSize + 5 events; first 5 should be evicted + for i := 0; i < MaxEventSize+5; i++ { + st.addEvent(Event{Message: time.Now().Format(time.RFC3339Nano)}) + time.Sleep(time.Microsecond) // ensure unique timestamps + } + + events := st.Events() + if len(events) != MaxEventSize { + t.Fatalf("len(Events()) = %d, want %d", len(events), MaxEventSize) + } +} + +func TestStatusEventsConcurrency(t *testing.T) { + st := &Status{ + events: make([]Event, 0, MaxEventSize), + } + + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(2) + go func() { + defer wg.Done() + st.addEvent(Event{Message: "writer"}) + }() + go func() { + defer wg.Done() + _ = st.Events() + }() + } + wg.Wait() +} + +type mockStats struct { + mu sync.Mutex + vals map[stats.Kind]uint64 + updated bool +} + +func newMockStats() *mockStats { + return &mockStats{vals: make(map[stats.Kind]uint64)} +} + +func (m *mockStats) Add(kind stats.Kind, n int64) { + m.mu.Lock() + defer m.mu.Unlock() + m.vals[kind] += uint64(n) + m.updated = true +} + +func (m *mockStats) Get(kind stats.Kind) uint64 { + m.mu.Lock() + defer m.mu.Unlock() + return m.vals[kind] +} + +func (m *mockStats) IsUpdated() bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.updated +} + +func (m *mockStats) Reset() { + m.mu.Lock() + defer m.mu.Unlock() + m.vals = make(map[stats.Kind]uint64) + m.updated = false +} + +func TestStatusStats(t *testing.T) { + ms := newMockStats() + st := &Status{ + events: make([]Event, 0, MaxEventSize), + stats: ms, + } + + if got := st.Stats(); got == nil { + t.Error("Stats() = nil, want non-nil") + } + + ms.Add(stats.KindTotalConns, 5) + if got := st.Stats().Get(stats.KindTotalConns); got != 5 { + t.Errorf("Stats().Get(TotalConns) = %d, want 5", got) + } +} + +func TestStatusStatsNilReceiver(t *testing.T) { + var st *Status + if got := st.Stats(); got != nil { + t.Errorf("nil Stats() = %v, want nil", got) + } +} + +func TestStatusStatsNil(t *testing.T) { + st := &Status{ + events: make([]Event, 0, MaxEventSize), + stats: nil, + } + if got := st.Stats(); got != nil { + t.Errorf("Stats() = %v, want nil", got) + } +}