wangbo
4b639df30a
官方 Gemini 平台配置的共享 HTTP Proxy 会拒绝香港出口的 CONNECT 请求，导致香港 Worker 的真实 VEO 与图片任务进入 upstream_submission_unknown。 新增仅按平台 UUID 生效的代理直连白名单，并只在香港 Worker 为官方 Gemini 平台启用；宁波与其他平台继续使用原代理策略。 验证：API 全量 go test、代理路由单测、bash -n、ShellCheck、cluster release helper、manual release 与差异检查均通过。
107 lines
3.2 KiB
Go
107 lines
3.2 KiB
Go
package runner
|
|
|
|
import (
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/netproxy"
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
|
)
|
|
|
|
type httpClientCache struct {
|
|
none *http.Client
|
|
global *http.Client
|
|
mu sync.Mutex
|
|
custom map[string]*http.Client
|
|
}
|
|
|
|
const providerHTTPClientTimeout = 10 * time.Minute
|
|
|
|
const (
|
|
providerHTTPMaxIdleConnections = 2048
|
|
providerHTTPMaxIdleConnectionsPerHost = 1024
|
|
)
|
|
|
|
func newHTTPClientCache() *httpClientCache {
|
|
return &httpClientCache{
|
|
none: newHTTPClient(nil),
|
|
global: newHTTPClient(http.ProxyFromEnvironment),
|
|
custom: map[string]*http.Client{},
|
|
}
|
|
}
|
|
|
|
func (s *Service) httpClientForCandidate(candidate store.RuntimeModelCandidate, simulated bool) (*http.Client, error) {
|
|
if simulated {
|
|
return s.httpClients.none, nil
|
|
}
|
|
// Some Worker sites have direct provider egress but are not authorized to
|
|
// use a platform's shared proxy. Keep this override explicitly scoped to
|
|
// platform UUIDs so one site's routing exception cannot bypass proxies for
|
|
// unrelated providers or platforms.
|
|
if platformIDListed(s.cfg.PlatformProxyBypassIDs, candidate.PlatformID) {
|
|
return s.httpClients.none, nil
|
|
}
|
|
config, err := netproxy.Normalize(netproxy.FromPlatformConfig(candidate.PlatformConfig))
|
|
if err != nil {
|
|
return nil, &clients.ClientError{Code: "invalid_proxy", Message: err.Error(), Retryable: false}
|
|
}
|
|
switch config.Mode {
|
|
case netproxy.ModeGlobal:
|
|
if strings.TrimSpace(s.cfg.GlobalHTTPProxy) != "" {
|
|
return s.httpClients.customClient(s.cfg.GlobalHTTPProxy)
|
|
}
|
|
return s.httpClients.global, nil
|
|
case netproxy.ModeCustom:
|
|
return s.httpClients.customClient(config.HTTPProxy)
|
|
default:
|
|
return s.httpClients.none, nil
|
|
}
|
|
}
|
|
|
|
func platformIDListed(raw string, platformID string) bool {
|
|
platformID = strings.TrimSpace(platformID)
|
|
if platformID == "" {
|
|
return false
|
|
}
|
|
for value := range strings.SplitSeq(raw, ",") {
|
|
if strings.TrimSpace(value) == platformID {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (c *httpClientCache) customClient(rawProxy string) (*http.Client, error) {
|
|
normalized, proxyURL, err := netproxy.ParseHTTPProxy(rawProxy)
|
|
if err != nil {
|
|
return nil, &clients.ClientError{Code: "invalid_proxy", Message: err.Error(), Retryable: false}
|
|
}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
if client := c.custom[normalized]; client != nil {
|
|
return client, nil
|
|
}
|
|
client := newHTTPClient(http.ProxyURL(proxyURL))
|
|
c.custom[normalized] = client
|
|
return client, nil
|
|
}
|
|
|
|
func newHTTPClient(proxy func(*http.Request) (*url.URL, error)) *http.Client {
|
|
transport := http.DefaultTransport.(*http.Transport).Clone()
|
|
transport.Proxy = proxy
|
|
// Media workers may keep hundreds of slow upstream tasks in flight and
|
|
// poll the same provider repeatedly. The standard library only retains two
|
|
// idle connections per host, which turns a high-concurrency polling load
|
|
// into avoidable TCP/TLS handshakes.
|
|
transport.MaxIdleConns = providerHTTPMaxIdleConnections
|
|
transport.MaxIdleConnsPerHost = providerHTTPMaxIdleConnectionsPerHost
|
|
return &http.Client{
|
|
Timeout: providerHTTPClientTimeout,
|
|
Transport: transport,
|
|
}
|
|
}
|