diff --git a/CHANGELOG.md b/CHANGELOG.md index b284290..1d51383 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- 新增独立 `transport/http2` 实验 Runner：支持 h2c/TLS h2、连接上限、全局请求并发闸门和生命周期统计；不改变 native HTTP/1.1 默认路径。 - native TLS 明确限制 ALPN 为 `http/1.1`：缺省自动宣告该协议，误配置 `h2` 等未实现协议时在启动阶段报错。 - native `Server.Stats()` 新增连接和请求过载拒绝计数；新增可信代理配置，访问日志只在 socket 对端属于配置网段时解析 `X-Forwarded-For`。 - 新增 transport-neutral `metrics` 包，支持请求数、错误数、in-flight、累计耗时和状态分布，并可直接注册 `/metrics` 接口。 diff --git a/README.md b/README.md index 6ee8406..7aa2b16 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,7 @@ go-omni/ ├── database/ # database/sql 生命周期和事务 ├── transport/nethttp/ # 标准库网络适配器 ├── transport/native/ # 自研 HTTP/1.1 服务端预览 +├── transport/http2/ # opt-in HTTP/2 实验适配器 ├── examples/ # 示例项目 └── cmd/ # go-omni 项目脚手架 ``` @@ -126,6 +127,8 @@ if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil { native TLS 当前只支持并宣告 ALPN `http/1.1`；配置 `h2` 会在启动时快速失败。需要 HTTP/2 时，可由 Nginx/Cloudflare 终止后回源 HTTP/1.1，或使用默认的 `transport/nethttp` 适配器。 +`transport/http2` 是独立的 opt-in 实验适配器，使用 `x/net/http2` 处理帧和流控，同时保留 GoOmni 的连接上限、全局请求并发闸门和生命周期边界。它不会改变 native 默认路径；实验入口、限制和验收结果见 [HTTP/2 进入条件](transport/native/HTTP2.md)。 + 自研网络层还提供 `server.Stats()`，可读取 accepted、rejected、active connections 和 active requests，与上面的业务请求 metrics 组合后能区分连接堆积、过载拒绝和 Handler 慢请求。`rejectedConnections` 表示连接闸门拒绝数，`rejectedRequests` 表示并发请求闸门返回 503 的次数。 业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库： diff --git a/ROADMAP.md b/ROADMAP.md index ead436f..2253bbe 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -50,6 +50,7 @@ - 已验证 `MaxConcurrentRequests` 满载返回 503，活动请求结束后槽位释放，后续请求可以恢复处理。 - native `Stats()` 已记录连接和请求过载拒绝数；访问日志已支持可信代理网段配置，默认拒绝伪造的 `X-Forwarded-For`。 - native TLS 已明确只宣告 HTTP/1.1 ALPN；HTTP/2 暂由边缘代理终止或使用 `transport/nethttp`，已整理独立协议设计、流控和压测门槛，见 [native HTTP/2 进入条件](transport/native/HTTP2.md)，完成后再接入 native。 +- 已加入独立 `transport/http2` 实验 Runner，验证 h2c/TLS h2、连接上限、全局请求并发拒绝和恢复；它使用 `x/net/http2` 作为临时帧/流控实现，尚未达到 native 生产协议的 GOAWAY、流控和长测门槛。 - 已用 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/go.mod b/go.mod index fae823a..44f6ab4 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.22 require ( github.com/go-playground/validator/v10 v10.20.0 github.com/go-sql-driver/mysql v1.9.3 + golang.org/x/net v0.25.0 ) require ( @@ -15,7 +16,6 @@ require ( github.com/leodido/go-urn v1.4.0 // indirect github.com/stretchr/testify v1.9.0 // indirect golang.org/x/crypto v0.23.0 // indirect - golang.org/x/net v0.25.0 // indirect golang.org/x/sys v0.20.0 // indirect golang.org/x/text v0.15.0 // indirect ) diff --git a/transport/README.md b/transport/README.md index 3374793..a14a590 100644 --- a/transport/README.md +++ b/transport/README.md @@ -14,6 +14,8 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码 `transport/nethttp` 是当前内置的标准库路由实现。它提供一套可移植的路径、方法和错误回退规则；自研网络层可以直接实现同一个 `httpx.Router` 契约。 +`transport/http2` 是独立的 opt-in 实验 Runner，负责连接接入、TLS/ALPN、连接上限和全局请求背压，底层帧解析与 stream 流控暂由 `golang.org/x/net/http2` 提供。它不替换 `transport/native`，也不改变默认启动器；只有通过 `App.RunWith(ctx, http2.New(nil), address)` 显式启用才会运行。 + `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` 的业务请求指标一起采集；读取不会阻塞连接生命周期。`RejectedConnections` 统计连接闸门拒绝，`RejectedRequests` 统计并发请求闸门返回 503。 diff --git a/transport/http2/server.go b/transport/http2/server.go new file mode 100644 index 0000000..6f2a3cd --- /dev/null +++ b/transport/http2/server.go @@ -0,0 +1,383 @@ +// Package http2 provides an opt-in HTTP/2 transport experiment. +// +// It owns the listener, TLS/ALPN setup, connection admission and global +// request backpressure while using x/net/http2 for frame parsing and stream +// flow control. It is deliberately separate from transport/native, whose +// production preview remains HTTP/1.1. +package http2 + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "net/http" + "sync" + "sync/atomic" + "time" + + xhttp2 "golang.org/x/net/http2" +) + +const ( + defaultShutdownTimeout = 10 * time.Second + defaultTLSHandshake = 10 * time.Second + http2ALPNProtocol = "h2" +) + +// Server is an opt-in HTTP/2 transport. Run uses cleartext h2c for local +// experiments; RunTLS or ServeTLS should be used for deployed TLS traffic. +// HTTP/2 stream limits and global request admission are independent: a client +// may open many streams, but only admitted handlers run concurrently. +type Server struct { + MaxConnections int + MaxConcurrentRequests int + MaxConcurrentStreams uint32 + IdleTimeout time.Duration + TLSHandshakeTimeout time.Duration + ShutdownTimeout time.Duration + TLSConfig *tls.Config + + mu sync.Mutex + handler http.Handler + listener net.Listener + closing bool + cancel context.CancelFunc + conns map[net.Conn]struct{} + connSlots chan struct{} + requestSlots chan struct{} + accepted atomic.Uint64 + rejectedConns atomic.Uint64 + rejectedReqs atomic.Uint64 + activeConns atomic.Int64 + activeRequests atomic.Int64 + wg sync.WaitGroup +} + +// Stats is a point-in-time view of HTTP/2 transport admission and lifecycle. +// Accepted and rejected counters are monotonic; active counters are gauges. +type Stats struct { + AcceptedConnections uint64 `json:"acceptedConnections"` + RejectedConnections uint64 `json:"rejectedConnections"` + RejectedRequests uint64 `json:"rejectedRequests"` + ActiveConnections int64 `json:"activeConnections"` + ActiveRequests int64 `json:"activeRequests"` +} + +// New creates an HTTP/2 transport with the supplied handler. The handler may +// also be supplied by App.RunWith through the Runner interface. +func New(handler http.Handler) *Server { + return &Server{handler: handler, conns: make(map[net.Conn]struct{})} +} + +// Stats returns a lock-free snapshot suitable for a metrics handler. +func (s *Server) Stats() Stats { + if s == nil { + return Stats{} + } + return Stats{ + AcceptedConnections: s.accepted.Load(), + RejectedConnections: s.rejectedConns.Load(), + RejectedRequests: s.rejectedReqs.Load(), + ActiveConnections: s.activeConns.Load(), + ActiveRequests: s.activeRequests.Load(), + } +} + +// Run implements app.Runner. Without TLSConfig it starts an h2c listener for +// local testing. Use RunTLS or TLSConfig for real HTTPS traffic. +func (s *Server) Run(ctx context.Context, handler http.Handler, address string) error { + if handler != nil { + s.handler = handler + } + if s.handler == nil { + return errors.New("go-omni/http2: nil handler") + } + if address == "" { + address = ":8080" + } + listener, err := net.Listen("tcp", address) + if err != nil { + return err + } + if s.TLSConfig != nil { + return s.ServeTLS(ctx, listener, s.TLSConfig) + } + return s.Serve(ctx, listener) +} + +// RunTLS loads a certificate pair and serves HTTP/2 over TLS with ALPN h2. +func (s *Server) RunTLS(ctx context.Context, handler http.Handler, address, certFile, keyFile string) error { + if handler != nil { + s.handler = handler + } + if s.handler == nil { + return errors.New("go-omni/http2: nil handler") + } + if address == "" { + address = ":8443" + } + certificate, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return err + } + listener, err := net.Listen("tcp", address) + if err != nil { + return err + } + return s.ServeTLS(ctx, listener, &tls.Config{Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12}) +} + +// ServeTLS wraps listener with TLS and requires ALPN h2. A caller-provided +// config is cloned, so the original value is never mutated. +func (s *Server) ServeTLS(ctx context.Context, listener net.Listener, config *tls.Config) error { + if listener == nil { + return errors.New("go-omni/http2: nil listener") + } + if config == nil { + return errors.New("go-omni/http2: nil TLS config") + } + prepared, err := prepareTLSConfig(config) + if err != nil { + _ = listener.Close() + return err + } + return s.Serve(ctx, tls.NewListener(listener, prepared)) +} + +func prepareTLSConfig(config *tls.Config) (*tls.Config, error) { + clone := config.Clone() + if len(clone.NextProtos) > 0 { + for _, protocol := range clone.NextProtos { + if protocol != http2ALPNProtocol { + return nil, fmt.Errorf("go-omni/http2: unsupported TLS ALPN protocol %q; only %q is supported", protocol, http2ALPNProtocol) + } + } + } + clone.NextProtos = []string{http2ALPNProtocol} + return clone, nil +} + +// Serve accepts h2c connections from listener until ctx is canceled. +func (s *Server) Serve(ctx context.Context, listener net.Listener) error { + if listener == nil { + return errors.New("go-omni/http2: nil listener") + } + if s.handler == nil { + return errors.New("go-omni/http2: nil handler") + } + if ctx == nil { + ctx = context.Background() + } + + s.mu.Lock() + if s.cancel != nil { + s.mu.Unlock() + return errors.New("go-omni/http2: server is already running") + } + baseCtx, cancel := context.WithCancel(ctx) + s.cancel = cancel + s.listener = listener + s.closing = false + if s.conns == nil { + s.conns = make(map[net.Conn]struct{}) + } + if s.MaxConnections > 0 { + s.connSlots = make(chan struct{}, s.MaxConnections) + } else { + s.connSlots = nil + } + if s.MaxConcurrentRequests > 0 { + s.requestSlots = make(chan struct{}, s.MaxConcurrentRequests) + } else { + s.requestSlots = nil + } + s.mu.Unlock() + + gate := &requestGate{server: s, next: s.handler, slots: s.requestSlots} + protocolServer := &xhttp2.Server{ + MaxConcurrentStreams: s.MaxConcurrentStreams, + IdleTimeout: s.IdleTimeout, + } + shutdownDone := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), s.shutdownTimeout()) + _ = s.Shutdown(shutdownCtx) + cancel() + case <-shutdownDone: + } + }() + + defer func() { + close(shutdownDone) + cancel() + s.mu.Lock() + s.listener = nil + s.cancel = nil + s.mu.Unlock() + }() + + for { + conn, err := listener.Accept() + if err != nil { + if ctx.Err() != nil || s.isClosing() { + break + } + return err + } + s.accepted.Add(1) + if slots := s.connSlots; slots != nil { + select { + case slots <- struct{}{}: + default: + s.rejectedConns.Add(1) + _ = conn.Close() + continue + } + } + s.mu.Lock() + if s.closing { + s.mu.Unlock() + if s.connSlots != nil { + <-s.connSlots + } + _ = conn.Close() + continue + } + s.conns[conn] = struct{}{} + s.activeConns.Add(1) + s.wg.Add(1) + s.mu.Unlock() + go s.serveConn(baseCtx, protocolServer, gate, conn) + } + return nil +} + +func (s *Server) shutdownTimeout() time.Duration { + if s.ShutdownTimeout > 0 { + return s.ShutdownTimeout + } + return defaultShutdownTimeout +} + +func (s *Server) serveConn(ctx context.Context, protocolServer *xhttp2.Server, handler http.Handler, conn net.Conn) { + defer s.wg.Done() + defer func() { + s.mu.Lock() + delete(s.conns, conn) + s.activeConns.Add(-1) + s.mu.Unlock() + if s.connSlots != nil { + <-s.connSlots + } + _ = conn.Close() + }() + if tlsConn, ok := conn.(*tls.Conn); ok { + _ = conn.SetDeadline(time.Now().Add(s.tlsHandshakeTimeout())) + if err := tlsConn.HandshakeContext(ctx); err != nil { + return + } + _ = conn.SetDeadline(time.Time{}) + } + protocolServer.ServeConn(conn, &xhttp2.ServeConnOpts{ + Context: ctx, + Handler: handler, + BaseConfig: &http.Server{ + IdleTimeout: s.IdleTimeout, + }, + }) +} + +func (s *Server) tlsHandshakeTimeout() time.Duration { + if s.TLSHandshakeTimeout > 0 { + return s.TLSHandshakeTimeout + } + return defaultTLSHandshake +} + +// Shutdown stops accepting connections and waits for active HTTP/2 sessions. +// If the context expires, remaining connections are closed so handlers can +// observe cancellation through their request contexts. +func (s *Server) Shutdown(ctx context.Context) error { + if s == nil { + return nil + } + if ctx == nil { + timeout := s.ShutdownTimeout + if timeout <= 0 { + timeout = defaultShutdownTimeout + } + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(context.Background(), timeout) + defer cancel() + } + s.mu.Lock() + if s.cancel == nil { + s.mu.Unlock() + return nil + } + s.closing = true + listener := s.listener + cancel := s.cancel + connections := make([]net.Conn, 0, len(s.conns)) + for conn := range s.conns { + connections = append(connections, conn) + } + s.mu.Unlock() + if listener != nil { + _ = listener.Close() + } + cancel() + + done := make(chan struct{}) + go func() { + s.wg.Wait() + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + for _, conn := range connections { + _ = conn.Close() + } + return ctx.Err() + } +} + +func (s *Server) isClosing() bool { + s.mu.Lock() + closing := s.closing + s.mu.Unlock() + return closing +} + +type requestGate struct { + server *Server + next http.Handler + slots chan struct{} +} + +func (g *requestGate) ServeHTTP(writer http.ResponseWriter, request *http.Request) { + if g.slots != nil { + select { + case g.slots <- struct{}{}: + g.server.activeRequests.Add(1) + defer func() { + <-g.slots + g.server.activeRequests.Add(-1) + }() + default: + g.server.rejectedReqs.Add(1) + http.Error(writer, "server busy\n", http.StatusServiceUnavailable) + return + } + } else { + g.server.activeRequests.Add(1) + defer g.server.activeRequests.Add(-1) + } + g.next.ServeHTTP(writer, request) +} diff --git a/transport/http2/server_test.go b/transport/http2/server_test.go new file mode 100644 index 0000000..c335b2c --- /dev/null +++ b/transport/http2/server_test.go @@ -0,0 +1,230 @@ +package http2 + +import ( + "context" + "crypto/rand" + "crypto/rsa" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "errors" + "math/big" + "net" + "net/http" + "sync" + "testing" + "time" + + xhttp2 "golang.org/x/net/http2" +) + +func TestServerHandlesH2CRequest(t *testing.T) { + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.ProtoMajor != 2 { + http.Error(writer, "not HTTP/2", http.StatusInternalServerError) + return + } + _, _ = writer.Write([]byte("ok")) + })) + address, stop := startServer(t, server) + defer stop() + + transport := &xhttp2.Transport{ + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, address string, _ *tls.Config) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, address) + }, + } + client := &http.Client{Transport: transport, Timeout: 2 * time.Second} + defer transport.CloseIdleConnections() + response, err := client.Get("http://" + address + "/health") + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + t.Fatalf("unexpected status: %d", response.StatusCode) + } + if stats := server.Stats(); stats.AcceptedConnections != 1 { + t.Fatalf("unexpected HTTP/2 stats: %+v", stats) + } +} + +func TestServerEnforcesGlobalRequestLimitAndRecovers(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path == "/hold" { + once.Do(func() { close(started) }) + <-release + } + _, _ = writer.Write([]byte("ok")) + })) + server.MaxConcurrentRequests = 1 + address, stop := startServer(t, server) + defer stop() + transport := &xhttp2.Transport{ + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, address string, _ *tls.Config) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, address) + }, + } + client := &http.Client{Transport: transport, Timeout: 3 * time.Second} + defer transport.CloseIdleConnections() + + firstDone := make(chan *http.Response, 1) + firstErr := make(chan error, 1) + go func() { + response, err := client.Get("http://" + address + "/hold") + if err != nil { + firstErr <- err + return + } + firstDone <- response + }() + select { + case <-started: + case err := <-firstErr: + t.Fatal(err) + case <-time.After(time.Second): + t.Fatal("first HTTP/2 stream did not start") + } + + response, err := client.Get("http://" + address + "/busy") + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("expected 503 at stream limit, got %d", response.StatusCode) + } + if stats := server.Stats(); stats.RejectedRequests != 1 { + t.Fatalf("expected one rejected stream, got %+v", stats) + } + + close(release) + select { + case response = <-firstDone: + response.Body.Close() + case err := <-firstErr: + t.Fatal(err) + case <-time.After(time.Second): + t.Fatal("held HTTP/2 stream did not finish") + } + response, err = client.Get("http://" + address + "/recovered") + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusOK { + t.Fatalf("request limit did not recover, got %d", response.StatusCode) + } +} + +func TestServerTLSUsesH2ALPN(t *testing.T) { + certificate, roots := testCertificate(t) + server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.TLS == nil || request.TLS.NegotiatedProtocol != "h2" { + http.Error(writer, "missing h2 ALPN", http.StatusInternalServerError) + return + } + writer.WriteHeader(http.StatusNoContent) + })) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { + done <- server.ServeTLS(context.Background(), listener, &tls.Config{ + Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12, + }) + }() + stop := func() { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := server.Shutdown(ctx); err != nil && !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("shutdown: %v", err) + } + if err := <-done; err != nil { + t.Errorf("serve: %v", err) + } + } + defer stop() + + transport := &xhttp2.Transport{TLSClientConfig: &tls.Config{ + RootCAs: roots, ServerName: "localhost", MinVersion: tls.VersionTLS12, + NextProtos: []string{"h2"}, + }} + client := &http.Client{Transport: transport, Timeout: 2 * time.Second} + defer transport.CloseIdleConnections() + response, err := client.Get("https://" + listener.Addr().String() + "/health") + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusNoContent { + t.Fatalf("unexpected TLS status: %d", response.StatusCode) + } +} + +func startServer(t *testing.T, server *Server) (string, func()) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- server.Serve(context.Background(), listener) }() + stop := func() { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := server.Shutdown(ctx); err != nil && !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("shutdown: %v", err) + } + if err := <-done; err != nil { + t.Errorf("serve: %v", err) + } + } + return listener.Addr().String(), stop +} + +func testCertificate(t *testing.T) (tls.Certificate, *x509.CertPool) { + t.Helper() + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatal(err) + } + serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 120)) + if err != nil { + t.Fatal(err) + } + template := &x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: "localhost"}, + DNSNames: []string{"localhost"}, + NotBefore: time.Now().Add(-time.Minute), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + leaf, err := x509.ParseCertificate(der) + if err != nil { + t.Fatal(err) + } + certificate := tls.Certificate{ + Certificate: [][]byte{der}, + PrivateKey: key, + } + certificate.Leaf = leaf + roots := x509.NewCertPool() + roots.AddCert(leaf) + return certificate, roots +}