From 95dc4568f6469195770d5b7e3d21074307d8f3d7 Mon Sep 17 00:00:00 2001 From: k2safe Date: Tue, 22 Sep 2026 18:19:39 +0800 Subject: [PATCH] feat: add trusted proxy identity and overload stats --- CHANGELOG.md | 1 + README.md | 7 ++- ROADMAP.md | 1 + app/app.go | 26 +++++--- app/router.go | 3 + app_test.go | 31 ++++++++++ config/config.go | 19 +++++- config/config_test.go | 16 ++++- config/env.go | 6 ++ httpx/clientIP.go | 105 ++++++++++++++++++++++++++++++++ httpx/clientIP_test.go | 76 +++++++++++++++++++++++ middleware/middleware.go | 13 +--- middleware/middleware_test.go | 31 ++++++++++ omni.go | 2 + transport/README.md | 6 +- transport/native/server.go | 39 +++++++----- transport/native/server_test.go | 6 ++ 17 files changed, 346 insertions(+), 42 deletions(-) create mode 100644 httpx/clientIP.go create mode 100644 httpx/clientIP_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index b410be9..03ed3f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- native `Server.Stats()` 新增连接和请求过载拒绝计数；新增可信代理配置，访问日志只在 socket 对端属于配置网段时解析 `X-Forwarded-For`。 - 新增 transport-neutral `metrics` 包，支持请求数、错误数、in-flight、累计耗时和状态分布，并可直接注册 `/metrics` 接口。 - native 新增 `Server.Stats()`，提供 accepted connections、active connections 和 active requests 生命周期指标。 - native 请求体改为流式读取，Content-Length 与 chunked body 不再在进入 Handler 前整体复制到内存；保留大小限制、读超时、trailer 校验和 Keep-Alive 排空。 diff --git a/README.md b/README.md index c13420d..17b746c 100644 --- a/README.md +++ b/README.md @@ -124,7 +124,7 @@ if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil { `transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 和 chunked 请求体、`http.Flusher` 流式响应、可选 TLS、读写超时和优雅关闭；不支持的传输编码会被拒绝。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。 -自研网络层还提供 `server.Stats()`，可读取 accepted、active connections 和 active requests，与上面的业务请求 metrics 组合后能区分连接堆积和 Handler 慢请求。 +自研网络层还提供 `server.Stats()`，可读取 accepted、rejected、active connections 和 active requests，与上面的业务请求 metrics 组合后能区分连接堆积、过载拒绝和 Handler 慢请求。`rejectedConnections` 表示连接闸门拒绝数，`rejectedRequests` 表示并发请求闸门返回 503 的次数。 业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库： @@ -164,7 +164,10 @@ application.OnShutdown(logger.Close) 常用变量包括 `GO_OMNI_ENABLE_LOGGER`、`GO_OMNI_LOG_LEVEL`、 `GO_OMNI_LOG_FORMAT`、`GO_OMNI_LOG_OUTPUT`、`GO_OMNI_SERVICE`、 -`GO_OMNI_ENVIRONMENT`、`GO_OMNI_HTTP_ADDR` 和 `GO_OMNI_SHUTDOWN_TIMEOUT`。数据库连接仍由 +`GO_OMNI_ENVIRONMENT`、`GO_OMNI_HTTP_ADDR`、`GO_OMNI_SHUTDOWN_TIMEOUT` 和 +`GO_OMNI_TRUSTED_PROXIES`。可信代理变量使用逗号分隔的 IP/CIDR，例如 +`GO_OMNI_TRUSTED_PROXIES=127.0.0.1,10.0.0.0/8`；只有来自这些地址的连接才会解析 +`X-Forwarded-For`，默认只记录 socket 对端地址。数据库连接仍由 `database.ConfigFromEnv` 从 `GO_OMNI_DB_*` 加载。 `App` 同时实现标准 `http.Handler`。`Run` 监听 `SIGINT` / `SIGTERM`；`RunContext` 则由调用方的 context 控制退出，生成项目已使用 `signal.NotifyContext` 接好信号。 diff --git a/ROADMAP.md b/ROADMAP.md index 224fe6f..eaf8d93 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -48,6 +48,7 @@ - 已补充慢请求头、慢 Content-Length body、慢 chunked body、慢 TLS 握手、大请求体、HTTP/1.0 Keep-Alive、HEAD/204 和管线请求验收；Linux/arm64 容器已通过全量 race、vet、build，以及 3 秒 TLS 并发/空闲连接矩阵。原始报告见 [tlsBaselineLinux.json](performance/tlsBaselineLinux.json)。 - 已验证 `MaxConnections` 满载时快速拒绝新连接、请求完成后释放容量；后续仍需在目标部署环境观察文件描述符、内核 socket 缓冲和反向代理联动。 - 已验证 `MaxConcurrentRequests` 满载返回 503，活动请求结束后槽位释放，后续请求可以恢复处理。 +- native `Stats()` 已记录连接和请求过载拒绝数；访问日志已支持可信代理网段配置，默认拒绝伪造的 `X-Forwarded-For`。 - 已用 16 个并发 worker 完成 3,200 次建连/断连 race 验收；MaxConnections 满载拒绝、槽位释放和 TLS 连接上限均已覆盖。 - 新增 `scripts/verifyLinux.py`，固定在 Go 1.25 Linux 容器内运行 race、vet、build 和可选 TLS/非 TLS perfcheck；perfcheck 记录堆、goroutine、连接和 Linux 文件描述符。 - Linux/arm64 已完成 30 秒、128 并发、512 空闲连接的 TLS 与非 TLS 持续回归，报告见 `performance/linuxSoak.json` 和 `performance/linuxSoakTLS.json`，均为 0 错误且连接、goroutine、FD 回收正常。 diff --git a/app/app.go b/app/app.go index c2c7e40..04b6881 100644 --- a/app/app.go +++ b/app/app.go @@ -21,12 +21,13 @@ import ( // App is the top-level GoOmni application. Its public behavior depends on // httpx.Router, leaving the network implementation replaceable. type App struct { - router httpx.Router - config config.Config - middlewares []httpx.Middleware - closers []func() error - closeOnce sync.Once - closeErr error + router httpx.Router + config config.Config + middlewares []httpx.Middleware + closers []func() error + closeOnce sync.Once + closeErr error + clientIPResolver *httpx.ClientIPResolver } // Runner is the small lifecycle boundary used by an HTTP server @@ -63,12 +64,17 @@ func configure(options ...config.Option) config.Config { } func newApp(router httpx.Router, settings config.Config) *App { + settings.TrustedProxies = append([]string(nil), settings.TrustedProxies...) + resolver, err := httpx.NewClientIPResolver(settings.TrustedProxies...) + if err != nil { + panic("go-omni: " + err.Error()) + } middlewares := []httpx.Middleware{middleware.RequestID(settings.RequestIDHeader)} if settings.EnableLogger { middlewares = append(middlewares, middleware.AccessLog()) } middlewares = append(middlewares, middleware.Recovery()) - return &App{router: router, config: settings, middlewares: middlewares} + return &App{router: router, config: settings, middlewares: middlewares, clientIPResolver: resolver} } // Router returns the transport contract used by this application. @@ -84,7 +90,11 @@ func (a *App) ServeHTTP(writer http.ResponseWriter, request *http.Request) { } // Config returns the effective application configuration. -func (a *App) Config() config.Config { return a.config } +func (a *App) Config() config.Config { + settings := a.config + settings.TrustedProxies = append([]string(nil), settings.TrustedProxies...) + return settings +} // Use adds middleware to routes registered after this call. func (a *App) Use(middleware ...httpx.Middleware) *App { diff --git a/app/router.go b/app/router.go index 192ce37..e5b991c 100644 --- a/app/router.go +++ b/app/router.go @@ -21,6 +21,9 @@ func (a *App) adapt(handler httpx.Handler, middleware ...httpx.Middleware) http. } } return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if len(a.config.TrustedProxies) > 0 { + request = request.WithContext(httpx.WithClientIPResolver(request.Context(), a.clientIPResolver)) + } requestContext := httpx.NewContext(request, writer, httpx.RouteParams(request.Context())) if err := wrapped(requestContext); err != nil { a.writeError(requestContext, err) diff --git a/app_test.go b/app_test.go index a18be6c..902476f 100644 --- a/app_test.go +++ b/app_test.go @@ -128,3 +128,34 @@ func TestAppUsesConfiguredAddressWhenNoOverrideIsProvided(t *testing.T) { t.Fatalf("expected configured address %q, got %q", settings.Address, got) } } + +func TestAppConfigCopiesTrustedProxies(t *testing.T) { + proxies := []string{"127.0.0.1"} + app := New(WithTrustedProxies(proxies...)) + proxies[0] = "10.0.0.0/8" + settings := app.Config() + if len(settings.TrustedProxies) != 1 || settings.TrustedProxies[0] != "127.0.0.1" { + t.Fatalf("trusted proxy config was not copied: %+v", settings.TrustedProxies) + } + settings.TrustedProxies[0] = "192.0.2.0/24" + if app.Config().TrustedProxies[0] != "127.0.0.1" { + t.Fatal("App.Config should return a copy") + } +} + +func TestAppTrustedProxyClientIPReachesHandlers(t *testing.T) { + app := New(WithTrustedProxies("127.0.0.1")) + app.GET("/client-ip", func(ctx *Context) error { return ctx.OK(ctx.ClientIP()) }) + request := httptest.NewRequest(http.MethodGet, "/client-ip", nil) + request.RemoteAddr = "127.0.0.1:443" + request.Header.Set("X-Forwarded-For", "198.51.100.7") + recorder := httptest.NewRecorder() + app.Engine().ServeHTTP(recorder, request) + var response Response + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + if response.Data != "198.51.100.7" { + t.Fatalf("expected trusted proxy client IP, got %#v", response.Data) + } +} diff --git a/config/config.go b/config/config.go index 31fb3f5..78fb187 100644 --- a/config/config.go +++ b/config/config.go @@ -4,6 +4,7 @@ import ( "strings" "time" + "git.yuntuo.hk/k2safe/go-omni/httpx" "git.yuntuo.hk/k2safe/go-omni/logx" ) @@ -14,6 +15,7 @@ type Config struct { Address string RequestIDHeader string EnableLogger bool + TrustedProxies []string ShutdownTimeout time.Duration Log logx.Config } @@ -38,6 +40,9 @@ func (c Config) Validate() error { if c.ShutdownTimeout <= 0 { return &ConfigError{Field: "shutdown_timeout", Message: "must be positive"} } + if _, err := httpx.NewClientIPResolver(c.TrustedProxies...); err != nil { + return &ConfigError{Field: "trusted_proxies", Message: err.Error()} + } return nil } @@ -66,10 +71,22 @@ func WithLogger(enabled bool) Option { return func(cfg *Config) { cfg.EnableLogger = enabled } } +// WithTrustedProxies allows only these IPs/CIDRs to supply X-Forwarded-For. +// An empty list disables forwarding-header trust. Invalid entries are rejected +// by Config.Validate and panic when passed directly to App.New. +func WithTrustedProxies(proxies ...string) Option { + proxies = append([]string(nil), proxies...) + return func(cfg *Config) { cfg.TrustedProxies = append([]string(nil), proxies...) } +} + // WithConfig applies a complete configuration loaded from a file or // environment. Later options can still override individual fields. func WithConfig(settings Config) Option { - return func(cfg *Config) { *cfg = settings } + settings.TrustedProxies = append([]string(nil), settings.TrustedProxies...) + return func(cfg *Config) { + *cfg = settings + cfg.TrustedProxies = append([]string(nil), settings.TrustedProxies...) + } } // WithLogConfig configures the logger used by the application runtime. diff --git a/config/config_test.go b/config/config_test.go index 8c5eece..da2141b 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -9,6 +9,7 @@ import ( func TestLoadFromEnvUsesDefaults(t *testing.T) { for _, key := range []string{ "GO_OMNI_HTTP_ADDR", "GO_OMNI_REQUEST_ID_HEADER", "GO_OMNI_ENABLE_LOGGER", + "GO_OMNI_TRUSTED_PROXIES", "GO_OMNI_LOG_LEVEL", "GO_OMNI_LOG_FORMAT", "GO_OMNI_LOG_OUTPUT", "GO_OMNI_LOG_FILE", "GO_OMNI_SERVICE", "GO_OMNI_ENVIRONMENT", "GO_OMNI_LOG_ADD_SOURCE", "GO_OMNI_SHUTDOWN_TIMEOUT", @@ -31,6 +32,7 @@ func TestLoadFromEnvUsesDefaults(t *testing.T) { func TestLoadFromEnvOverridesSettings(t *testing.T) { t.Setenv("GO_OMNI_HTTP_ADDR", "127.0.0.1:9090") t.Setenv("GO_OMNI_REQUEST_ID_HEADER", "X-Correlation-ID") + t.Setenv("GO_OMNI_TRUSTED_PROXIES", "127.0.0.1, 10.0.0.0/8") t.Setenv("GO_OMNI_ENABLE_LOGGER", "true") t.Setenv("GO_OMNI_LOG_LEVEL", "debug") t.Setenv("GO_OMNI_LOG_FORMAT", "text") @@ -44,7 +46,7 @@ func TestLoadFromEnvOverridesSettings(t *testing.T) { if err != nil { t.Fatal(err) } - if cfg.Address != "127.0.0.1:9090" || cfg.RequestIDHeader != "X-Correlation-ID" || !cfg.EnableLogger { + if cfg.Address != "127.0.0.1:9090" || cfg.RequestIDHeader != "X-Correlation-ID" || !cfg.EnableLogger || len(cfg.TrustedProxies) != 2 || cfg.TrustedProxies[1] != "10.0.0.0/8" { t.Fatalf("unexpected application settings: %+v", cfg) } if cfg.ShutdownTimeout != 3*time.Second || cfg.Log.Level != "debug" || cfg.Log.Format != "text" || cfg.Log.Output != "stdout" || cfg.Log.Service != "billing" || cfg.Log.Environment != "test" || !cfg.Log.AddSource { @@ -65,6 +67,11 @@ func TestLoadFromEnvRejectsInvalidValues(t *testing.T) { } t.Setenv("GO_OMNI_SHUTDOWN_TIMEOUT", "1s") + t.Setenv("GO_OMNI_TRUSTED_PROXIES", "proxy.example.com") + if _, err := LoadFromEnv(""); err == nil || !strings.Contains(err.Error(), "trusted_proxies") { + t.Fatalf("expected invalid trusted proxy error, got %v", err) + } + t.Setenv("GO_OMNI_TRUSTED_PROXIES", "") cfg := Default() cfg.Address = "" if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "address") { @@ -80,11 +87,16 @@ func TestLoadFromEnvRejectsInvalidValues(t *testing.T) { func TestWithConfigAndWithLogConfig(t *testing.T) { settings := Default() settings.Log.Service = "orders" + settings.TrustedProxies = []string{"127.0.0.1"} cfg := Default() WithConfig(settings)(&cfg) - if cfg.Log.Service != "orders" { + if cfg.Log.Service != "orders" || len(cfg.TrustedProxies) != 1 { t.Fatalf("complete config was not applied: %+v", cfg) } + settings.TrustedProxies[0] = "10.0.0.0/8" + if cfg.TrustedProxies[0] != "127.0.0.1" { + t.Fatal("WithConfig should copy trusted proxy settings") + } WithLogConfig(Default().Log)(&cfg) if cfg.Log.Service != "" { diff --git a/config/env.go b/config/env.go index bfce5c8..7ca2aa3 100644 --- a/config/env.go +++ b/config/env.go @@ -20,6 +20,12 @@ func LoadFromEnv(prefix string) (Config, error) { cfg := Default() cfg.Address = envString(prefix+"HTTP_ADDR", cfg.Address) cfg.RequestIDHeader = envString(prefix+"REQUEST_ID_HEADER", cfg.RequestIDHeader) + if value := strings.TrimSpace(os.Getenv(prefix + "TRUSTED_PROXIES")); value != "" { + cfg.TrustedProxies = strings.Split(value, ",") + for i := range cfg.TrustedProxies { + cfg.TrustedProxies[i] = strings.TrimSpace(cfg.TrustedProxies[i]) + } + } cfg.Log.Level = envString(prefix+"LOG_LEVEL", cfg.Log.Level) cfg.Log.Format = envString(prefix+"LOG_FORMAT", cfg.Log.Format) cfg.Log.Output = envString(prefix+"LOG_OUTPUT", cfg.Log.Output) diff --git a/httpx/clientIP.go b/httpx/clientIP.go new file mode 100644 index 0000000..02a01a5 --- /dev/null +++ b/httpx/clientIP.go @@ -0,0 +1,105 @@ +package httpx + +import ( + "context" + "fmt" + "net" + "net/http" + "net/netip" + "strings" +) + +// ClientIPResolver trusts X-Forwarded-For only from explicitly configured +// proxy addresses. It is immutable and safe to share between requests. +// The zero value (and a nil resolver) trusts no proxy. +type ClientIPResolver struct { + trusted []netip.Prefix +} + +// NewClientIPResolver accepts IP addresses or CIDR networks, never hostnames. +func NewClientIPResolver(proxies ...string) (*ClientIPResolver, error) { + r := &ClientIPResolver{} + for _, value := range proxies { + value = strings.TrimSpace(value) + if address, err := netip.ParseAddr(value); err == nil && address.Zone() == "" { + address = address.Unmap() + r.trusted = append(r.trusted, netip.PrefixFrom(address, address.BitLen())) + continue + } + prefix, err := netip.ParsePrefix(value) + if err != nil { + return nil, fmt.Errorf("invalid trusted proxy %q: expected an IP address or CIDR", value) + } + if prefix.Addr().Is4In6() { + if prefix.Bits() < 96 { + return nil, fmt.Errorf("invalid trusted proxy %q: mapped IPv4 prefix must be at least /96", value) + } + prefix = netip.PrefixFrom(prefix.Addr().Unmap(), prefix.Bits()-96) + } + r.trusted = append(r.trusted, prefix.Masked()) + } + return r, nil +} + +func (r *ClientIPResolver) trusts(address netip.Addr) bool { + if r != nil { + for _, prefix := range r.trusted { + if prefix.Contains(address) { + return true + } + } + } + return false +} + +// ClientIP walks X-Forwarded-For from right to left, stopping at the first +// untrusted address. Missing or malformed forwarding data falls back to the +// socket peer. X-Real-IP, Forwarded and provider-specific headers are ignored. +func (r *ClientIPResolver) ClientIP(request *http.Request) string { + if request == nil { + return "" + } + host := request.RemoteAddr + if value, _, err := net.SplitHostPort(host); err == nil { + host = value + } + peer, err := netip.ParseAddr(host) + if err != nil { + return "" + } + peer = peer.WithZone("").Unmap() + if !r.trusts(peer) { + return peer.String() + } + values := request.Header.Values("X-Forwarded-For") + var chain []netip.Addr + for _, value := range values { + for _, part := range strings.Split(value, ",") { + address, err := netip.ParseAddr(strings.TrimSpace(part)) + if err != nil || address.Zone() != "" { + return peer.String() + } + chain = append(chain, address.Unmap()) + } + } + current := peer + for i := len(chain) - 1; i >= 0 && r.trusts(current); i-- { + current = chain[i] + } + return current.String() +} + +type clientIPResolverKey struct{} + +// WithClientIPResolver configures ClientIP for Context values derived from ctx. +// Adapters should set this before running application middleware. +func WithClientIPResolver(ctx context.Context, resolver *ClientIPResolver) context.Context { + return context.WithValue(ctx, clientIPResolverKey{}, resolver) +} + +// ClientIP returns the verified client address without a port. By default it +// returns the socket peer; App's trusted proxy configuration enables XFF. +func (c *Context) ClientIP() string { + resolver, _ := c.Context().Value(clientIPResolverKey{}).(*ClientIPResolver) + return resolver.ClientIP(c.request) +} diff --git a/httpx/clientIP_test.go b/httpx/clientIP_test.go new file mode 100644 index 0000000..38cc9a6 --- /dev/null +++ b/httpx/clientIP_test.go @@ -0,0 +1,76 @@ +package httpx_test + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "git.yuntuo.hk/k2safe/go-omni/httpx" +) + +func TestClientIPResolverUsesSocketPeerByDefault(t *testing.T) { + resolver, err := httpx.NewClientIPResolver() + if err != nil { + t.Fatal(err) + } + request := newClientIPRequest("192.0.2.10:443", "198.51.100.7") + if got := resolver.ClientIP(request); got != "192.0.2.10" { + t.Fatalf("untrusted forwarding header changed client IP: %q", got) + } +} + +func TestClientIPResolverWalksTrustedForwardingChain(t *testing.T) { + resolver, err := httpx.NewClientIPResolver("10.0.0.0/8") + if err != nil { + t.Fatal(err) + } + request := newClientIPRequest("10.0.0.5:443", "198.51.100.7, 10.0.0.4") + if got := resolver.ClientIP(request); got != "198.51.100.7" { + t.Fatalf("expected original client IP, got %q", got) + } + + request = newClientIPRequest("10.0.0.5:443", "198.51.100.7, 203.0.113.9, 10.0.0.4") + if got := resolver.ClientIP(request); got != "203.0.113.9" { + t.Fatalf("expected first untrusted hop, got %q", got) + } +} + +func TestClientIPResolverRejectsMalformedForwardingData(t *testing.T) { + resolver, err := httpx.NewClientIPResolver("127.0.0.1") + if err != nil { + t.Fatal(err) + } + request := newClientIPRequest("127.0.0.1:443", "198.51.100.7, not-an-ip") + if got := resolver.ClientIP(request); got != "127.0.0.1" { + t.Fatalf("malformed forwarding header should fall back to peer, got %q", got) + } +} + +func TestNewClientIPResolverRejectsHostnames(t *testing.T) { + if _, err := httpx.NewClientIPResolver("proxy.example.com"); err == nil || !strings.Contains(err.Error(), "invalid trusted proxy") { + t.Fatalf("expected hostname validation error, got %v", err) + } +} + +func TestContextClientIPUsesResolverFromRequestContext(t *testing.T) { + resolver, err := httpx.NewClientIPResolver("127.0.0.1") + if err != nil { + t.Fatal(err) + } + request := newClientIPRequest("127.0.0.1:443", "198.51.100.7") + request = request.WithContext(httpx.WithClientIPResolver(request.Context(), resolver)) + ctx := httpx.NewContext(request, httptest.NewRecorder(), nil) + if got := ctx.ClientIP(); got != "198.51.100.7" { + t.Fatalf("expected context client IP, got %q", got) + } +} + +func newClientIPRequest(remote, forwarded string) *http.Request { + request := httptest.NewRequest(http.MethodGet, "http://example.test/", nil) + request.RemoteAddr = remote + if forwarded != "" { + request.Header.Set("X-Forwarded-For", forwarded) + } + return request +} diff --git a/middleware/middleware.go b/middleware/middleware.go index 80b983e..ed5de5e 100644 --- a/middleware/middleware.go +++ b/middleware/middleware.go @@ -2,7 +2,6 @@ package middleware import ( "fmt" - "net/http" "time" apierrors "git.yuntuo.hk/k2safe/go-omni/errors" @@ -64,19 +63,9 @@ func AccessLog() httpx.Middleware { "route", httpx.RoutePattern(ctx.Context()), "status", status, "durationMs", time.Since(started).Seconds()*1000, - "clientIp", clientIP(ctx.Request()), + "clientIp", ctx.ClientIP(), ) return err } } } - -func clientIP(request *http.Request) string { - if request == nil { - return "" - } - if value := request.Header.Get("X-Forwarded-For"); value != "" { - return value - } - return request.RemoteAddr -} diff --git a/middleware/middleware_test.go b/middleware/middleware_test.go index 5be9d77..da43b75 100644 --- a/middleware/middleware_test.go +++ b/middleware/middleware_test.go @@ -55,3 +55,34 @@ func TestAccessLogReportsCommittedStatus(t *testing.T) { }) } } + +func TestAccessLogUsesVerifiedClientIP(t *testing.T) { + loggerOutput := bytes.Buffer{} + logger, err := logx.New(logx.Config{Format: "json", Writer: &loggerOutput}) + if err != nil { + t.Fatal(err) + } + logx.SetDefault(logger) + defer logx.SetDefault(nil) + resolver, err := httpx.NewClientIPResolver("127.0.0.1") + if err != nil { + t.Fatal(err) + } + request := httptest.NewRequest("GET", "/health", nil) + request.RemoteAddr = "127.0.0.1:443" + request.Header.Set("X-Forwarded-For", "198.51.100.7") + request = request.WithContext(httpx.WithClientIPResolver(request.Context(), resolver)) + ctx := httpx.NewContext(request, httptest.NewRecorder(), nil) + if err := middleware.AccessLog()(func(c *httpx.Context) error { return c.OK("ok") })(ctx); err != nil { + t.Fatal(err) + } + var entry struct { + ClientIP string `json:"clientIp"` + } + if err := json.Unmarshal(loggerOutput.Bytes(), &entry); err != nil { + t.Fatal(err) + } + if entry.ClientIP != "198.51.100.7" { + t.Fatalf("expected verified client IP, got %q", entry.ClientIP) + } +} diff --git a/omni.go b/omni.go index a05844f..8f50ae1 100644 --- a/omni.go +++ b/omni.go @@ -41,6 +41,8 @@ func WithRequestIDHeader(header string) Option { return config.WithRequestIDHead func WithLogger(enabled bool) Option { return config.WithLogger(enabled) } +func WithTrustedProxies(proxies ...string) Option { return config.WithTrustedProxies(proxies...) } + func WithConfig(settings Config) Option { return config.WithConfig(settings) } func WithLogConfig(settings LogConfig) Option { return config.WithLogConfig(settings) } diff --git a/transport/README.md b/transport/README.md index ba8a04b..ac88ce0 100644 --- a/transport/README.md +++ b/transport/README.md @@ -16,7 +16,7 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码 `transport/native` 是第一版自研 HTTP/1.1 服务端：它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭，再把请求交给现有 `http.Handler`。请求体支持 `Content-Length` 和 chunked，并统一受 `MaxBodyBytes` 限制；其他传输编码会被拒绝。它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接，因此不会影响现有 Handler / Service。 -native 提供 `server.Stats()` 读取已接受连接数、当前连接数和当前请求数，适合和 `metrics.Metrics` 的业务请求指标一起采集；读取不会阻塞连接生命周期。 +native 提供 `server.Stats()` 读取已接受连接数、连接拒绝数、当前连接数和当前请求数，适合和 `metrics.Metrics` 的业务请求指标一起采集；读取不会阻塞连接生命周期。`RejectedConnections` 统计连接闸门拒绝，`RejectedRequests` 统计并发请求闸门返回 503。 业务 Handler 需要流式返回时可以断言 `http.Flusher` 并调用 `Flush()`；HTTP/1.1 会自动使用 chunked 响应，普通未 Flush 的响应仍自动生成 `Content-Length`。HTTP/1.0 没有 chunked framing，会退回到写完后关闭连接。 @@ -24,7 +24,9 @@ Handler 运行超过约 5 ms 时，native 启动一次阻塞读取监测；读 如果客户端提前发送了下一条请求，监测器会保留管线数据并退出，避免预读不断占用内存。这种情况下，断开取消会延后到写失败或 Handler 返回；业务仍应为下游调用设置自己的超时。 -服务端不是把连接交给一个全局串行 worker，而是每条连接独立处理请求；Keep-Alive 负责复用客户端连接，`Server` 内部的 `sync.Pool` 负责复用小型 Reader、响应对象和连接状态。Reader 默认只有 4 KiB，需要处理大请求头时按需增长。需要保护内存上限时，可以设置 `MaxConnections` 和 `MaxConcurrentRequests`；超出连接上限的连接会被关闭，超出请求并发上限的请求返回 503，不会无限排队。 +服务端不是把连接交给一个全局串行 worker，而是每条连接独立处理请求；Keep-Alive 负责复用客户端连接，`Server` 内部的 `sync.Pool` 负责复用小型 Reader、响应对象和连接状态。Reader 默认只有 4 KiB，需要处理大请求头时按需增长。需要保护内存上限时，可以设置 `MaxConnections` 和 `MaxConcurrentRequests`；超出连接上限的连接会被关闭，超出请求并发上限的请求返回 503，不会无限排队，拒绝数可从 `Stats()` 采集。 + +访问日志默认只记录 TCP socket 对端地址。部署在 Nginx、Cloudflare 等反向代理后面时，通过 `omni.WithTrustedProxies("127.0.0.1/32")` 或 `GO_OMNI_TRUSTED_PROXIES` 配置可信代理网段，框架才会从 `X-Forwarded-For` 右向左校验并取得真实客户端地址；未配置时转发头不会被信任。 这个设计借鉴了 Nginx 的资源边界，但没有照搬它的进程模型：Nginx 用少量 worker 进程和事件机制管理大量连接，并用 `worker_connections` 限制每个 worker 的同时连接数；Go runtime 自己已经用 epoll / kqueue 等网络轮询器唤醒阻塞的 goroutine，因此 native 采用“accept loop + 每连接轻量 goroutine + 请求并发闸门”的等价实现。这样业务 Handler 可以继续执行同步代码，慢请求不会阻塞其他连接；只有在真实压测证明 goroutine 调度或协议状态机成为瓶颈时，才需要继续下沉为手写事件循环。 diff --git a/transport/native/server.go b/transport/native/server.go index 8096c09..08a5d9f 100644 --- a/transport/native/server.go +++ b/transport/native/server.go @@ -53,27 +53,32 @@ type Server struct { // when the listener and TLS configuration are assembled by the caller. TLSConfig *tls.Config - mu sync.Mutex - handler http.Handler - listener net.Listener - conns map[*connection]struct{} - closing bool - cancel context.CancelFunc - wg sync.WaitGroup - connSlots chan struct{} - requestSlots chan struct{} - accepted atomic.Uint64 - activeConns atomic.Int64 - activeRequests atomic.Int64 - readerPool sync.Pool - writerPool sync.Pool - connectionPool sync.Pool + mu sync.Mutex + handler http.Handler + listener net.Listener + conns map[*connection]struct{} + closing bool + cancel context.CancelFunc + wg sync.WaitGroup + connSlots chan struct{} + requestSlots chan struct{} + accepted atomic.Uint64 + rejectedConns atomic.Uint64 + rejectedRequests atomic.Uint64 + activeConns atomic.Int64 + activeRequests atomic.Int64 + readerPool sync.Pool + writerPool sync.Pool + connectionPool sync.Pool } // Stats is a point-in-time view of the native server lifecycle. Counters are // monotonic unless their name starts with Active. type Stats struct { AcceptedConnections uint64 `json:"acceptedConnections"` + // Rejections count admission decisions, even if writing a 503 fails. + RejectedConnections uint64 `json:"rejectedConnections"` + RejectedRequests uint64 `json:"rejectedRequests"` ActiveConnections int64 `json:"activeConnections"` ActiveRequests int64 `json:"activeRequests"` } @@ -86,6 +91,8 @@ func (s *Server) Stats() Stats { } return Stats{ AcceptedConnections: s.accepted.Load(), + RejectedConnections: s.rejectedConns.Load(), + RejectedRequests: s.rejectedRequests.Load(), ActiveConnections: s.activeConns.Load(), ActiveRequests: s.activeRequests.Load(), } @@ -252,6 +259,7 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error { select { case slots <- struct{}{}: default: + s.rejectedConns.Add(1) _ = conn.Close() continue } @@ -424,6 +432,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { s.activeRequests.Add(-1) return default: + s.rejectedRequests.Add(1) writer.Header().Set("Content-Type", "text/plain; charset=utf-8") writer.WriteHeader(http.StatusServiceUnavailable) writer.body.WriteString("server busy\n") diff --git a/transport/native/server_test.go b/transport/native/server_test.go index b5518e4..177a90a 100644 --- a/transport/native/server_test.go +++ b/transport/native/server_test.go @@ -711,6 +711,9 @@ func TestServerConcurrentRequestLimit(t *testing.T) { if response.StatusCode != http.StatusServiceUnavailable { t.Fatalf("expected 503 at concurrency limit, got %d", response.StatusCode) } + if stats := server.Stats(); stats.RejectedRequests != 1 { + t.Fatalf("expected one rejected request, got %+v", stats) + } close(release) response, err = http.ReadResponse(bufio.NewReader(first), &http.Request{Method: http.MethodGet}) if err != nil { @@ -842,6 +845,9 @@ func TestServerConnectionLimitReleasesAfterRequest(t *testing.T) { t.Fatal("connection over MaxConnections was not rejected") } second.Close() + if stats := server.Stats(); stats.RejectedConnections < 1 { + t.Fatalf("expected rejected connection counter, got %+v", stats) + } close(release) firstResponse, err := http.ReadResponse(bufio.NewReader(first), &http.Request{Method: http.MethodGet})