From f0aca019f34a9500da12edc2da235b2ede2bba8cbecde913ce0b9a61ebf64fc8 Mon Sep 17 00:00:00 2001 From: Lixen Wraith Date: Fri, 17 Jul 2026 18:29:26 -0400 Subject: [PATCH] v0.15.0 deprecated http/tcp plugins (fasthttp/gnet2), migrated stdtcp/stdhttp to tcp/http --- cmd/logwisp/bootstrap.go | 2 - cmd/logwisp/main.go | 1 + doc/sinks.md | 22 +- go.mod | 12 - go.sum | 30 -- internal/config/config.go | 26 +- internal/sink/http/http.go | 569 +++++++++++++--------------- internal/sink/stdhttp/stdhttp.go | 484 ------------------------ internal/sink/stdtcp/stdtcp.go | 399 ------------------- internal/sink/tcp/tcp.go | 631 +++++++++++++------------------ test/chain-aggregate-test.sh | 16 +- test/chain-test.sh | 5 +- 12 files changed, 551 insertions(+), 1646 deletions(-) delete mode 100644 internal/sink/stdhttp/stdhttp.go delete mode 100644 internal/sink/stdtcp/stdtcp.go diff --git a/cmd/logwisp/bootstrap.go b/cmd/logwisp/bootstrap.go index 916cd9b..ac861bf 100644 --- a/cmd/logwisp/bootstrap.go +++ b/cmd/logwisp/bootstrap.go @@ -16,8 +16,6 @@ import ( _ "logwisp/internal/sink/http" _ "logwisp/internal/sink/httpchain" _ "logwisp/internal/sink/null" - _ "logwisp/internal/sink/stdhttp" - _ "logwisp/internal/sink/stdtcp" _ "logwisp/internal/sink/tcp" _ "logwisp/internal/sink/tcpchain" diff --git a/cmd/logwisp/main.go b/cmd/logwisp/main.go index 56ed3b5..8c80a58 100644 --- a/cmd/logwisp/main.go +++ b/cmd/logwisp/main.go @@ -75,6 +75,7 @@ func main() { svc, statusReporterCancel, err := bootstrapInitial(ctx, cfg) if err != nil { logger.Error("msg", "Failed to initialize service", "error", err) + shutdownLogger() os.Exit(1) } diff --git a/doc/sinks.md b/doc/sinks.md index c792bd6..df7e3b8 100644 --- a/doc/sinks.md +++ b/doc/sinks.md @@ -81,7 +81,9 @@ port = 8080 stream_path = "/stream" status_path = "/status" buffer_size = 1000 -write_timeout_ms = 10000 +client_buffer_size = 256 +write_timeout_ms = 0 +max_connections = 0 ``` **Configuration Options:** @@ -92,12 +94,14 @@ write_timeout_ms = 10000 | `port` | int | Required | Listen port | | `stream_path` | string | "/stream" | SSE stream endpoint | | `status_path` | string | "/status" | Status endpoint | -| `buffer_size` | int | 1000 | Internal buffer size | -| `write_timeout_ms` | int | 10000 | Write timeout | +| `buffer_size` | int | 1000 | Sink input queue size | +| `client_buffer_size` | int | 256 | Per-client send queue size | +| `write_timeout_ms` | int | 0 | Write deadline per event (0 = none) | +| `max_connections` | int | 0 | Concurrent connection cap (0 = unlimited) | ### TCP Sink -TCP streaming server for debugging. +TCP streaming server for debugging and raw client forwarding. ```toml [[pipelines.plugin_sinks]] @@ -107,8 +111,11 @@ type = "tcp" host = "0.0.0.0" port = 9090 buffer_size = 1000 +client_buffer_size = 256 +write_timeout_ms = 5000 keep_alive = true keep_alive_period_ms = 30000 +max_connections = 0 ``` **Configuration Options:** @@ -117,10 +124,13 @@ keep_alive_period_ms = 30000 |--------|------|---------|-------------| | `host` | string | "0.0.0.0" | Bind address | | `port` | int | Required | Listen port | -| `buffer_size` | int | 1000 | Internal buffer size | +| `buffer_size` | int | 1000 | Sink input queue size | +| `client_buffer_size` | int | 256 | Per-client send queue size | +| `write_timeout_ms` | int | 5000 | Write timeout | | `keep_alive` | bool | true | Enable TCP keep-alive | | `keep_alive_period_ms` | int | 30000 | Keep-alive interval | -| `write_timeout_ms` | int | 10000 | Write timeout | +| `max_connections` | int | 0 | Concurrent connection cap (0 = unlimited) | +``` ### Null Sink diff --git a/go.mod b/go.mod index 2e7ae89..456e0c4 100644 --- a/go.mod +++ b/go.mod @@ -5,22 +5,10 @@ go 1.26.0 require ( github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98 github.com/lixenwraith/log v0.1.1-0.20260717175128-82eea9846ccd - github.com/panjf2000/gnet/v2 v2.10.0 - github.com/valyala/fasthttp v1.72.0 ) require ( github.com/BurntSushi/toml v1.6.0 // indirect - github.com/andybalholm/brotli v1.2.2 // indirect - github.com/davecgh/go-spew v1.1.1 // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect - github.com/klauspost/compress v1.19.0 // indirect - github.com/panjf2000/ants/v2 v2.12.1 // indirect - github.com/valyala/bytebufferpool v1.0.0 // indirect - go.uber.org/multierr v1.11.0 // indirect - go.uber.org/zap v1.28.0 // indirect - golang.org/x/sync v0.22.0 // indirect - golang.org/x/sys v0.47.0 // indirect - gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 2e0fc53..203dd83 100644 --- a/go.sum +++ b/go.sum @@ -1,48 +1,18 @@ github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= -github.com/andybalholm/brotli v1.2.2 h1:HzTuoo2ErYQqf5qvcJInB8uvqSVxRttzkFexPWtnceM= -github.com/andybalholm/brotli v1.2.2/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= -github.com/klauspost/compress v1.19.0 h1:sXLILfc9jV2QYWkzFOPWStmcUVH2RHEB1JCdY2oVvCQ= -github.com/klauspost/compress v1.19.0/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98 h1:VEFo2WsYgM5YYfzAfvXRKigCWr3zEz6h4M59N5TXMpk= github.com/lixenwraith/config v0.1.1-0.20260712172228-ccd280ba6a98/go.mod h1:J9ydxY7he4Dz+S59xKs/kDQ2YAv+zOLF05b3Rr2/ogE= -github.com/lixenwraith/log v0.1.1-0.20251117213308-9ae1b6669bf3 h1:brSUhER7EZ28aMRFTSovZskiIoobUiAWlx+DYXYvWXQ= -github.com/lixenwraith/log v0.1.1-0.20251117213308-9ae1b6669bf3/go.mod h1:MY59N65ltw/9uTqJKwCRKjxO6w3CApXsLXCVuaekbu0= github.com/lixenwraith/log v0.1.1-0.20260717175128-82eea9846ccd h1:06Rk4DLvJW1kdeclH5HwywizgtMI64PHNE/2sBeuiwA= github.com/lixenwraith/log v0.1.1-0.20260717175128-82eea9846ccd/go.mod h1:2+qURSVdWcX7REFH+3jUtIbtc+NtsAPecSvXLeGyK6U= -github.com/panjf2000/ants/v2 v2.12.1 h1:BWvU2wHpyXWxhhNXsGB6JXLCNbshyLd1QxvoAmZnu10= -github.com/panjf2000/ants/v2 v2.12.1/go.mod h1:tSQuaNQ6r6NRhPt+IZVUevvDyFMTs+eS4ztZc52uJTY= -github.com/panjf2000/gnet/v2 v2.10.0 h1:rC4jNF+jtXj/FH+8JOIQ3XxjD+yBunYBLKg9TE3dc4g= -github.com/panjf2000/gnet/v2 v2.10.0/go.mod h1:f9wdbOFsdbZqlSvXctWbPRW5bB/W++q8Zqz+D7tQIVQ= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= -github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= -github.com/valyala/fasthttp v1.72.0 h1:R7kYdoWhn1ye1fVpP+cDHDJwYm3NkwLliwgzJ/Abg7M= -github.com/valyala/fasthttp v1.72.0/go.mod h1:zsbLTYqcpIktdQytlVBwIjY9La5d6bs990nBxWg8efk= -github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= -github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= -go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= -go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= -go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= -go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= -go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo= -go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q= -go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= -go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= -gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/config/config.go b/internal/config/config.go index add25de..617d442 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -273,27 +273,6 @@ type FileSinkOptions struct { // TCPSinkOptions defines settings for a TCP server sink type TCPSinkOptions struct { - Host string `toml:"host"` - Port int64 `toml:"port"` - BufferSize int64 `toml:"buffer_size"` - WriteTimeout int64 `toml:"write_timeout_ms"` - KeepAlivePeriod int64 `toml:"keep_alive_period_ms"` - KeepAlive bool `toml:"keep_alive"` -} - -// HTTPSinkOptions defines settings for an HTTP SSE server sink -type HTTPSinkOptions struct { - StreamPath string `toml:"stream_path"` - StatusPath string `toml:"status_path"` - Host string `toml:"host"` - Port int64 `toml:"port"` - BufferSize int64 `toml:"buffer_size"` - WriteTimeout int64 `toml:"write_timeout_ms"` -} - -// StdTCPSinkOptions defines settings for a stdlib TCP streaming server sink. -// TOML keys mirror TCPSinkOptions for drop-in type swap ("tcp" -> "stdtcp"). -type StdTCPSinkOptions struct { Host string `toml:"host"` Port int64 `toml:"port"` BufferSize int64 `toml:"buffer_size"` // sink input queue @@ -305,9 +284,8 @@ type StdTCPSinkOptions struct { // Future: TLS (cert_file/key_file/client_ca), auth (token/mTLS) blocks } -// StdHTTPSinkOptions defines settings for a stdlib HTTP SSE streaming sink. -// TOML keys mirror HTTPSinkOptions for drop-in type swap ("http" -> "stdhttp"). -type StdHTTPSinkOptions struct { +// HTTPSinkOptions defines settings for an HTTP SSE server sink +type HTTPSinkOptions struct { Host string `toml:"host"` Port int64 `toml:"port"` StreamPath string `toml:"stream_path"` diff --git a/internal/sink/http/http.go b/internal/sink/http/http.go index 2e12762..6b5b0d3 100644 --- a/internal/sink/http/http.go +++ b/internal/sink/http/http.go @@ -1,11 +1,14 @@ package http import ( - "bufio" "context" "encoding/json" + "errors" "fmt" "net" + "net/http" + "strconv" + "strings" "sync" "sync/atomic" "time" @@ -19,8 +22,6 @@ import ( lconfig "github.com/lixenwraith/config" "github.com/lixenwraith/log" - "github.com/lixenwraith/log/compat" - "github.com/valyala/fasthttp" ) func init() { @@ -29,7 +30,19 @@ func init() { } } -// HTTPSink streams log entries via Server-Sent Events (SSE) +const ( + DefaultHTTPHost = "0.0.0.0" + DefaultHTTPBufferSize = 1000 + DefaultHTTPClientBufferSize = 256 + DefaultHTTPStreamPath = "/stream" + DefaultHTTPStatusPath = "/status" + HTTPReadHeaderTimeout = 10 * time.Second + HTTPShutdownTimeout = 2 * time.Second +) + +// HTTPSink streams log entries via Server-Sent Events +// Server.WriteTimeout is deliberately unset (it would terminate long-lived SSE streams) +// per-write deadlines are applied via http.ResponseController type HTTPSink struct { // Plugin identity and session management id string @@ -37,49 +50,43 @@ type HTTPSink struct { // Configuration config *config.HTTPSinkOptions + addr string // Network - server *fasthttp.Server + server *http.Server // Application input chan core.TransportEvent logger *log.Logger + // Client registry + clients map[uint64]*sseClient + clientsMu sync.Mutex + nextClientID atomic.Uint64 + // Runtime done chan struct{} + stopOnce sync.Once wg sync.WaitGroup startTime time.Time - // Broker - clients map[uint64]chan []byte - clientsMu sync.RWMutex - unregister chan uint64 - nextClientID atomic.Uint64 - - // Client session tracking - clientSessions map[uint64]string // clientID -> sessionID - sessionsMu sync.RWMutex + writeTimeout time.Duration // Statistics - activeClients atomic.Int64 - totalProcessed atomic.Uint64 - lastProcessed atomic.Value // time.Time + activeClients atomic.Int64 + totalProcessed atomic.Uint64 + droppedWrites atomic.Uint64 + rejectedClients atomic.Uint64 + lastProcessed atomic.Value // time.Time } -const ( - // Server lifecycle - HttpServerStartTimeout = 100 * time.Millisecond - HttpServerShutdownTimeout = 2 * time.Second +// sseClient is a registered stream consumer with a bounded send queue +type sseClient struct { + send chan []byte + sessionID string +} - // Defaults - DefaultHTTPHost = "0.0.0.0" - DefaultHTTPBufferSize = 1000 - DefaultHTTPStreamPath = "/stream" - DefaultHTTPStatusPath = "/status" - HTTPMaxPort = 65535 -) - -// NewHTTPSinkPlugin creates an HTTP sink through plugin factory +// NewHTTPSinkPlugin creates a http sink through plugin factory func NewHTTPSinkPlugin( id string, configMap map[string]any, @@ -87,57 +94,61 @@ func NewHTTPSinkPlugin( proxy *session.Proxy, ) (sink.Sink, error) { opts := &config.HTTPSinkOptions{ - Host: DefaultHTTPHost, - Port: 0, - WriteTimeout: 0, // SSE indefinite streaming + Host: DefaultHTTPHost, + WriteTimeoutMS: 0, // SSE indefinite streaming } - if err := lconfig.ScanMap(configMap, opts); err != nil { return nil, fmt.Errorf("failed to parse config: %w", err) } - - // Validate - if opts.Port <= 0 || opts.Port > HTTPMaxPort { - return nil, fmt.Errorf("port must be between 1 and %d", HTTPMaxPort) - } - - // Defaults - if opts.BufferSize <= 0 { - opts.BufferSize = DefaultHTTPBufferSize + if err := lconfig.Port(opts.Port); err != nil { + return nil, fmt.Errorf("port: %w", err) } if opts.StreamPath == "" { opts.StreamPath = DefaultHTTPStreamPath + } else if !strings.HasPrefix(opts.StreamPath, "/") { + return nil, fmt.Errorf("stream_path: must start with '/'") } if opts.StatusPath == "" { opts.StatusPath = DefaultHTTPStatusPath + } else if !strings.HasPrefix(opts.StatusPath, "/") { + return nil, fmt.Errorf("status_path: must start with '/'") + } + if opts.StreamPath == opts.StatusPath { + return nil, fmt.Errorf("stream_path and status_path must differ") + } + if opts.BufferSize <= 0 { + opts.BufferSize = DefaultHTTPBufferSize + } + if opts.ClientBufferSize <= 0 { + opts.ClientBufferSize = DefaultHTTPClientBufferSize } h := &HTTPSink{ - id: id, - proxy: proxy, - config: opts, - input: make(chan core.TransportEvent, opts.BufferSize), - done: make(chan struct{}), - logger: logger, - clients: make(map[uint64]chan []byte), - unregister: make(chan uint64), - clientSessions: make(map[uint64]string), + id: id, + proxy: proxy, + config: opts, + addr: net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)), + input: make(chan core.TransportEvent, opts.BufferSize), + done: make(chan struct{}), + logger: logger, + clients: make(map[uint64]*sseClient), + writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond, } h.lastProcessed.Store(time.Time{}) - logger.Info("msg", "HTTP sink initialized", + logger.Info("msg", " HTTP sink initialized", "component", "http_sink", "instance_id", id, "host", opts.Host, "port", opts.Port, "stream_path", opts.StreamPath, "status_path", opts.StatusPath) - return h, nil } // Capabilities returns supported capabilities func (h *HTTPSink) Capabilities() []core.Capability { + // CapTLS/CapAuth appended when transport security lands return []core.Capability{ core.CapSessionAware, core.CapMultiSession, @@ -149,33 +160,38 @@ func (h *HTTPSink) Input() chan<- core.TransportEvent { return h.input } -// Start initializes the HTTP server and begins the broker loop +// Start binds the listener and serves stream/status endpoints func (h *HTTPSink) Start(ctx context.Context) error { + // IPv4-only, parity with existing network sinks. + // TLS extension point: wrap ln with tls.NewListener (or set + // server.TLSConfig and use ServeTLS); single seam, handlers unchanged. + ln, err := net.Listen("tcp4", h.addr) + if err != nil { + return fmt.Errorf("http sink bind %s: %w", h.addr, err) + } + + mux := http.NewServeMux() + // Method-scoped patterns: mux answers 405 with Allow header on non-GET + mux.HandleFunc(http.MethodGet+" "+h.config.StreamPath, h.handleStream) + mux.HandleFunc(http.MethodGet+" "+h.config.StatusPath, h.handleStatus) + + // Auth extension point: wrap mux with auth middleware once credentials + // land, e.g. handler = authMiddleware(cfg)(handler) + var handler http.Handler = mux + + h.server = &http.Server{ + Handler: handler, + ReadHeaderTimeout: HTTPReadHeaderTimeout, + // WriteTimeout unset by design: SSE responses are long-lived. + // Per-write deadlines via ResponseController in handleStream. + } h.startTime = time.Now() - // Start central broker goroutine h.wg.Add(1) go h.brokerLoop(ctx) - fasthttpLogger := compat.NewFastHTTPAdapter(h.logger) - - h.server = &fasthttp.Server{ - Name: fmt.Sprintf("LogWisp/%s", version.Short()), - Handler: h.requestHandler, - DisableKeepalive: false, - StreamRequestBody: true, - Logger: fasthttpLogger, - WriteTimeout: time.Duration(h.config.WriteTimeout) * time.Millisecond, - } - - addr := fmt.Sprintf("%s:%d", h.config.Host, h.config.Port) - - ln, err := net.Listen("tcp4", addr) - if err != nil { - return fmt.Errorf("http sink bind %s: %w", addr, err) - } go func() { - if err := h.server.Serve(ln); err != nil { + if err := h.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { h.logger.Error("msg", "HTTP server terminated", "component", "http_sink", "instance_id", h.id, @@ -183,297 +199,201 @@ func (h *HTTPSink) Start(ctx context.Context) error { } }() - // Monitor context for shutdown go func() { - <-ctx.Done() - if h.server != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), HttpServerShutdownTimeout) - defer cancel() - h.server.ShutdownWithContext(shutdownCtx) + select { + case <-ctx.Done(): + h.shutdown() + case <-h.done: } }() - h.logger.Info("msg", "HTTP server started", + h.logger.Info("msg", " HTTP server started", "component", "http_sink", "instance_id", h.id, - "host", h.config.Host, - "port", h.config.Port) + "addr", h.addr) return nil } -// Stop gracefully shuts down the HTTP server and all client connections +// Stop gracefully shuts down the sink func (h *HTTPSink) Stop() { h.logger.Info("msg", "Stopping HTTP sink", "component", "http_sink", "instance_id", h.id) - close(h.done) - - if h.server != nil { - ctx, cancel := context.WithTimeout(context.Background(), HttpServerShutdownTimeout) - defer cancel() - h.server.ShutdownWithContext(ctx) - } - + h.shutdown() h.wg.Wait() - close(h.unregister) - - h.clientsMu.Lock() - for _, ch := range h.clients { - close(ch) - } - h.clients = make(map[uint64]chan []byte) - h.clientsMu.Unlock() - - h.logger.Info("msg", "HTTP sink stopped", + h.logger.Info("msg", " HTTP sink stopped", "component", "http_sink", "instance_id", h.id, "total_processed", h.totalProcessed.Load()) } -// GetStats returns sink statistics -func (h *HTTPSink) GetStats() sink.SinkStats { - lastProc, _ := h.lastProcessed.Load().(time.Time) +// shutdown funnels ctx-cancel and Stop() teardown through a single path. +// done is closed first so SSE handlers exit and Shutdown can complete; +// Server.Close force-closes any handler stalled in a deadline-free write. +func (h *HTTPSink) shutdown() { + h.stopOnce.Do(func() { + close(h.done) + if h.server != nil { + sctx, cancel := context.WithTimeout(context.Background(), HTTPShutdownTimeout) + defer cancel() + if err := h.server.Shutdown(sctx); err != nil { + h.server.Close() + } + } + }) +} - return sink.SinkStats{ - ID: h.id, - Type: "http", - TotalProcessed: h.totalProcessed.Load(), - ActiveConnections: h.activeClients.Load(), - StartTime: h.startTime, - LastProcessed: lastProc, - Details: map[string]any{ - "host": h.config.Host, - "port": h.config.Port, - "buffer_size": h.config.BufferSize, - "endpoints": map[string]string{ - "stream": h.config.StreamPath, - "status": h.config.StatusPath, - }, - }, +// removeClient unregisters a client; the first caller closes the send +// channel and removes the session. Broker (stale-session eviction) and +// stream handler (disconnect) may race here safely. +func (h *HTTPSink) removeClient(id uint64) { + h.clientsMu.Lock() + c, ok := h.clients[id] + if ok { + delete(h.clients, id) + } + h.clientsMu.Unlock() + if ok { + close(c.send) + h.proxy.RemoveSession(c.sessionID) } } -// brokerLoop manages client connections and broadcasts transport events +// brokerLoop fans out transport events to all client queues, non-blocking, +// and evicts clients whose sessions were idle-expired by the session manager func (h *HTTPSink) brokerLoop(ctx context.Context) { defer h.wg.Done() - for { select { case <-ctx.Done(): - h.logger.Debug("msg", "Broker loop stopping due to context cancellation", - "component", "http_sink") return - case <-h.done: - h.logger.Debug("msg", "Broker loop stopping due to shutdown signal", - "component", "http_sink") return - - case clientID := <-h.unregister: - h.clientsMu.Lock() - if clientChan, exists := h.clients[clientID]; exists { - delete(h.clients, clientID) - close(clientChan) - h.logger.Debug("msg", "Unregistered client", - "component", "http_sink", - "client_id", clientID) - } - h.clientsMu.Unlock() - - h.sessionsMu.Lock() - delete(h.clientSessions, clientID) - h.sessionsMu.Unlock() - case event, ok := <-h.input: if !ok { - h.logger.Debug("msg", "Input channel closed, broker stopping", - "component", "http_sink") return } - h.totalProcessed.Add(1) h.lastProcessed.Store(time.Now()) - h.clientsMu.RLock() - clientCount := len(h.clients) - if clientCount > 0 { - var staleClients []uint64 - - for id, ch := range h.clients { - h.sessionsMu.RLock() - sessionID, hasSession := h.clientSessions[id] - h.sessionsMu.RUnlock() - - if !hasSession { - staleClients = append(staleClients, id) - continue - } - - // Check session still exists via proxy - if _, exists := h.proxy.GetSession(sessionID); !exists { - staleClients = append(staleClients, id) - continue - } - - select { - case ch <- event.Payload: - h.proxy.UpdateActivity(sessionID) - default: - h.logger.Debug("msg", "Dropped event for slow client", - "component", "http_sink", - "client_id", id) - } + var stale []uint64 + h.clientsMu.Lock() + for id, c := range h.clients { + if _, exists := h.proxy.GetSession(c.sessionID); !exists { + stale = append(stale, id) + continue } - - if len(staleClients) > 0 { - go func() { - for _, clientID := range staleClients { - select { - case h.unregister <- clientID: - case <-h.done: - return - } - } - }() + select { + case c.send <- event.Payload: + h.proxy.UpdateActivity(c.sessionID) + default: + h.droppedWrites.Add(1) } } - h.clientsMu.RUnlock() + h.clientsMu.Unlock() + + for _, id := range stale { + h.removeClient(id) + } } } } -// requestHandler is the main entry point for all incoming HTTP requests -func (h *HTTPSink) requestHandler(ctx *fasthttp.RequestCtx) { - // IPv4-only enforcement - silent drop IPv6 - remoteAddr := ctx.RemoteAddr() - if tcpAddr, ok := remoteAddr.(*net.TCPAddr); ok { - if tcpAddr.IP.To4() == nil { - h.logger.Debug("msg", "IPv6 connection rejected", - "component", "http_sink", "remote_addr", remoteAddr.String()) - ctx.SetConnectionClose() - return - } +// handleStream serves one client's SSE stream +func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) { + if h.config.MaxConnections > 0 && h.activeClients.Load() >= h.config.MaxConnections { + h.rejectedClients.Add(1) + http.Error(w, "too many clients", http.StatusServiceUnavailable) + return } - path := string(ctx.Path()) + rc := http.NewResponseController(w) + remote := r.RemoteAddr - switch path { - case h.config.StatusPath: - h.handleStatus(ctx) - case h.config.StreamPath: - h.handleStream(ctx) - default: - ctx.SetStatusCode(fasthttp.StatusNotFound) - ctx.SetContentType("application/json") - json.NewEncoder(ctx).Encode(map[string]any{ - "error": "Not Found", - }) - } -} - -// handleStream manages a client's Server-Sent Events (SSE) stream -func (h *HTTPSink) handleStream(ctx *fasthttp.RequestCtx) { - remoteAddrStr := ctx.RemoteAddr().String() - - // Create session via proxy - sess := h.proxy.CreateSession(remoteAddrStr, map[string]any{ + sess := h.proxy.CreateSession(remote, map[string]any{ "type": "http_client", }) - // Set SSE headers - ctx.Response.Header.Set("Content-Type", "text/event-stream") - ctx.Response.Header.Set("Cache-Control", "no-cache") - ctx.Response.Header.Set("Connection", "keep-alive") - ctx.Response.Header.Set("Access-Control-Allow-Origin", "*") - ctx.Response.Header.Set("X-Accel-Buffering", "no") - - // Register client with broker - clientID := h.nextClientID.Add(1) - clientChan := make(chan []byte, h.config.BufferSize) + c := &sseClient{ + send: make(chan []byte, h.config.ClientBufferSize), + sessionID: sess.ID, + } + id := h.nextClientID.Add(1) h.clientsMu.Lock() - h.clients[clientID] = clientChan + h.clients[id] = c h.clientsMu.Unlock() - h.sessionsMu.Lock() - h.clientSessions[clientID] = sess.ID - h.sessionsMu.Unlock() + count := h.activeClients.Add(1) + h.logger.Debug("msg", "HTTP client connected", + "component", "http_sink", + "remote_addr", remote, + "session_id", sess.ID, + "client_id", id, + "active_clients", count) - streamFunc := func(w *bufio.Writer) { - connectCount := h.activeClients.Add(1) - h.logger.Debug("msg", "HTTP client connected", + defer func() { + h.removeClient(id) + newCount := h.activeClients.Add(-1) + h.logger.Debug("msg", "HTTP client disconnected", "component", "http_sink", - "remote_addr", remoteAddrStr, + "remote_addr", remote, "session_id", sess.ID, - "client_id", clientID, - "active_clients", connectCount) + "client_id", id, + "active_clients", newCount) + }() - defer func() { - disconnectCount := h.activeClients.Add(-1) - h.logger.Debug("msg", "HTTP client disconnected", - "component", "http_sink", - "remote_addr", remoteAddrStr, - "session_id", sess.ID, - "client_id", clientID, - "active_clients", disconnectCount) + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("X-Accel-Buffering", "no") + w.WriteHeader(http.StatusOK) - select { - case h.unregister <- clientID: - case <-h.done: - } - - h.proxy.RemoveSession(sess.ID) - }() - - // Send connected event with metadata - connectionInfo := map[string]any{ - "client_id": fmt.Sprintf("%d", clientID), - "session_id": sess.ID, - "instance_id": h.id, - "stream_path": h.config.StreamPath, - "status_path": h.config.StatusPath, - "buffer_size": h.config.BufferSize, - } - data, _ := json.Marshal(connectionInfo) - fmt.Fprintf(w, "event: connected\ndata: %s\n\n", data) - if err := w.Flush(); err != nil { - return - } - - for { - select { - case payload, ok := <-clientChan: - if !ok { - return - } - - if err := h.writeSSE(w, payload); err != nil { - return - } - - if err := w.Flush(); err != nil { - return - } - - h.proxy.UpdateActivity(sess.ID) - - case <-h.done: - fmt.Fprintf(w, "event: disconnect\ndata: {\"reason\":\"server_shutdown\"}\n\n") - w.Flush() - return - } - } + // Connected event with metadata, parity with fasthttp sink + info, _ := json.Marshal(map[string]any{ + "client_id": strconv.FormatUint(id, 10), + "session_id": sess.ID, + "instance_id": h.id, + "stream_path": h.config.StreamPath, + "status_path": h.config.StatusPath, + "buffer_size": h.config.ClientBufferSize, + }) + fmt.Fprintf(w, "event: connected\ndata: %s\n\n", info) + if err := rc.Flush(); err != nil { + return } - ctx.SetBodyStreamWriter(streamFunc) + clientGone := r.Context().Done() + for { + select { + case payload, ok := <-c.send: + if !ok { + return // broker evicted (stale session) + } + if h.writeTimeout > 0 { + _ = rc.SetWriteDeadline(time.Now().Add(h.writeTimeout)) + } + if err := writeSSE(w, payload); err != nil { + return + } + if err := rc.Flush(); err != nil { + return + } + h.proxy.UpdateActivity(sess.ID) + case <-clientGone: + return + case <-h.done: + fmt.Fprintf(w, "event: disconnect\ndata: {\"reason\":\"server_shutdown\"}\n\n") + rc.Flush() + return + } + } } // handleStatus provides a JSON status report -func (h *HTTPSink) handleStatus(ctx *fasthttp.RequestCtx) { - ctx.SetContentType("application/json") - +func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) { status := map[string]any{ "service": "LogWisp", "version": version.Short(), @@ -491,41 +411,59 @@ func (h *HTTPSink) handleStatus(ctx *fasthttp.RequestCtx) { "status": h.config.StatusPath, }, "statistics": map[string]any{ - "total_processed": h.totalProcessed.Load(), + "total_processed": h.totalProcessed.Load(), + "dropped_writes": h.droppedWrites.Load(), + "rejected_clients": h.rejectedClients.Load(), }, } - data, _ := json.Marshal(status) - ctx.SetBody(data) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(status) } -// writeSSE formats payload into SSE data format -func (h *HTTPSink) writeSSE(w *bufio.Writer, payload []byte) error { - // Handle multi-line payloads per W3C SSE spec - lines := splitLines(payload) - for _, line := range lines { +// GetStats returns sink statistics +func (h *HTTPSink) GetStats() sink.SinkStats { + lastProc, _ := h.lastProcessed.Load().(time.Time) + return sink.SinkStats{ + ID: h.id, + Type: "http", + TotalProcessed: h.totalProcessed.Load(), + ActiveConnections: h.activeClients.Load(), + StartTime: h.startTime, + LastProcessed: lastProc, + Details: map[string]any{ + "host": h.config.Host, + "port": h.config.Port, + "buffer_size": h.config.BufferSize, + "dropped_writes": h.droppedWrites.Load(), + "rejected_clients": h.rejectedClients.Load(), + "endpoints": map[string]string{ + "stream": h.config.StreamPath, + "status": h.config.StatusPath, + }, + }, + } +} + +// writeSSE frames a payload per the W3C SSE spec (multi-line safe) +func writeSSE(w http.ResponseWriter, payload []byte) error { + for _, line := range splitLines(payload) { if _, err := fmt.Fprintf(w, "data: %s\n", line); err != nil { return err } } - // Empty line terminates event - if _, err := w.WriteString("\n"); err != nil { - return err - } - return nil + _, err := fmt.Fprint(w, "\n") + return err } -// splitLines splits payload by newlines, handling different line endings +// splitLines splits payload by newlines, trimming a single trailing newline func splitLines(data []byte) [][]byte { if len(data) == 0 { return nil } - - // Trim trailing newline if present if data[len(data)-1] == '\n' { data = data[:len(data)-1] } - var lines [][]byte start := 0 for i := 0; i < len(data); i++ { @@ -537,7 +475,6 @@ func splitLines(data []byte) [][]byte { if start < len(data) { lines = append(lines, data[start:]) } - if len(lines) == 0 { return [][]byte{data} } diff --git a/internal/sink/stdhttp/stdhttp.go b/internal/sink/stdhttp/stdhttp.go deleted file mode 100644 index 759f91b..0000000 --- a/internal/sink/stdhttp/stdhttp.go +++ /dev/null @@ -1,484 +0,0 @@ -package stdhttp - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "net" - "net/http" - "strconv" - "strings" - "sync" - "sync/atomic" - "time" - - "logwisp/internal/config" - "logwisp/internal/core" - "logwisp/internal/plugin" - "logwisp/internal/session" - "logwisp/internal/sink" - "logwisp/internal/version" - - lconfig "github.com/lixenwraith/config" - "github.com/lixenwraith/log" -) - -func init() { - if err := plugin.RegisterSink("stdhttp", NewStdHTTPSinkPlugin); err != nil { - panic(fmt.Sprintf("failed to register stdhttp sink: %v", err)) - } -} - -const ( - DefaultStdHTTPHost = "0.0.0.0" - DefaultStdHTTPBufferSize = 1000 - DefaultStdHTTPClientBufferSize = 256 - DefaultStdHTTPStreamPath = "/stream" - DefaultStdHTTPStatusPath = "/status" - StdHTTPReadHeaderTimeout = 10 * time.Second - StdHTTPShutdownTimeout = 2 * time.Second -) - -// StdHTTPSink streams log entries via Server-Sent Events using only the -// standard library. Functional peer of the fasthttp-based http sink. -// -// Server.WriteTimeout is deliberately unset (it would terminate long-lived -// SSE streams); per-write deadlines are applied via http.ResponseController. -type StdHTTPSink struct { - // Plugin identity and session management - id string - proxy *session.Proxy - - // Configuration - config *config.StdHTTPSinkOptions - addr string - - // Network - server *http.Server - - // Application - input chan core.TransportEvent - logger *log.Logger - - // Client registry - clients map[uint64]*sseClient - clientsMu sync.Mutex - nextClientID atomic.Uint64 - - // Runtime - done chan struct{} - stopOnce sync.Once - wg sync.WaitGroup - startTime time.Time - - writeTimeout time.Duration - - // Statistics - activeClients atomic.Int64 - totalProcessed atomic.Uint64 - droppedWrites atomic.Uint64 - rejectedClients atomic.Uint64 - lastProcessed atomic.Value // time.Time -} - -// sseClient is a registered stream consumer with a bounded send queue -type sseClient struct { - send chan []byte - sessionID string -} - -// NewStdHTTPSinkPlugin creates a stdhttp sink through plugin factory -func NewStdHTTPSinkPlugin( - id string, - configMap map[string]any, - logger *log.Logger, - proxy *session.Proxy, -) (sink.Sink, error) { - opts := &config.StdHTTPSinkOptions{ - Host: DefaultStdHTTPHost, - WriteTimeoutMS: 0, // SSE indefinite streaming - } - if err := lconfig.ScanMap(configMap, opts); err != nil { - return nil, fmt.Errorf("failed to parse config: %w", err) - } - if err := lconfig.Port(opts.Port); err != nil { - return nil, fmt.Errorf("port: %w", err) - } - if opts.StreamPath == "" { - opts.StreamPath = DefaultStdHTTPStreamPath - } else if !strings.HasPrefix(opts.StreamPath, "/") { - return nil, fmt.Errorf("stream_path: must start with '/'") - } - if opts.StatusPath == "" { - opts.StatusPath = DefaultStdHTTPStatusPath - } else if !strings.HasPrefix(opts.StatusPath, "/") { - return nil, fmt.Errorf("status_path: must start with '/'") - } - if opts.StreamPath == opts.StatusPath { - return nil, fmt.Errorf("stream_path and status_path must differ") - } - if opts.BufferSize <= 0 { - opts.BufferSize = DefaultStdHTTPBufferSize - } - if opts.ClientBufferSize <= 0 { - opts.ClientBufferSize = DefaultStdHTTPClientBufferSize - } - - h := &StdHTTPSink{ - id: id, - proxy: proxy, - config: opts, - addr: net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)), - input: make(chan core.TransportEvent, opts.BufferSize), - done: make(chan struct{}), - logger: logger, - clients: make(map[uint64]*sseClient), - writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond, - } - h.lastProcessed.Store(time.Time{}) - - logger.Info("msg", "Std HTTP sink initialized", - "component", "stdhttp_sink", - "instance_id", id, - "host", opts.Host, - "port", opts.Port, - "stream_path", opts.StreamPath, - "status_path", opts.StatusPath) - return h, nil -} - -// Capabilities returns supported capabilities -func (h *StdHTTPSink) Capabilities() []core.Capability { - // CapTLS/CapAuth appended when transport security lands - return []core.Capability{ - core.CapSessionAware, - core.CapMultiSession, - } -} - -// Input returns the channel for sending transport events -func (h *StdHTTPSink) Input() chan<- core.TransportEvent { - return h.input -} - -// Start binds the listener and serves stream/status endpoints -func (h *StdHTTPSink) Start(ctx context.Context) error { - // IPv4-only, parity with existing network sinks. - // TLS extension point: wrap ln with tls.NewListener (or set - // server.TLSConfig and use ServeTLS); single seam, handlers unchanged. - ln, err := net.Listen("tcp4", h.addr) - if err != nil { - return fmt.Errorf("stdhttp sink bind %s: %w", h.addr, err) - } - - mux := http.NewServeMux() - // Method-scoped patterns: mux answers 405 with Allow header on non-GET - mux.HandleFunc(http.MethodGet+" "+h.config.StreamPath, h.handleStream) - mux.HandleFunc(http.MethodGet+" "+h.config.StatusPath, h.handleStatus) - - // Auth extension point: wrap mux with auth middleware once credentials - // land, e.g. handler = authMiddleware(cfg)(handler) - var handler http.Handler = mux - - h.server = &http.Server{ - Handler: handler, - ReadHeaderTimeout: StdHTTPReadHeaderTimeout, - // WriteTimeout unset by design: SSE responses are long-lived. - // Per-write deadlines via ResponseController in handleStream. - } - h.startTime = time.Now() - - h.wg.Add(1) - go h.brokerLoop(ctx) - - go func() { - if err := h.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { - h.logger.Error("msg", "HTTP server terminated", - "component", "stdhttp_sink", - "instance_id", h.id, - "error", err) - } - }() - - go func() { - select { - case <-ctx.Done(): - h.shutdown() - case <-h.done: - } - }() - - h.logger.Info("msg", "Std HTTP server started", - "component", "stdhttp_sink", - "instance_id", h.id, - "addr", h.addr) - return nil -} - -// Stop gracefully shuts down the sink -func (h *StdHTTPSink) Stop() { - h.logger.Info("msg", "Stopping std HTTP sink", - "component", "stdhttp_sink", - "instance_id", h.id) - - h.shutdown() - h.wg.Wait() - - h.logger.Info("msg", "Std HTTP sink stopped", - "component", "stdhttp_sink", - "instance_id", h.id, - "total_processed", h.totalProcessed.Load()) -} - -// shutdown funnels ctx-cancel and Stop() teardown through a single path. -// done is closed first so SSE handlers exit and Shutdown can complete; -// Server.Close force-closes any handler stalled in a deadline-free write. -func (h *StdHTTPSink) shutdown() { - h.stopOnce.Do(func() { - close(h.done) - if h.server != nil { - sctx, cancel := context.WithTimeout(context.Background(), StdHTTPShutdownTimeout) - defer cancel() - if err := h.server.Shutdown(sctx); err != nil { - h.server.Close() - } - } - }) -} - -// removeClient unregisters a client; the first caller closes the send -// channel and removes the session. Broker (stale-session eviction) and -// stream handler (disconnect) may race here safely. -func (h *StdHTTPSink) removeClient(id uint64) { - h.clientsMu.Lock() - c, ok := h.clients[id] - if ok { - delete(h.clients, id) - } - h.clientsMu.Unlock() - if ok { - close(c.send) - h.proxy.RemoveSession(c.sessionID) - } -} - -// brokerLoop fans out transport events to all client queues, non-blocking, -// and evicts clients whose sessions were idle-expired by the session manager -func (h *StdHTTPSink) brokerLoop(ctx context.Context) { - defer h.wg.Done() - for { - select { - case <-ctx.Done(): - return - case <-h.done: - return - case event, ok := <-h.input: - if !ok { - return - } - h.totalProcessed.Add(1) - h.lastProcessed.Store(time.Now()) - - var stale []uint64 - h.clientsMu.Lock() - for id, c := range h.clients { - if _, exists := h.proxy.GetSession(c.sessionID); !exists { - stale = append(stale, id) - continue - } - select { - case c.send <- event.Payload: - h.proxy.UpdateActivity(c.sessionID) - default: - h.droppedWrites.Add(1) - } - } - h.clientsMu.Unlock() - - for _, id := range stale { - h.removeClient(id) - } - } - } -} - -// handleStream serves one client's SSE stream -func (h *StdHTTPSink) handleStream(w http.ResponseWriter, r *http.Request) { - if h.config.MaxConnections > 0 && h.activeClients.Load() >= h.config.MaxConnections { - h.rejectedClients.Add(1) - http.Error(w, "too many clients", http.StatusServiceUnavailable) - return - } - - rc := http.NewResponseController(w) - remote := r.RemoteAddr - - sess := h.proxy.CreateSession(remote, map[string]any{ - "type": "stdhttp_client", - }) - - c := &sseClient{ - send: make(chan []byte, h.config.ClientBufferSize), - sessionID: sess.ID, - } - id := h.nextClientID.Add(1) - - h.clientsMu.Lock() - h.clients[id] = c - h.clientsMu.Unlock() - - count := h.activeClients.Add(1) - h.logger.Debug("msg", "HTTP client connected", - "component", "stdhttp_sink", - "remote_addr", remote, - "session_id", sess.ID, - "client_id", id, - "active_clients", count) - - defer func() { - h.removeClient(id) - newCount := h.activeClients.Add(-1) - h.logger.Debug("msg", "HTTP client disconnected", - "component", "stdhttp_sink", - "remote_addr", remote, - "session_id", sess.ID, - "client_id", id, - "active_clients", newCount) - }() - - w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache") - w.Header().Set("Access-Control-Allow-Origin", "*") - w.Header().Set("X-Accel-Buffering", "no") - w.WriteHeader(http.StatusOK) - - // Connected event with metadata, parity with fasthttp sink - info, _ := json.Marshal(map[string]any{ - "client_id": strconv.FormatUint(id, 10), - "session_id": sess.ID, - "instance_id": h.id, - "stream_path": h.config.StreamPath, - "status_path": h.config.StatusPath, - "buffer_size": h.config.ClientBufferSize, - }) - fmt.Fprintf(w, "event: connected\ndata: %s\n\n", info) - if err := rc.Flush(); err != nil { - return - } - - clientGone := r.Context().Done() - for { - select { - case payload, ok := <-c.send: - if !ok { - return // broker evicted (stale session) - } - if h.writeTimeout > 0 { - _ = rc.SetWriteDeadline(time.Now().Add(h.writeTimeout)) - } - if err := writeSSE(w, payload); err != nil { - return - } - if err := rc.Flush(); err != nil { - return - } - h.proxy.UpdateActivity(sess.ID) - case <-clientGone: - return - case <-h.done: - fmt.Fprintf(w, "event: disconnect\ndata: {\"reason\":\"server_shutdown\"}\n\n") - rc.Flush() - return - } - } -} - -// handleStatus provides a JSON status report -func (h *StdHTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) { - status := map[string]any{ - "service": "LogWisp", - "version": version.Short(), - "instance_id": h.id, - "server": map[string]any{ - "type": "stdhttp", - "host": h.config.Host, - "port": h.config.Port, - "active_clients": h.activeClients.Load(), - "buffer_size": h.config.BufferSize, - "uptime_seconds": int(time.Since(h.startTime).Seconds()), - }, - "endpoints": map[string]string{ - "stream": h.config.StreamPath, - "status": h.config.StatusPath, - }, - "statistics": map[string]any{ - "total_processed": h.totalProcessed.Load(), - "dropped_writes": h.droppedWrites.Load(), - "rejected_clients": h.rejectedClients.Load(), - }, - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(status) -} - -// GetStats returns sink statistics -func (h *StdHTTPSink) GetStats() sink.SinkStats { - lastProc, _ := h.lastProcessed.Load().(time.Time) - return sink.SinkStats{ - ID: h.id, - Type: "stdhttp", - TotalProcessed: h.totalProcessed.Load(), - ActiveConnections: h.activeClients.Load(), - StartTime: h.startTime, - LastProcessed: lastProc, - Details: map[string]any{ - "host": h.config.Host, - "port": h.config.Port, - "buffer_size": h.config.BufferSize, - "dropped_writes": h.droppedWrites.Load(), - "rejected_clients": h.rejectedClients.Load(), - "endpoints": map[string]string{ - "stream": h.config.StreamPath, - "status": h.config.StatusPath, - }, - }, - } -} - -// writeSSE frames a payload per the W3C SSE spec (multi-line safe) -func writeSSE(w http.ResponseWriter, payload []byte) error { - for _, line := range splitLines(payload) { - if _, err := fmt.Fprintf(w, "data: %s\n", line); err != nil { - return err - } - } - _, err := fmt.Fprint(w, "\n") - return err -} - -// splitLines splits payload by newlines, trimming a single trailing newline -func splitLines(data []byte) [][]byte { - if len(data) == 0 { - return nil - } - if data[len(data)-1] == '\n' { - data = data[:len(data)-1] - } - var lines [][]byte - start := 0 - for i := 0; i < len(data); i++ { - if data[i] == '\n' { - lines = append(lines, data[start:i]) - start = i + 1 - } - } - if start < len(data) { - lines = append(lines, data[start:]) - } - if len(lines) == 0 { - return [][]byte{data} - } - return lines -} diff --git a/internal/sink/stdtcp/stdtcp.go b/internal/sink/stdtcp/stdtcp.go deleted file mode 100644 index 2d57970..0000000 --- a/internal/sink/stdtcp/stdtcp.go +++ /dev/null @@ -1,399 +0,0 @@ -package stdtcp - -import ( - "context" - "errors" - "fmt" - "net" - "strconv" - "sync" - "sync/atomic" - "time" - - "logwisp/internal/config" - "logwisp/internal/core" - "logwisp/internal/plugin" - "logwisp/internal/session" - "logwisp/internal/sink" - - lconfig "github.com/lixenwraith/config" - "github.com/lixenwraith/log" -) - -func init() { - if err := plugin.RegisterSink("stdtcp", NewStdTCPSinkPlugin); err != nil { - panic(fmt.Sprintf("failed to register stdtcp sink: %v", err)) - } -} - -const ( - DefaultStdTCPHost = "0.0.0.0" - DefaultStdTCPBufferSize = 1000 - DefaultStdTCPClientBufferSize = 256 - DefaultStdTCPWriteTimeoutMS = 5000 - DefaultStdTCPKeepAlivePeriodMS = 30000 -) - -// StdTCPSink streams formatted log entries to connected TCP clients using -// only the standard library. Functional peer of the gnet-based tcp sink. -// -// Concurrency model: one broadcast loop fans out into bounded per-client -// queues; each connection owns a writer goroutine (drains queue) and a -// reader goroutine (disconnect detection). A stalled client drops events, -// never the pipeline. -type StdTCPSink struct { - // Plugin identity and session management - id string - proxy *session.Proxy - - // Configuration - config *config.StdTCPSinkOptions - addr string - - // Network - listener net.Listener - - // Application - input chan core.TransportEvent - logger *log.Logger - - // Client registry - clients map[uint64]*tcpClient - clientsMu sync.Mutex - nextClientID atomic.Uint64 - - // Runtime - done chan struct{} - stopOnce sync.Once - wg sync.WaitGroup - startTime time.Time - - writeTimeout time.Duration - - // Statistics - activeConns atomic.Int64 - totalProcessed atomic.Uint64 - writeErrors atomic.Uint64 - droppedWrites atomic.Uint64 - rejectedConns atomic.Uint64 - lastProcessed atomic.Value // time.Time -} - -// tcpClient pairs a connection with its bounded send queue. -// send is written by the broadcast loop (non-blocking) and drained by the -// writer goroutine; closed signals reader-detected disconnect. -type tcpClient struct { - conn net.Conn - send chan []byte - sessionID string - closed chan struct{} -} - -// NewStdTCPSinkPlugin creates a stdtcp sink through plugin factory -func NewStdTCPSinkPlugin( - id string, - configMap map[string]any, - logger *log.Logger, - proxy *session.Proxy, -) (sink.Sink, error) { - opts := &config.StdTCPSinkOptions{ - Host: DefaultStdTCPHost, - KeepAlive: true, - } - if err := lconfig.ScanMap(configMap, opts); err != nil { - return nil, fmt.Errorf("failed to parse config: %w", err) - } - if err := lconfig.Port(opts.Port); err != nil { - return nil, fmt.Errorf("port: %w", err) - } - if opts.BufferSize <= 0 { - opts.BufferSize = DefaultStdTCPBufferSize - } - if opts.ClientBufferSize <= 0 { - opts.ClientBufferSize = DefaultStdTCPClientBufferSize - } - if opts.WriteTimeoutMS <= 0 { - opts.WriteTimeoutMS = DefaultStdTCPWriteTimeoutMS - } - if opts.KeepAlivePeriodMS <= 0 { - opts.KeepAlivePeriodMS = DefaultStdTCPKeepAlivePeriodMS - } - - t := &StdTCPSink{ - id: id, - proxy: proxy, - config: opts, - addr: net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)), - input: make(chan core.TransportEvent, opts.BufferSize), - done: make(chan struct{}), - logger: logger, - clients: make(map[uint64]*tcpClient), - writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond, - } - t.lastProcessed.Store(time.Time{}) - - logger.Info("msg", "Std TCP sink initialized", - "component", "stdtcp_sink", - "instance_id", id, - "host", opts.Host, - "port", opts.Port) - return t, nil -} - -// Capabilities returns supported capabilities -func (t *StdTCPSink) Capabilities() []core.Capability { - // CapTLS/CapAuth appended when transport security lands - return []core.Capability{ - core.CapSessionAware, - core.CapMultiSession, - } -} - -// Input returns the channel for sending transport events -func (t *StdTCPSink) Input() chan<- core.TransportEvent { - return t.input -} - -// listen creates the server listener. -// TLS extension point: wrap the returned listener with tls.NewListener here -// once cert config lands; no other code path changes. mTLS peer identity is -// then available via conn.(*tls.Conn).ConnectionState() in the auth hook. -func (t *StdTCPSink) listen() (net.Listener, error) { - lc := net.ListenConfig{} - if t.config.KeepAlive { - lc.KeepAliveConfig = net.KeepAliveConfig{ - Enable: true, - Idle: time.Duration(t.config.KeepAlivePeriodMS) * time.Millisecond, - } - } - // IPv4-only, parity with existing network sinks - return lc.Listen(context.Background(), "tcp4", t.addr) -} - -// Start binds the listener and launches accept and broadcast loops -func (t *StdTCPSink) Start(ctx context.Context) error { - ln, err := t.listen() - if err != nil { - return fmt.Errorf("stdtcp sink bind %s: %w", t.addr, err) - } - t.listener = ln - t.startTime = time.Now() - - t.wg.Add(2) - go t.acceptLoop() - go t.broadcastLoop(ctx) - - // Pipeline context cancellation mirrors gnet engine stop: cease accepting - // and tear down existing connections - go func() { - select { - case <-ctx.Done(): - t.shutdown() - case <-t.done: - } - }() - - t.logger.Info("msg", "Std TCP server started", - "component", "stdtcp_sink", - "instance_id", t.id, - "addr", t.addr) - return nil -} - -// Stop gracefully shuts down the sink -func (t *StdTCPSink) Stop() { - t.logger.Info("msg", "Stopping std TCP sink", - "component", "stdtcp_sink", - "instance_id", t.id) - - t.shutdown() - t.wg.Wait() - - t.logger.Info("msg", "Std TCP sink stopped", - "component", "stdtcp_sink", - "instance_id", t.id, - "total_processed", t.totalProcessed.Load()) -} - -// shutdown funnels ctx-cancel and Stop() teardown through a single path -func (t *StdTCPSink) shutdown() { - t.stopOnce.Do(func() { - close(t.done) - if t.listener != nil { - t.listener.Close() // unblocks acceptLoop - } - t.clientsMu.Lock() - for _, c := range t.clients { - c.conn.Close() // unblocks per-connection readers - } - t.clientsMu.Unlock() - }) -} - -// acceptLoop accepts client connections until listener close -func (t *StdTCPSink) acceptLoop() { - defer t.wg.Done() - for { - conn, err := t.listener.Accept() - if err != nil { - if errors.Is(err, net.ErrClosed) { - return - } - t.logger.Warn("msg", "Accept error", - "component", "stdtcp_sink", - "error", err) - continue - } - - if t.config.MaxConnections > 0 && t.activeConns.Load() >= t.config.MaxConnections { - // Load/admit race can over-admit by a conn under burst; acceptable - t.rejectedConns.Add(1) - conn.Close() - continue - } - - // Auth extension point: credential/peer verification runs here, - // pre-registration (password preamble read or TLS peer cert check) - - t.wg.Add(1) - go t.handleConn(conn) - } -} - -// handleConn registers the client and runs its writer; a companion reader -// goroutine drains inbound bytes for disconnect detection -func (t *StdTCPSink) handleConn(conn net.Conn) { - defer t.wg.Done() - remote := conn.RemoteAddr().String() - - sess := t.proxy.CreateSession(remote, map[string]any{ - "type": "stdtcp_client", - "remote_addr": remote, - }) - - c := &tcpClient{ - conn: conn, - send: make(chan []byte, t.config.ClientBufferSize), - sessionID: sess.ID, - closed: make(chan struct{}), - } - id := t.nextClientID.Add(1) - - t.clientsMu.Lock() - t.clients[id] = c - t.clientsMu.Unlock() - - count := t.activeConns.Add(1) - t.logger.Debug("msg", "TCP connection opened", - "component", "stdtcp_sink", - "remote_addr", remote, - "session_id", sess.ID, - "active_connections", count) - - defer func() { - t.clientsMu.Lock() - delete(t.clients, id) - t.clientsMu.Unlock() - conn.Close() - <-c.closed // reader has exited - t.proxy.RemoveSession(sess.ID) - newCount := t.activeConns.Add(-1) - t.logger.Debug("msg", "TCP connection closed", - "component", "stdtcp_sink", - "remote_addr", remote, - "active_connections", newCount) - }() - - // Reader: sink is write-only; drain and discard inbound bytes to detect - // disconnect and refresh session activity on client traffic - go func() { - defer close(c.closed) - buf := make([]byte, 4096) - for { - n, err := conn.Read(buf) - if n > 0 { - t.proxy.UpdateActivity(sess.ID) - } - if err != nil { - return - } - } - }() - - // Writer: synchronous stdlib write with deadline. A failed write means - // the kernel buffer stayed full for the full deadline - connection is - // dead or hopelessly stalled, so disconnect immediately (no gnet-style - // consecutive-error counter needed for transient async callback errors). - for { - select { - case data := <-c.send: - if t.writeTimeout > 0 { - conn.SetWriteDeadline(time.Now().Add(t.writeTimeout)) - } - if _, err := conn.Write(data); err != nil { - t.writeErrors.Add(1) - t.logger.Debug("msg", "Write failed, closing client", - "component", "stdtcp_sink", - "remote_addr", remote, - "error", err) - return - } - t.proxy.UpdateActivity(sess.ID) - case <-c.closed: - return - case <-t.done: - return - } - } -} - -// broadcastLoop fans out transport events to all client queues, non-blocking -func (t *StdTCPSink) broadcastLoop(ctx context.Context) { - defer t.wg.Done() - for { - select { - case <-ctx.Done(): - return - case <-t.done: - return - case event, ok := <-t.input: - if !ok { - return - } - t.totalProcessed.Add(1) - t.lastProcessed.Store(time.Now()) - - t.clientsMu.Lock() - for _, c := range t.clients { - select { - case c.send <- event.Payload: - default: - // Slow client: drop its event, never stall siblings - t.droppedWrites.Add(1) - } - } - t.clientsMu.Unlock() - } - } -} - -// GetStats returns sink statistics -func (t *StdTCPSink) GetStats() sink.SinkStats { - lastProc, _ := t.lastProcessed.Load().(time.Time) - return sink.SinkStats{ - ID: t.id, - Type: "stdtcp", - TotalProcessed: t.totalProcessed.Load(), - ActiveConnections: t.activeConns.Load(), - StartTime: t.startTime, - LastProcessed: lastProc, - Details: map[string]any{ - "host": t.config.Host, - "port": t.config.Port, - "buffer_size": t.config.BufferSize, - "write_errors": t.writeErrors.Load(), - "dropped_writes": t.droppedWrites.Load(), - "rejected_conns": t.rejectedConns.Load(), - }, - } -} diff --git a/internal/sink/tcp/tcp.go b/internal/sink/tcp/tcp.go index 4f18c22..b20c10e 100644 --- a/internal/sink/tcp/tcp.go +++ b/internal/sink/tcp/tcp.go @@ -1,10 +1,11 @@ package tcp import ( - "bytes" "context" + "errors" "fmt" "net" + "strconv" "sync" "sync/atomic" "time" @@ -17,8 +18,6 @@ import ( lconfig "github.com/lixenwraith/config" "github.com/lixenwraith/log" - "github.com/lixenwraith/log/compat" - "github.com/panjf2000/gnet/v2" ) func init() { @@ -27,7 +26,18 @@ func init() { } } -// TCPSink streams log entries to connected TCP clients +const ( + DefaultTCPHost = "0.0.0.0" + DefaultTCPBufferSize = 1000 + DefaultTCPClientBufferSize = 256 + DefaultTCPWriteTimeoutMS = 5000 + DefaultTCPKeepAlivePeriodMS = 30000 +) + +// TCPSink streams formatted log entries to connected TCP clients +// Concurrency model: one broadcast loop fans out into bounded per-client queues +// each connection owns a writer goroutine (drains queue) and a reader goroutine (disconnect detection) +// A stalled client drops events, never the pipeline. type TCPSink struct { // Plugin identity and session management id string @@ -35,106 +45,101 @@ type TCPSink struct { // Configuration config *config.TCPSinkOptions + addr string // Network - server *tcpServer - engine *gnet.Engine - engineMu sync.Mutex - booted chan struct{} + listener net.Listener // Application input chan core.TransportEvent logger *log.Logger + // Client registry + clients map[uint64]*tcpClient + clientsMu sync.Mutex + nextClientID atomic.Uint64 + // Runtime done chan struct{} + stopOnce sync.Once wg sync.WaitGroup startTime time.Time + writeTimeout time.Duration + // Statistics activeConns atomic.Int64 totalProcessed atomic.Uint64 + writeErrors atomic.Uint64 + droppedWrites atomic.Uint64 + rejectedConns atomic.Uint64 lastProcessed atomic.Value // time.Time - - // Error tracking - writeErrors atomic.Uint64 - consecutiveWriteErrors map[gnet.Conn]int - errorMu sync.Mutex } -const ( - // Server lifecycle - TCPServerStartTimeout = 2 * time.Second - TCPServerShutdownTimeout = 2 * time.Second +// tcpClient pairs a connection with its bounded send queue. +// send is written by the broadcast loop (non-blocking) and drained by the +// writer goroutine; closed signals reader-detected disconnect. +type tcpClient struct { + conn net.Conn + send chan []byte + sessionID string + closed chan struct{} +} - // Connection management - TCPMaxConsecutiveWriteErrors = 3 - TCPMaxPort = 65535 - - // Defaults - DefaultTCPHost = "0.0.0.0" - DefaultTCPBufferSize = 1000 - DefaultTCPWriteTimeoutMS = 5000 - DefaultTCPKeepAlivePeriod = 30000 -) - -// NewTCPSinkPlugin creates a TCP sink through plugin factory +// NewTCPSinkPlugin creates a tcp sink through plugin factory func NewTCPSinkPlugin( id string, configMap map[string]any, logger *log.Logger, proxy *session.Proxy, ) (sink.Sink, error) { - // Create config struct with defaults opts := &config.TCPSinkOptions{ Host: DefaultTCPHost, - Port: 0, KeepAlive: true, } - - // Parse config map into struct if err := lconfig.ScanMap(configMap, opts); err != nil { return nil, fmt.Errorf("failed to parse config: %w", err) } - - // Validate if err := lconfig.Port(opts.Port); err != nil { return nil, fmt.Errorf("port: %w", err) } - - // Defaults if opts.BufferSize <= 0 { opts.BufferSize = DefaultTCPBufferSize } - if opts.WriteTimeout <= 0 { - opts.WriteTimeout = DefaultTCPWriteTimeoutMS + if opts.ClientBufferSize <= 0 { + opts.ClientBufferSize = DefaultTCPClientBufferSize } - if opts.KeepAlivePeriod <= 0 { - opts.KeepAlivePeriod = DefaultTCPKeepAlivePeriod + if opts.WriteTimeoutMS <= 0 { + opts.WriteTimeoutMS = DefaultTCPWriteTimeoutMS + } + if opts.KeepAlivePeriodMS <= 0 { + opts.KeepAlivePeriodMS = DefaultTCPKeepAlivePeriodMS } t := &TCPSink{ - id: id, - proxy: proxy, - config: opts, - input: make(chan core.TransportEvent, opts.BufferSize), - done: make(chan struct{}), - logger: logger, - consecutiveWriteErrors: make(map[gnet.Conn]int), + id: id, + proxy: proxy, + config: opts, + addr: net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)), + input: make(chan core.TransportEvent, opts.BufferSize), + done: make(chan struct{}), + logger: logger, + clients: make(map[uint64]*tcpClient), + writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond, } t.lastProcessed.Store(time.Time{}) - logger.Info("msg", "TCP sink initialized", + logger.Info("msg", " TCP sink initialized", "component", "tcp_sink", "instance_id", id, "host", opts.Host, "port", opts.Port) - return t, nil } // Capabilities returns supported capabilities func (t *TCPSink) Capabilities() []core.Capability { + // CapTLS/CapAuth appended when transport security lands return []core.Capability{ core.CapSessionAware, core.CapMultiSession, @@ -146,129 +151,232 @@ func (t *TCPSink) Input() chan<- core.TransportEvent { return t.input } -// Start initializes the TCP server and begins the broadcast loop -func (t *TCPSink) Start(ctx context.Context) error { - t.server = &tcpServer{ - sink: t, - clients: make(map[gnet.Conn]*tcpClient), - } - // Fresh channel per Start - t.booted = make(chan struct{}) - - t.startTime = time.Now() - - // Start broadcast loop - t.wg.Add(1) - go func() { - defer t.wg.Done() - t.broadcastLoop(ctx) - }() - - // Configure gnet - addr := fmt.Sprintf("tcp://%s:%d", t.config.Host, t.config.Port) - gnetLogger := compat.NewGnetAdapter(t.logger) - - opts := []gnet.Option{ - gnet.WithLogger(gnetLogger), - gnet.WithMulticore(true), - gnet.WithReusePort(true), - } - - // Apply TCP keep-alive settings from config +// listen creates the server listener. +// TLS extension point: wrap the returned listener with tls.NewListener here +// once cert config lands; no other code path changes. mTLS peer identity is +// then available via conn.(*tls.Conn).ConnectionState() in the auth hook. +func (t *TCPSink) listen() (net.Listener, error) { + lc := net.ListenConfig{} if t.config.KeepAlive { - opts = append(opts, - gnet.WithTCPKeepAlive(time.Duration(t.config.KeepAlivePeriod)*time.Millisecond), - ) - } - - // Start gnet server - errChan := make(chan error, 1) - go func() { - t.logger.Info("msg", "Starting TCP server", - "component", "tcp_sink", - "host", t.config.Host, - "port", t.config.Port) - - err := gnet.Run(t.server, addr, opts...) - if err != nil { - t.logger.Error("msg", "TCP server failed", - "component", "tcp_sink", - "error", err) + lc.KeepAliveConfig = net.KeepAliveConfig{ + Enable: true, + Idle: time.Duration(t.config.KeepAlivePeriodMS) * time.Millisecond, } - errChan <- err - }() - - // Monitor context for shutdown - go func() { - <-ctx.Done() - t.engineMu.Lock() - if t.engine != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - defer cancel() - (*t.engine).Stop(shutdownCtx) - } - t.engineMu.Unlock() - }() - - // Wait briefly for server to start or fail - select { - case err := <-errChan: - close(t.done) - t.wg.Wait() - return err - // Bind confirmation via OnBoot - case <-t.booted: - t.logger.Info("msg", "TCP server started", - "component", "tcp_sink", - "instance_id", t.id, - "port", t.config.Port) - return nil - // Timeout failure - case <-time.After(TCPServerStartTimeout): - t.engineMu.Lock() - if t.engine != nil { - stopCtx, cancel := context.WithTimeout(context.Background(), TCPServerShutdownTimeout) - (*t.engine).Stop(stopCtx) - cancel() - } - t.engineMu.Unlock() - close(t.done) - t.wg.Wait() - return fmt.Errorf("tcp sink start timeout on %s", addr) } + // IPv4-only, parity with existing network sinks + return lc.Listen(context.Background(), "tcp4", t.addr) } -// Stop gracefully shuts down the TCP sink +// Start binds the listener and launches accept and broadcast loops +func (t *TCPSink) Start(ctx context.Context) error { + ln, err := t.listen() + if err != nil { + return fmt.Errorf("tcp sink bind %s: %w", t.addr, err) + } + t.listener = ln + t.startTime = time.Now() + + t.wg.Add(2) + go t.acceptLoop() + go t.broadcastLoop(ctx) + + // Pipeline context cancellation mirrors gnet engine stop: cease accepting + // and tear down existing connections + go func() { + select { + case <-ctx.Done(): + t.shutdown() + case <-t.done: + } + }() + + t.logger.Info("msg", " TCP server started", + "component", "tcp_sink", + "instance_id", t.id, + "addr", t.addr) + return nil +} + +// Stop gracefully shuts down the sink func (t *TCPSink) Stop() { t.logger.Info("msg", "Stopping TCP sink", "component", "tcp_sink", "instance_id", t.id) - close(t.done) - - // Stop gnet engine - t.engineMu.Lock() - engine := t.engine - t.engineMu.Unlock() - - if engine != nil { - ctx, cancel := context.WithTimeout(context.Background(), TCPServerShutdownTimeout) - defer cancel() - (*engine).Stop(ctx) - } - + t.shutdown() t.wg.Wait() - t.logger.Info("msg", "TCP sink stopped", + t.logger.Info("msg", " TCP sink stopped", "component", "tcp_sink", "instance_id", t.id, "total_processed", t.totalProcessed.Load()) } +// shutdown funnels ctx-cancel and Stop() teardown through a single path +func (t *TCPSink) shutdown() { + t.stopOnce.Do(func() { + close(t.done) + if t.listener != nil { + t.listener.Close() // unblocks acceptLoop + } + t.clientsMu.Lock() + for _, c := range t.clients { + c.conn.Close() // unblocks per-connection readers + } + t.clientsMu.Unlock() + }) +} + +// acceptLoop accepts client connections until listener close +func (t *TCPSink) acceptLoop() { + defer t.wg.Done() + for { + conn, err := t.listener.Accept() + if err != nil { + if errors.Is(err, net.ErrClosed) { + return + } + t.logger.Warn("msg", "Accept error", + "component", "tcp_sink", + "error", err) + continue + } + + if t.config.MaxConnections > 0 && t.activeConns.Load() >= t.config.MaxConnections { + // Load/admit race can over-admit by a conn under burst; acceptable + t.rejectedConns.Add(1) + conn.Close() + continue + } + + // Auth extension point: credential/peer verification runs here, + // pre-registration (password preamble read or TLS peer cert check) + + t.wg.Add(1) + go t.handleConn(conn) + } +} + +// handleConn registers the client and runs its writer; a companion reader +// goroutine drains inbound bytes for disconnect detection +func (t *TCPSink) handleConn(conn net.Conn) { + defer t.wg.Done() + remote := conn.RemoteAddr().String() + + sess := t.proxy.CreateSession(remote, map[string]any{ + "type": "tcp_client", + "remote_addr": remote, + }) + + c := &tcpClient{ + conn: conn, + send: make(chan []byte, t.config.ClientBufferSize), + sessionID: sess.ID, + closed: make(chan struct{}), + } + id := t.nextClientID.Add(1) + + t.clientsMu.Lock() + t.clients[id] = c + t.clientsMu.Unlock() + + count := t.activeConns.Add(1) + t.logger.Debug("msg", "TCP connection opened", + "component", "tcp_sink", + "remote_addr", remote, + "session_id", sess.ID, + "active_connections", count) + + defer func() { + t.clientsMu.Lock() + delete(t.clients, id) + t.clientsMu.Unlock() + conn.Close() + <-c.closed // reader has exited + t.proxy.RemoveSession(sess.ID) + newCount := t.activeConns.Add(-1) + t.logger.Debug("msg", "TCP connection closed", + "component", "tcp_sink", + "remote_addr", remote, + "active_connections", newCount) + }() + + // Reader: sink is write-only; drain and discard inbound bytes to detect + // disconnect and refresh session activity on client traffic + go func() { + defer close(c.closed) + buf := make([]byte, 4096) + for { + n, err := conn.Read(buf) + if n > 0 { + t.proxy.UpdateActivity(sess.ID) + } + if err != nil { + return + } + } + }() + + // Writer: synchronous lib write with deadline. A failed write means + // the kernel buffer stayed full for the full deadline - connection is + // dead or hopelessly stalled, so disconnect immediately (no gnet-style + // consecutive-error counter needed for transient async callback errors). + for { + select { + case data := <-c.send: + if t.writeTimeout > 0 { + conn.SetWriteDeadline(time.Now().Add(t.writeTimeout)) + } + if _, err := conn.Write(data); err != nil { + t.writeErrors.Add(1) + t.logger.Debug("msg", "Write failed, closing client", + "component", "tcp_sink", + "remote_addr", remote, + "error", err) + return + } + t.proxy.UpdateActivity(sess.ID) + case <-c.closed: + return + case <-t.done: + return + } + } +} + +// broadcastLoop fans out transport events to all client queues, non-blocking +func (t *TCPSink) broadcastLoop(ctx context.Context) { + defer t.wg.Done() + for { + select { + case <-ctx.Done(): + return + case <-t.done: + return + case event, ok := <-t.input: + if !ok { + return + } + t.totalProcessed.Add(1) + t.lastProcessed.Store(time.Now()) + + t.clientsMu.Lock() + for _, c := range t.clients { + select { + case c.send <- event.Payload: + default: + // Slow client: drop its event, never stall siblings + t.droppedWrites.Add(1) + } + } + t.clientsMu.Unlock() + } + } +} + // GetStats returns sink statistics func (t *TCPSink) GetStats() sink.SinkStats { lastProc, _ := t.lastProcessed.Load().(time.Time) - return sink.SinkStats{ ID: t.id, Type: "tcp", @@ -277,217 +385,12 @@ func (t *TCPSink) GetStats() sink.SinkStats { StartTime: t.startTime, LastProcessed: lastProc, Details: map[string]any{ - "host": t.config.Host, - "port": t.config.Port, - "buffer_size": t.config.BufferSize, - "write_errors": t.writeErrors.Load(), + "host": t.config.Host, + "port": t.config.Port, + "buffer_size": t.config.BufferSize, + "write_errors": t.writeErrors.Load(), + "dropped_writes": t.droppedWrites.Load(), + "rejected_conns": t.rejectedConns.Load(), }, } } - -// tcpServer implements gnet.EventHandler -type tcpServer struct { - gnet.BuiltinEventEngine - sink *TCPSink - clients map[gnet.Conn]*tcpClient - mu sync.RWMutex -} - -// tcpClient represents a connected TCP client -type tcpClient struct { - conn gnet.Conn - buffer bytes.Buffer - sessionID string -} - -// broadcastLoop sends transport events to all connected clients -func (t *TCPSink) broadcastLoop(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case event, ok := <-t.input: - if !ok { - return - } - t.totalProcessed.Add(1) - t.lastProcessed.Store(time.Now()) - t.broadcastData(event.Payload) - case <-t.done: - return - } - } -} - -// OnBoot is called when the server starts -func (s *tcpServer) OnBoot(eng gnet.Engine) gnet.Action { - s.sink.engineMu.Lock() - s.sink.engine = &eng - s.sink.engineMu.Unlock() - - // Listener is bound at this point; unblock Start - close(s.sink.booted) - - s.sink.logger.Debug("msg", "TCP server booted", - "component", "tcp_sink", - "instance_id", s.sink.id) - return gnet.None -} - -// OnOpen is called when a new connection is established -func (s *tcpServer) OnOpen(c gnet.Conn) (out []byte, action gnet.Action) { - remoteAddr := c.RemoteAddr() - remoteAddrStr := remoteAddr.String() - - s.sink.logger.Debug("msg", "TCP connection attempt", - "component", "tcp_sink", - "remote_addr", remoteAddrStr) - - // Reject IPv6 connections - if tcpAddr, ok := remoteAddr.(*net.TCPAddr); ok { - if tcpAddr.IP.To4() == nil { - s.sink.logger.Warn("msg", "IPv6 connection rejected", - "component", "tcp_sink", - "remote_addr", remoteAddrStr) - return []byte("IPv4-only (IPv6 not supported)\n"), gnet.Close - } - } - - // Apply write timeout from config - if s.sink.config.WriteTimeout > 0 { - c.SetWriteDeadline(time.Now().Add(time.Duration(s.sink.config.WriteTimeout) * time.Millisecond)) - } - - // Create session via proxy - sess := s.sink.proxy.CreateSession(remoteAddrStr, map[string]any{ - "type": "tcp_client", - "remote_addr": remoteAddrStr, - }) - - client := &tcpClient{ - conn: c, - sessionID: sess.ID, - } - - s.mu.Lock() - s.clients[c] = client - s.mu.Unlock() - - newCount := s.sink.activeConns.Add(1) - s.sink.logger.Debug("msg", "TCP connection opened", - "component", "tcp_sink", - "remote_addr", remoteAddrStr, - "session_id", sess.ID, - "active_connections", newCount) - - return nil, gnet.None -} - -// OnClose is called when a connection is closed -func (s *tcpServer) OnClose(c gnet.Conn, err error) gnet.Action { - remoteAddrStr := c.RemoteAddr().String() - - s.mu.RLock() - client, exists := s.clients[c] - s.mu.RUnlock() - - if exists && client.sessionID != "" { - s.sink.proxy.RemoveSession(client.sessionID) - s.sink.logger.Debug("msg", "Session removed", - "component", "tcp_sink", - "session_id", client.sessionID, - "remote_addr", remoteAddrStr) - } - - s.mu.Lock() - delete(s.clients, c) - s.mu.Unlock() - - s.sink.errorMu.Lock() - delete(s.sink.consecutiveWriteErrors, c) - s.sink.errorMu.Unlock() - - newCount := s.sink.activeConns.Add(-1) - s.sink.logger.Debug("msg", "TCP connection closed", - "component", "tcp_sink", - "remote_addr", remoteAddrStr, - "active_connections", newCount, - "error", err) - - return gnet.None -} - -// OnTraffic is called when data is received from a connection -func (s *tcpServer) OnTraffic(c gnet.Conn) gnet.Action { - s.mu.RLock() - client, exists := s.clients[c] - s.mu.RUnlock() - - // Update session activity - if exists && client.sessionID != "" { - s.sink.proxy.UpdateActivity(client.sessionID) - } - - // TCP sink doesn't expect data from clients, discard safely - if bufLen := c.InboundBuffered(); bufLen > 0 { - c.Next(bufLen) - } - return gnet.None -} - -// broadcastData sends data to all connected clients -func (t *TCPSink) broadcastData(data []byte) { - t.server.mu.RLock() - defer t.server.mu.RUnlock() - - for conn, client := range t.server.clients { - // Update session activity - if client.sessionID != "" { - t.proxy.UpdateActivity(client.sessionID) - } - - // Refresh write deadline on each write if configured - if t.config.WriteTimeout > 0 { - conn.SetWriteDeadline(time.Now().Add(time.Duration(t.config.WriteTimeout) * time.Millisecond)) - } - - conn.AsyncWrite(data, func(c gnet.Conn, err error) error { - if err != nil { - t.writeErrors.Add(1) - t.handleWriteError(c, err) - } else { - t.errorMu.Lock() - delete(t.consecutiveWriteErrors, c) - t.errorMu.Unlock() - } - return nil - }) - } -} - -// handleWriteError manages errors during async writes -func (t *TCPSink) handleWriteError(c gnet.Conn, err error) { - remoteAddrStr := c.RemoteAddr().String() - - t.errorMu.Lock() - defer t.errorMu.Unlock() - - t.consecutiveWriteErrors[c]++ - errorCount := t.consecutiveWriteErrors[c] - - t.logger.Debug("msg", "AsyncWrite error", - "component", "tcp_sink", - "remote_addr", remoteAddrStr, - "error", err, - "consecutive_errors", errorCount) - - // Close connection max consecutive write errors - if errorCount >= TCPMaxConsecutiveWriteErrors { - t.logger.Warn("msg", "Closing connection due to repeated write errors", - "component", "tcp_sink", - "remote_addr", remoteAddrStr, - "error_count", errorCount) - delete(t.consecutiveWriteErrors, c) - c.Close() - } -} diff --git a/test/chain-aggregate-test.sh b/test/chain-aggregate-test.sh index f532e44..ba200bf 100755 --- a/test/chain-aggregate-test.sh +++ b/test/chain-aggregate-test.sh @@ -1,8 +1,8 @@ #!/usr/bin/env bash # logwisp chain topology test # -# random --> tcp_chain sink --> :15801 tcp_chain src --> :15803 stdtcp sink -# random --> http_chain sink --> :15802 http_chain src --> :15804 stdhttp sink (SSE) +# random --> tcp_chain sink --> :15801 tcp_chain src --> :15803 tcp sink +# random --> http_chain sink --> :15802 http_chain src --> :15804 http sink (SSE) # # Usage: # ./chain_aggregate_test.sh manual mode: 2 edge daemons + relay foreground @@ -163,14 +163,14 @@ port = $PORT_HTTP_CHAIN [[pipelines.plugin_sinks]] id = "out_tcp" -type = "stdtcp" +type = "tcp" [pipelines.plugin_sinks.config] host = "127.0.0.1" port = $PORT_TCP_SINK [[pipelines.plugin_sinks]] id = "out_http" -type = "stdhttp" +type = "http" [pipelines.plugin_sinks.config] host = "127.0.0.1" port = $PORT_HTTP_SINK @@ -220,8 +220,10 @@ check() { # label condition_result # 1. TCP chain: edge-tcp -> relay -> tcp sink tcp_out="$(tcp_read "$PORT_TCP_SINK" 4)" -nt=$(grep -c '"node":"edge-tcp"' <<< "$tcp_out") -nh=$(grep -c '"node":"edge-http"' <<< "$tcp_out") +#nt=$(grep -c '"node":"edge-tcp"' <<< "$tcp_out") +#nh=$(grep -c '"node":"edge-http"' <<< "$tcp_out") +nt=$(grep -c '"source":"edge-tcp/' <<< "$tcp_out") +nh=$(grep -c '"source":"edge-http/' <<< "$tcp_out") check "tcp sink: aggregated edge-tcp ($nt) + edge-http ($nh)" $(( nt >= 1 && nh >= 1 )) # 2. HTTP chain: edge-http -> relay -> SSE sink @@ -232,7 +234,7 @@ check "http sink: aggregated edge-tcp ($nt) + edge-http ($nh)" $(( nt >= 1 && nh # 3. HTTP sink status endpoint status="$(curl -s --max-time 3 "http://127.0.0.1:$PORT_HTTP_SINK/status" || true)" -proc=$(grep -o '"total_processed":[0-9]*' <<< "$status" | grep -o '[0-9]*' || echo 0) +proc=$(grep -o '"total_processed"[ :] *[0-9]*' <<< "$status" | grep -o '[0-9]*' || echo 0) check "status endpoint: total_processed=$proc > 0" $(( proc > 0 )) echo "================================================================" diff --git a/test/chain-test.sh b/test/chain-test.sh index 7bb8456..254bff6 100755 --- a/test/chain-test.sh +++ b/test/chain-test.sh @@ -223,7 +223,8 @@ check() { # label condition_result # 1. TCP chain: edge-tcp -> relay -> tcp sink tcp_out="$(tcp_read "$PORT_TCP_SINK" 4)" -n=$(grep -c '"node":"edge-tcp"' <<< "$tcp_out") +#n=$(grep -c '"node":"edge-tcp"' <<< "$tcp_out") +n=$(grep -c '"source":"edge-tcp/' <<< "$tcp_out") check "tcp path: entries on :$PORT_TCP_SINK with node=edge-tcp ($n lines)" $(( n >= 1 )) # 2. HTTP chain: edge-http -> relay -> SSE sink @@ -233,7 +234,7 @@ check "http path: SSE events on :$PORT_HTTP_SINK with node=edge-http ($n events) # 3. HTTP sink status endpoint status="$(curl -s --max-time 3 "http://127.0.0.1:$PORT_HTTP_SINK/status" || true)" -proc=$(grep -o '"total_processed":[0-9]*' <<< "$status" | grep -o '[0-9]*' || echo 0) +proc=$(grep -o '"total_processed"[ :] *[0-9]*' <<< "$status" | grep -o '[0-9]*' || echo 0) check "status endpoint: total_processed=$proc > 0" $(( proc > 0 )) echo "================================================================"