feat: add trusted proxy identity and overload stats
This commit is contained in:
@@ -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 调度或协议状态机成为瓶颈时,才需要继续下沉为手写事件循环。
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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})
|
||||
|
||||
Reference in New Issue
Block a user