From bda9cc03dfaeb6b3bbaf1ad38c8d68ddadf0d7eb Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 7 Aug 2026 15:27:17 +0800 Subject: [PATCH] feat: configure gateway transport pools --- deploy/config/local.yaml | 5 ++ docs/configuration/reference.md | 35 ++++++++++++ internal/config/config.go | 15 ++++++ internal/config/config_test.go | 45 ++++++++++++++++ internal/config/validate.go | 56 ++++++++++++++++++++ internal/gateway/bootstrap/bootstrap.go | 17 +++++- internal/gateway/bootstrap/bootstrap_test.go | 28 ++++++++++ internal/gateway/transport/transport.go | 2 + internal/gateway/transport/transport_test.go | 26 +++++++++ 9 files changed, 228 insertions(+), 1 deletion(-) diff --git a/deploy/config/local.yaml b/deploy/config/local.yaml index fb30f37..d44a339 100644 --- a/deploy/config/local.yaml +++ b/deploy/config/local.yaml @@ -16,6 +16,11 @@ gateway: limits: maxConcurrentConnections: 20000 requestsPerMinutePerClient: 60000 + transport: + maxIdleConns: 20000 + maxIdleConnsPerHost: 32 + maxConnsPerHost: 32 + tunnelBufferBytes: 32768 retry: maxAttempts: 2 retryMethods: [GET, HEAD] diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 8a6b2c1..7f0e4f8 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -333,6 +333,16 @@ gateway: auth: {mode: none} limits: maxConcurrentConnections: 50000 + transport: + dialTimeout: 10s + handshakeTimeout: 15s + responseHeaderTimeout: 30s + idleConnTimeout: 90s + maxIdleConns: 20000 + maxIdleConnsPerHost: 32 + maxConnsPerHost: 32 + tunnelBufferBytes: 32768 + tunnelIdleTimeout: 5m retry: maxAttempts: 2 retryMethods: [GET, HEAD] @@ -358,6 +368,31 @@ gateway: - 保留地址、CGNAT 与云元数据端点始终拒绝，不能通过私网/链路本地开关放行。 - `maxConcurrentConnections` 是入口准入上限，不是 Proxy 容量上限。 +### 5.1 Gateway 本地传输层 + +**gateway.transport** 只配置每个 Gateway Worker 的本地 HTTP Transport 和 CONNECT +隧道 I/O，不读取或写入 Redis、PostgreSQL、Provider，也不参与热路径分配。修改 +这些字段需要滚动重启对应 Gateway Worker 后生效。 + +- **dialTimeout**、**handshakeTimeout**、**responseHeaderTimeout**、 + **idleConnTimeout**：分别限制拨号、上游 TLS/CONNECT 握手、HTTP 响应头和空闲 + HTTP 连接的时长。 +- **maxIdleConns**：单个本地 HTTP Transport 的总空闲连接上限。Gateway 分别为代理 + 转发和 direct fallback 建立 Transport，最坏情况下两者都可能保有空闲连接。 +- **maxIdleConnsPerHost**：单个上游 Host 的空闲 HTTP 连接上限。 +- **maxConnsPerHost**：单个上游 Host 的 HTTP 活跃加空闲连接上限。它不替代 + **limits.maxConcurrentConnections**，也不限制已经建立的 CONNECT 隧道。 +- **tunnelBufferBytes**：每个活跃 CONNECT 隧道的每个复制方向使用一个缓冲区，内存 + 预算至少按 活跃隧道数 * 2 * tunnelBufferBytes 计算。 +- **tunnelIdleTimeout**：CONNECT 隧道双向没有数据活动时的最长存活时间。 + +所有字段均为可选项；数值 0 表示沿用进程默认值：拨号 10s、握手 15s、 +响应头 30s、空闲 HTTP 连接 90s、总空闲连接 1024、每 Host 空闲连接 +64、隧道缓冲 32768 字节、隧道空闲 5m。计数和时长不得为负数； +maxIdleConnsPerHost 不得大于显式配置的 maxIdleConns。示例中的 20000/32 +是高并发 HTTP 转发的起点，不是每秒请求量的换算公式，应同时按可用文件描述符、 +上游限额和观测到的复用率调节。 + ## 6. Distribution ```yaml diff --git a/internal/config/config.go b/internal/config/config.go index e976d93..d0449fd 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -66,6 +66,7 @@ type Listener struct { Limits Limits `yaml:"limits"` Retry Retry `yaml:"retry"` StickySession StickySession `yaml:"stickySession"` + Transport GatewayTransport `yaml:"transport"` DestinationPolicy DestinationPolicy `yaml:"destinationPolicy"` } @@ -127,6 +128,20 @@ type StickySession struct { MaxEntries int `yaml:"maxEntries"` } +// GatewayTransport configures the local Gateway process connection pools and +// tunnel I/O. It intentionally contains no remote state or provider settings. +type GatewayTransport struct { + DialTimeout Duration `yaml:"dialTimeout"` + HandshakeTimeout Duration `yaml:"handshakeTimeout"` + ResponseHeaderTimeout Duration `yaml:"responseHeaderTimeout"` + IdleConnTimeout Duration `yaml:"idleConnTimeout"` + MaxIdleConns int `yaml:"maxIdleConns"` + MaxIdleConnsPerHost int `yaml:"maxIdleConnsPerHost"` + MaxConnsPerHost int `yaml:"maxConnsPerHost"` + TunnelBufferBytes int `yaml:"tunnelBufferBytes"` + TunnelIdleTimeout Duration `yaml:"tunnelIdleTimeout"` +} + type DestinationPolicy struct { DenyPrivateNetworks *bool `yaml:"denyPrivateNetworks"` DenyLoopback *bool `yaml:"denyLoopback"` diff --git a/internal/config/config_test.go b/internal/config/config_test.go index db1573e..df50d24 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -469,6 +469,51 @@ func TestValidateGatewayStickySession(t *testing.T) { } } +func TestValidateGatewayTransport(t *testing.T) { + t.Parallel() + + cfg := mustLoadValidConfig(t) + cfg.Gateway.Transport = GatewayTransport{ + DialTimeout: Duration(time.Second), + HandshakeTimeout: Duration(2 * time.Second), + ResponseHeaderTimeout: Duration(3 * time.Second), + IdleConnTimeout: Duration(time.Minute), + MaxIdleConns: 500, + MaxIdleConnsPerHost: 20, + MaxConnsPerHost: 10, + TunnelBufferBytes: 32 << 10, + TunnelIdleTimeout: Duration(5 * time.Minute), + } + if err := Validate(cfg); err != nil { + t.Fatalf("Validate(valid gateway transport) error = %v", err) + } + + cfg.Gateway.Transport.DialTimeout = Duration(-time.Second) + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "transport.dialTimeout") { + t.Fatalf("Validate(negative dial timeout) error = %v", err) + } + + cfg.Gateway.Transport.DialTimeout = Duration(time.Second) + cfg.Gateway.Transport.MaxConnsPerHost = -1 + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "transport.maxConnsPerHost") { + t.Fatalf("Validate(negative max connections) error = %v", err) + } + + cfg.Gateway.Transport.MaxConnsPerHost = 10 + cfg.Gateway.Transport.MaxIdleConns = 19 + cfg.Gateway.Transport.MaxIdleConnsPerHost = 20 + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "maxIdleConnsPerHost") { + t.Fatalf("Validate(inconsistent idle pool) error = %v", err) + } + + cfg.Gateway.Transport.MaxIdleConns = 500 + cfg.Gateway.Transport.MaxIdleConnsPerHost = 20 + cfg.Distribution.Transport = GatewayTransport{MaxConnsPerHost: 10} + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "distribution transport") { + t.Fatalf("Validate(distribution transport) error = %v", err) + } +} + func TestValidateMetricsListener(t *testing.T) { t.Parallel() cfg := mustLoadValidConfig(t) diff --git a/internal/config/validate.go b/internal/config/validate.go index 5ef8763..17d6587 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -113,6 +113,9 @@ func Validate(cfg *Config) error { if err := validateGatewayStickySession(cfg.Gateway); err != nil { return err } + if err := validateGatewayTransport(cfg.Gateway); err != nil { + return err + } } if cfg.Distribution.Enabled { if err := validateDistributionClientPolicies(cfg.Distribution.Auth, cfg.Upstreams); err != nil { @@ -121,6 +124,9 @@ func Validate(cfg *Config) error { if cfg.Distribution.StickySession.Enabled { return fmt.Errorf("validate distribution stickySession: only supported on gateway") } + if gatewayTransportConfigured(cfg.Distribution.Transport) { + return fmt.Errorf("validate distribution transport: only supported on gateway") + } clientIdentificationMode := cfg.Distribution.ClientIdentification.Mode if clientIdentificationMode == "" { clientIdentificationMode = "sourceIP" @@ -154,6 +160,9 @@ func Validate(cfg *Config) error { if cfg.Admin.Enabled && cfg.Admin.StickySession.Enabled { return fmt.Errorf("validate admin stickySession: only supported on gateway") } + if cfg.Admin.Enabled && gatewayTransportConfigured(cfg.Admin.Transport) { + return fmt.Errorf("validate admin transport: only supported on gateway") + } return nil } @@ -177,6 +186,53 @@ func validateGatewayStickySession(listener Listener) error { return nil } +func validateGatewayTransport(listener Listener) error { + item := listener.Transport + for _, timeout := range []struct { + name string + value Duration + }{ + {name: "dialTimeout", value: item.DialTimeout}, + {name: "handshakeTimeout", value: item.HandshakeTimeout}, + {name: "responseHeaderTimeout", value: item.ResponseHeaderTimeout}, + {name: "idleConnTimeout", value: item.IdleConnTimeout}, + {name: "tunnelIdleTimeout", value: item.TunnelIdleTimeout}, + } { + if timeout.value.Value() < 0 { + return fmt.Errorf("validate gateway transport.%s: must be non-negative", timeout.name) + } + } + for _, limit := range []struct { + name string + value int + }{ + {name: "maxIdleConns", value: item.MaxIdleConns}, + {name: "maxIdleConnsPerHost", value: item.MaxIdleConnsPerHost}, + {name: "maxConnsPerHost", value: item.MaxConnsPerHost}, + {name: "tunnelBufferBytes", value: item.TunnelBufferBytes}, + } { + if limit.value < 0 || limit.value > MaximumPoolSize { + return fmt.Errorf("validate gateway transport.%s: must be in [0, %d]", limit.name, MaximumPoolSize) + } + } + if item.MaxIdleConns > 0 && item.MaxIdleConnsPerHost > item.MaxIdleConns { + return fmt.Errorf("validate gateway transport.maxIdleConnsPerHost: must not exceed maxIdleConns") + } + return nil +} + +func gatewayTransportConfigured(item GatewayTransport) bool { + return item.DialTimeout != 0 || + item.HandshakeTimeout != 0 || + item.ResponseHeaderTimeout != 0 || + item.IdleConnTimeout != 0 || + item.MaxIdleConns != 0 || + item.MaxIdleConnsPerHost != 0 || + item.MaxConnsPerHost != 0 || + item.TunnelBufferBytes != 0 || + item.TunnelIdleTimeout != 0 +} + func validCanonicalHeaderName(value string) bool { if value == "" || http.CanonicalHeaderKey(value) != value { return false diff --git a/internal/gateway/bootstrap/bootstrap.go b/internal/gateway/bootstrap/bootstrap.go index 8669c1f..5b1e7be 100644 --- a/internal/gateway/bootstrap/bootstrap.go +++ b/internal/gateway/bootstrap/bootstrap.go @@ -156,7 +156,7 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio storeOptions.CapacityInvariantObserver = collector } store := snapshot.NewStoreWithOptions(options.ClusterID, options.WorkerID, storeOptions) - proxyTransport := transport.New(transport.Config{}, snapshotCredentialResolver{store: store}) + proxyTransport := transport.New(gatewayTransportConfig(configuration.Gateway), snapshotCredentialResolver{store: store}) outcomes, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{ Capacity: defaultOutcomeQueueCapacity, MaxBatch: min(defaultOutcomeBatchSize, configuration.ControlPlane.MaxRuntimeCounters), Metrics: outcomeMetrics, @@ -243,6 +243,21 @@ func newRuntime(ctx context.Context, configuration *config.Config, options Optio return &runtime{connection: connection, group: group}, nil } +func gatewayTransportConfig(listener config.Listener) transport.Config { + item := listener.Transport + return transport.Config{ + DialTimeout: item.DialTimeout.Value(), + HandshakeTimeout: item.HandshakeTimeout.Value(), + ResponseHeaderTimeout: item.ResponseHeaderTimeout.Value(), + IdleConnTimeout: item.IdleConnTimeout.Value(), + MaxIdleConns: item.MaxIdleConns, + MaxIdleConnsPerHost: item.MaxIdleConnsPerHost, + MaxConnsPerHost: item.MaxConnsPerHost, + TunnelBufferBytes: item.TunnelBufferBytes, + TunnelIdleTimeout: item.TunnelIdleTimeout.Value(), + } +} + type generatedRuntimeClient struct { client controlplanev1.WorkerControlPlaneClient } diff --git a/internal/gateway/bootstrap/bootstrap_test.go b/internal/gateway/bootstrap/bootstrap_test.go index bf34614..25eb2f6 100644 --- a/internal/gateway/bootstrap/bootstrap_test.go +++ b/internal/gateway/bootstrap/bootstrap_test.go @@ -145,6 +145,34 @@ func TestControlPlaneTransportRequiresDedicatedGatewayTLS(t *testing.T) { } } +func TestGatewayTransportConfigMapsListenerSettings(t *testing.T) { + t.Parallel() + + got := gatewayTransportConfig(config.Listener{Transport: config.GatewayTransport{ + DialTimeout: config.Duration(time.Second), + HandshakeTimeout: config.Duration(2 * time.Second), + ResponseHeaderTimeout: config.Duration(3 * time.Second), + IdleConnTimeout: config.Duration(time.Minute), + MaxIdleConns: 500, + MaxIdleConnsPerHost: 20, + MaxConnsPerHost: 10, + TunnelBufferBytes: 32 << 10, + TunnelIdleTimeout: config.Duration(5 * time.Minute), + }}) + + if got.DialTimeout != time.Second || + got.HandshakeTimeout != 2*time.Second || + got.ResponseHeaderTimeout != 3*time.Second || + got.IdleConnTimeout != time.Minute || + got.MaxIdleConns != 500 || + got.MaxIdleConnsPerHost != 20 || + got.MaxConnsPerHost != 10 || + got.TunnelBufferBytes != 32<<10 || + got.TunnelIdleTimeout != 5*time.Minute { + t.Fatalf("gatewayTransportConfig() = %+v", got) + } +} + func TestSnapshotReadinessRequiresCurrentSnapshot(t *testing.T) { t.Parallel() diff --git a/internal/gateway/transport/transport.go b/internal/gateway/transport/transport.go index 425c5e5..4d98c9a 100644 --- a/internal/gateway/transport/transport.go +++ b/internal/gateway/transport/transport.go @@ -40,6 +40,7 @@ type Config struct { IdleConnTimeout time.Duration MaxIdleConns int MaxIdleConnsPerHost int + MaxConnsPerHost int MaxErrorResponseBytes int64 MaxResponseHeaderBytes int64 TunnelBufferBytes int @@ -120,6 +121,7 @@ func newHTTPTransport(config Config, proxy func(*http.Request) (*url.URL, error) ForceAttemptHTTP2: true, MaxIdleConns: config.MaxIdleConns, MaxIdleConnsPerHost: config.MaxIdleConnsPerHost, + MaxConnsPerHost: config.MaxConnsPerHost, IdleConnTimeout: config.IdleConnTimeout, TLSHandshakeTimeout: config.HandshakeTimeout, ResponseHeaderTimeout: config.ResponseHeaderTimeout, diff --git a/internal/gateway/transport/transport_test.go b/internal/gateway/transport/transport_test.go index ad729f3..3bc751b 100644 --- a/internal/gateway/transport/transport_test.go +++ b/internal/gateway/transport/transport_test.go @@ -88,6 +88,32 @@ func TestRoundTripDirectForwardsWithoutProxyAuthorization(t *testing.T) { } } +func TestNewAppliesConfiguredConnectionPoolLimits(t *testing.T) { + t.Parallel() + + client := New(Config{ + MaxIdleConns: 256, + MaxIdleConnsPerHost: 24, + MaxConnsPerHost: 12, + }, nil) + t.Cleanup(client.CloseIdleConnections) + + for name, item := range map[string]*http.Transport{ + "proxy": client.client, + "direct": client.direct, + } { + if item.MaxIdleConns != 256 { + t.Fatalf("%s MaxIdleConns = %d, want 256", name, item.MaxIdleConns) + } + if item.MaxIdleConnsPerHost != 24 { + t.Fatalf("%s MaxIdleConnsPerHost = %d, want 24", name, item.MaxIdleConnsPerHost) + } + if item.MaxConnsPerHost != 12 { + t.Fatalf("%s MaxConnsPerHost = %d, want 12", name, item.MaxConnsPerHost) + } + } +} + func TestOpenDirectTunnelDialsTarget(t *testing.T) { t.Parallel()