From 67ecff77160a870b36c456fcce6255bfd813e482 Mon Sep 17 00:00:00 2001 From: mangoknight Date: Wed, 2 Sep 2026 13:56:09 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(proxy):=20=E5=87=BA=E5=8F=A3=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E6=B1=A0=EF=BC=8C=E9=87=8D=E8=AF=95=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E6=8D=A2=E5=87=BA=E5=8F=A3=E5=B9=B6=E6=8C=89=E5=BA=94=E7=94=A8?= =?UTF-8?q?=E5=B1=82=E4=BF=A1=E5=8F=B7=E7=86=94=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 回源此前只有单个 http 代理，且 RetryRequest 的三次重试复用同一个代理， 一条链路坏掉时表现为 "All attempts fail: #1 EOF #2 EOF #3 EOF" —— 同一条 死路撞三遍。多中心部署下每个 zone 各自出公网，单出口被限速或被墙即全区 回源失败。 新增 pkg/proxypool： - Pick() 每次调用推进轮询游标，因此同一请求的多次重试自动落到不同出口， 无需改动 RetryRequest 的签名；支持按权重展开轮询环。 - 按应用层信号熔断：传输错误、429、5xx。刻意不计入 401/403，那是 gated 仓库的权限拒绝而非出口故障。冷却到期放行一个半开试探请求，成功即恢复。 - 探活打真实回源域名并实际读取一段 body。gost 的 LB 只做 TCP/CONNECT 层 健康检查，隧道建起来但内部传输超时的情况它看不见，应用层判定只能放在 这里。 - 每成员独立 transport 与连接池；HEAD 单独一份阻止跟随重定向的客户端。 - 私有网段/回环/localhost 旁路：兄弟节点互拉、回环上传口走公网出口必然 失败，还会把健康出口误判成坏的。 接入 pkg/util：constructRoute 统一选路并把结果回灌给池子；流式传输中途 失败一并计入，但客户端主动取消不算在出口头上（remote_task 的 ctx done 分支改为包装 ctx.Err() 以便 errors.Is 识别）。全池熔断时回退直连 bpHfNetLoc，告警日志限频每分钟一条。 配置新增 proxyPool 段；未启用时行为完全不变，只配了旧的 dynamicProxy.httpProxy 会退化成单成员池，存量部署无需改配置。配置有误 时启动即失败，避免带着一个"看起来配了、实际没生效"的池子运行。 附 Prometheus 告警规则与 per-member 指标。 Co-Authored-By: Claude Opus 5 (1M context) --- cmd/main.go | 5 + config/config.yaml | 24 ++ config/prometheus/proxypool_alerts.yml | 58 ++++ internal/downloader/remote_task.go | 5 +- internal/service/sys_service.go | 4 +- pkg/config/config.go | 52 ++++ pkg/config/proxypool_cfg_test.go | 22 ++ pkg/proxypool/metrics.go | 70 +++++ pkg/proxypool/pool.go | 383 +++++++++++++++++++++++++ pkg/proxypool/pool_test.go | 240 ++++++++++++++++ pkg/proxypool/probe.go | 93 ++++++ pkg/proxypool/probe_test.go | 112 ++++++++ pkg/util/http_util.go | 125 ++++++-- pkg/util/proxy.go | 129 +++++++++ pkg/util/proxy_test.go | 328 +++++++++++++++++++++ 15 files changed, 1623 insertions(+), 27 deletions(-) create mode 100644 config/prometheus/proxypool_alerts.yml create mode 100644 pkg/config/proxypool_cfg_test.go create mode 100644 pkg/proxypool/metrics.go create mode 100644 pkg/proxypool/pool.go create mode 100644 pkg/proxypool/pool_test.go create mode 100644 pkg/proxypool/probe.go create mode 100644 pkg/proxypool/probe_test.go create mode 100644 pkg/util/proxy.go create mode 100644 pkg/util/proxy_test.go diff --git a/cmd/main.go b/cmd/main.go index c9ba8c7..c5cbc6b 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -26,6 +26,7 @@ import ( "dingospeed/pkg/app" "dingospeed/pkg/config" log "dingospeed/pkg/logger" + "dingospeed/pkg/util" ) var ( @@ -53,6 +54,10 @@ func main() { } log.InitLogger() + // 代理池必须在任何回源请求之前建好，探活循环随之启动。 + if err = util.InitProxyPool(); err != nil { + panic(err) + } myapp, f, err := wireApp(conf) if err != nil { panic(err) diff --git a/config/config.yaml b/config/config.yaml index 823a63d..85f56b0 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -90,6 +90,30 @@ dynamicProxy: maxContinuousFails: 5 #连续失败次数超过该值，则认为代理不可用 webhook: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=73662ac1-1055-48a7-8c89-37964b5f4fdc111 # 企业微信机器人Webhook地址 +# 出口代理池。启用后 dynamicProxy 的单代理与定时探活自动让位给这里。 +# 每次取用出口都会轮换，因此一次请求的多次重试会落在不同出口上； +# 出口按应用层信号（传输中断/429/403/5xx）熔断，冷却后半开自动恢复。 +proxyPool: + enabled: false # 置 true 且 members 非空时生效 + probeTarget: "https://hf-mirror.com/api/models/bert-base-uncased" # 探活地址，必须是代理实际要访问的域名 + probeInterval: 60 # 探活周期，秒 + probeTimeout: 10 # 单次探活超时，秒 + failThreshold: 3 # 连续失败多少次熔断该出口 + cooldown: 300 # 熔断冷却时长，秒；到期后放行一个半开试探请求 + dialTimeout: 10 # 建连超时，秒 + responseHeaderTimeout: 60 # 等响应头超时，秒；隧道通但对端不吐数据靠它暴露 + fallbackDirect: true # 全池熔断时回退直连 bpHfNetLoc + noProxy: [] # 额外免代理域名后缀；私有网段与回环已默认旁路，无需列出 + members: [] + # members 示例（gost 集群各节点的本地 LB 端口）： + # members: + # - name: hd-02 + # url: http://10.201.3.93:8121 + # weight: 2 + # - name: hd-04 + # url: http://10.220.70.213:8121 + # weight: 1 + modelscope: officialBaseURL: https://www.modelscope.cn # ModelScope官方基础地址 chunkSize: 8388608 # 8MB分块，16*1024*1024的数值结果 diff --git a/config/prometheus/proxypool_alerts.yml b/config/prometheus/proxypool_alerts.yml new file mode 100644 index 0000000..b51b0be --- /dev/null +++ b/config/prometheus/proxypool_alerts.yml @@ -0,0 +1,58 @@ +# 代理池告警规则。挂到 Prometheus 的 rule_files 下即可。 +# 指标由 dingospeed 的 /metrics 暴露（server.metrics: true）。 +groups: + - name: dingospeed-proxypool + rules: + # 单个出口熔断：还有别的出口顶着，属于可容忍的降级，但要知道是哪个坏了。 + - alert: ProxyPoolMemberDown + expr: proxypool_member_healthy == 0 + for: 5m + labels: + severity: warning + annotations: + summary: "出口 {{ $labels.member }} 已熔断" + description: "实例 {{ $labels.instance }} 的代理出口 {{ $labels.member }} 连续失败被摘除超过 5 分钟，请检查该 gost 节点的公网出口。" + + # 全池熔断：回源已经退化为直连备用域名，属于故障态。 + - alert: ProxyPoolExhausted + expr: proxypool_available_members == 0 + for: 2m + labels: + severity: critical + annotations: + summary: "代理池全部出口不可用" + description: "实例 {{ $labels.instance }} 的所有出口均已熔断，回源已回退直连 bpHfNetLoc，跨境下载可能大面积失败。" + + # 冗余度不足：只剩一个出口时它一挂就全池熔断，需要提前补。 + - alert: ProxyPoolLowRedundancy + expr: proxypool_available_members == 1 + for: 15m + labels: + severity: warning + annotations: + summary: "代理池只剩 1 个可用出口" + description: "实例 {{ $labels.instance }} 已无冗余，该出口再失败即触发全池熔断。" + + # 出口失败率偏高：还没到熔断阈值但已经在拖慢下载。 + - alert: ProxyPoolMemberHighFailureRate + expr: | + sum by (instance, member) (rate(proxypool_member_request_total{result="fail"}[10m])) + / + clamp_min(sum by (instance, member) (rate(proxypool_member_request_total[10m])), 0.001) + > 0.3 + for: 10m + labels: + severity: warning + annotations: + summary: "出口 {{ $labels.member }} 失败率超过 30%" + description: "近 10 分钟失败率 {{ $value | humanizePercentage }}，该出口可能已被限速或部分链路不通。" + + # 频繁回退直连说明池子整体不健康，直连通常更慢且更容易被拦。 + - alert: ProxyPoolFrequentDirectFallback + expr: rate(proxypool_fallback_direct_total[10m]) > 0 + for: 10m + labels: + severity: warning + annotations: + summary: "回源持续回退直连" + description: "实例 {{ $labels.instance }} 近 10 分钟持续出现全池熔断后的直连回退。" diff --git a/internal/downloader/remote_task.go b/internal/downloader/remote_task.go index ad9aa32..7655547 100644 --- a/internal/downloader/remote_task.go +++ b/internal/downloader/remote_task.go @@ -271,7 +271,10 @@ func (r *RemoteFileTask) getFileRangeFromRemote(startPos, endPos int64, contentC select { case contentChan <- chunk[:n]: case <-r.Context.Done(): - return fmt.Errorf("form remote ctx done") + // 包装 ctx.Err() 而非返回裸字符串， + // 上层（如代理池计分）要靠 errors.Is 把 + // 「客户端主动取消」和「出口故障」区分开。 + return fmt.Errorf("form remote ctx done: %w", r.Context.Err()) } } chunkByteLen += n // 原始数量 diff --git a/internal/service/sys_service.go b/internal/service/sys_service.go index 374e3a4..db92b87 100644 --- a/internal/service/sys_service.go +++ b/internal/service/sys_service.go @@ -40,7 +40,9 @@ func NewSysService(schedulerDao *dao.SchedulerDao) *SysService { if config.SysConfig.DiskClean.Enabled { go sysSvc.cycleCheckDiskUsage() } - if config.SysConfig.DynamicProxy.HttpProxyConnTest { + // 代理池启用后由 proxypool 自己的探活负责健康判定； + // 旧的全局二值 ProxyIsAvailable 会与池子的熔断状态互相打架，必须停掉。 + if config.SysConfig.DynamicProxy.HttpProxyConnTest && !config.SysConfig.IsProxyPoolEnabled() { go sysSvc.cycleTestProxyConnectivity() } }) diff --git a/pkg/config/config.go b/pkg/config/config.go index 7163348..93cdff4 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -45,6 +45,7 @@ type Config struct { TokenBucketLimit TokenBucketLimit `json:"tokenBucketLimit" yaml:"tokenBucketLimit"` DiskClean DiskClean `json:"diskClean" yaml:"diskClean"` DynamicProxy DynamicProxy `json:"dynamicProxy" yaml:"dynamicProxy"` + ProxyPool ProxyPool `json:"proxyPool" yaml:"proxyPool"` Scheduler Scheduler `json:"scheduler" yaml:"scheduler"` Upload Upload `json:"upload" yaml:"upload"` mu sync.RWMutex @@ -159,6 +160,39 @@ type DynamicProxy struct { Webhook string `json:"webhook " yaml:"webhook"` } +// ProxyPool 出口代理池。多中心部署下每个 zone 各自出公网， +// 单出口一旦被限速或被墙，重试打在同一条死路上毫无意义； +// 池化后每次重试自动换出口，坏出口按应用层信号熔断。 +type ProxyPool struct { + Enabled bool `json:"enabled" yaml:"enabled"` + Members []ProxyPoolMember `json:"members" yaml:"members"` + // ProbeTarget 探活地址，必须是代理实际要访问的域名。 + ProbeTarget string `json:"probeTarget" yaml:"probeTarget"` + // ProbeInterval 探活周期，单位秒，默认 60。 + ProbeInterval int `json:"probeInterval" yaml:"probeInterval"` + // ProbeTimeout 单次探活超时，单位秒，默认 10。 + ProbeTimeout int `json:"probeTimeout" yaml:"probeTimeout"` + // FailThreshold 连续失败多少次后熔断该出口，默认 3。 + FailThreshold int `json:"failThreshold" yaml:"failThreshold"` + // Cooldown 熔断冷却时长，单位秒，默认 300。 + Cooldown int `json:"cooldown" yaml:"cooldown"` + // DialTimeout 建连超时，单位秒，默认 10。 + DialTimeout int `json:"dialTimeout" yaml:"dialTimeout"` + // ResponseHeaderTimeout 等响应头超时，单位秒，默认 60。 + // 隧道通但对端不吐数据的故障主要靠它暴露。 + ResponseHeaderTimeout int `json:"responseHeaderTimeout" yaml:"responseHeaderTimeout"` + // FallbackDirect 全池熔断时是否回退直连备用域名（bpHfNetLoc），默认 true。 + FallbackDirect *bool `json:"fallbackDirect" yaml:"fallbackDirect"` + // NoProxy 额外免代理域名后缀，私有网段与回环已默认旁路，无需在此列出。 + NoProxy []string `json:"noProxy" yaml:"noProxy"` +} + +type ProxyPoolMember struct { + Name string `json:"name" yaml:"name"` + URL string `json:"url" yaml:"url"` + Weight int `json:"weight" yaml:"weight"` +} + type Modelscope struct { OfficialBaseURL string `yaml:"officialBaseURL"` ChunkSize int64 `yaml:"chunkSize"` @@ -395,6 +429,24 @@ func (c *Config) GetUploadStagingCleanupInterval() time.Duration { return time.Duration(c.Upload.StagingCleanupIntervalMinutes) * time.Minute } +// IsProxyPoolEnabled 代理池是否生效。 +// 兼容旧配置：只配了 dynamicProxy.httpProxy 时，视为一个单成员池， +// 这样存量部署不改配置也能拿到熔断与旁路能力。 +func (c *Config) IsProxyPoolEnabled() bool { + if c.ProxyPool.Enabled && len(c.ProxyPool.Members) > 0 { + return true + } + return c.DynamicProxy.HttpProxy != "" +} + +// GetProxyPoolFallbackDirect 全池熔断时是否回退直连备用域名，未配置时默认开启。 +func (c *Config) GetProxyPoolFallbackDirect() bool { + if c.ProxyPool.FallbackDirect == nil { + return true + } + return *c.ProxyPool.FallbackDirect +} + func (c *Config) IsCluster() bool { return c.GetSchedulerModel() == consts.SchedulerModeCluster } diff --git a/pkg/config/proxypool_cfg_test.go b/pkg/config/proxypool_cfg_test.go new file mode 100644 index 0000000..fde38c0 --- /dev/null +++ b/pkg/config/proxypool_cfg_test.go @@ -0,0 +1,22 @@ +package config + +import "testing" + +func TestScanParsesProxyPool(t *testing.T) { + c, err := Scan("../../config/config.yaml") + if err != nil { + t.Fatalf("解析配置失败: %v", err) + } + if c.ProxyPool.ProbeInterval != 60 || c.ProxyPool.Cooldown != 300 || c.ProxyPool.FailThreshold != 3 { + t.Fatalf("proxyPool 字段未正确解析: %+v", c.ProxyPool) + } + if c.ProxyPool.FallbackDirect == nil || !*c.ProxyPool.FallbackDirect { + t.Fatal("fallbackDirect 未解析为 true") + } + if c.ProxyPool.Enabled { + t.Fatal("默认应为关闭") + } + if !c.GetProxyPoolFallbackDirect() { + t.Fatal("GetProxyPoolFallbackDirect 应为 true") + } +} diff --git a/pkg/proxypool/metrics.go b/pkg/proxypool/metrics.go new file mode 100644 index 0000000..5bf8b29 --- /dev/null +++ b/pkg/proxypool/metrics.go @@ -0,0 +1,70 @@ +// Copyright (c) 2025 dingodb.com, Inc. All Rights Reserved +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http:www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package proxypool + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var ( + // MemberHealthy 每个出口的健康状态，1=可用 0=熔断中。 + MemberHealthy = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "proxypool_member_healthy", + Help: "Proxy pool member health, 1 healthy 0 tripped", + }, []string{"member"}) + + // MemberRequestTotal 按出口与结果统计请求数。 + MemberRequestTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "proxypool_member_request_total", + Help: "Total requests routed through each proxy pool member", + }, []string{"member", "result"}) + + // AvailableMembers 当前可用出口数，掉到 0 意味着已全部回退直连。 + AvailableMembers = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "proxypool_available_members", + Help: "Number of proxy pool members currently usable", + }) + + // FallbackDirectTotal 全池熔断后回退直连的次数。 + FallbackDirectTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "proxypool_fallback_direct_total", + Help: "Total times the pool was exhausted and traffic fell back to direct", + }) +) + +func observe(m *Member, ok bool) { + result := "fail" + if ok { + result = "ok" + } + MemberRequestTotal.WithLabelValues(m.name, result).Inc() +} + +func (p *Pool) refreshGauges() { + if p == nil { + return + } + avail := 0 + for _, m := range p.members { + if m.Healthy() { + avail++ + MemberHealthy.WithLabelValues(m.name).Set(1) + } else { + MemberHealthy.WithLabelValues(m.name).Set(0) + } + } + AvailableMembers.Set(float64(avail)) +} diff --git a/pkg/proxypool/pool.go b/pkg/proxypool/pool.go new file mode 100644 index 0000000..99af870 --- /dev/null +++ b/pkg/proxypool/pool.go @@ -0,0 +1,383 @@ +// Copyright (c) 2025 dingodb.com, Inc. All Rights Reserved +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http:www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package proxypool 维护一组可用的出口代理（gost 集群等），对下载链路提供 +// “每次取用即轮换”的选路能力，并按应用层信号对坏出口做熔断与半开恢复。 +// +// 为什么需要它：gost 自带的负载均衡只在 TCP/CONNECT 层做健康检查， +// 隧道建立成功但内部 HTTP 传输超时、被限速、被墙的情况它一概看不见。 +// 应用层的健康判定只能放在 dingospeed 里。 +package proxypool + +import ( + "fmt" + "net" + "net/http" + "net/url" + "strings" + "sync" + "sync/atomic" + "time" +) + +// MemberConfig 单个出口代理的静态配置。 +type MemberConfig struct { + Name string `json:"name" yaml:"name"` + URL string `json:"url" yaml:"url"` + Weight int `json:"weight" yaml:"weight"` +} + +// Config 代理池配置。 +type Config struct { + Enabled bool + Members []MemberConfig + ProbeTarget string + ProbeInterval time.Duration + ProbeTimeout time.Duration + FailThreshold int + Cooldown time.Duration + DialTimeout time.Duration + // ResponseHeaderTimeout 等响应头的上限。这是识别「隧道建起来了但对端不吐数据」 + // 这类故障的主要手段，不能为 0。 + ResponseHeaderTimeout time.Duration + // ReqTimeout 为成员客户端的整体超时，0 表示不限制（大文件流式下载需要 0）。 + ReqTimeout time.Duration + // NoProxy 额外的免代理后缀（域名或 IP 前缀），私有网段已默认旁路。 + NoProxy []string +} + +// Member 一个出口代理，自带独立的 http.Client 与熔断状态。 +type Member struct { + name string + rawURL string + weight int + proxyURL *url.URL + client *http.Client + // headClient 单独一份：HEAD 必须阻止跟随重定向， + // 上层要靠 302 的 Location 头拿 CDN 真实地址，跟过去就拿不到了。 + headClient *http.Client + + mu sync.Mutex + fails int // 连续失败次数 + openUntil time.Time // 熔断到期时刻，零值表示未熔断 + + okTotal atomic.Int64 + failTotal atomic.Int64 +} + +func (m *Member) Name() string { return m.name } + +// Client 返回该出口对应方法的客户端。 +func (m *Member) Client(method string) *http.Client { + if method == http.MethodHead { + return m.headClient + } + return m.client +} + +// tryAcquire 判断此刻能否使用该成员；熔断到期时放行一个半开试探请求。 +// 半开的互斥靠把 openUntil 顺延一个冷却周期实现：拿到试探资格的那个调用 +// 会把窗口推到未来，后续调用因此看到「仍在熔断」而退出，无需额外的标志位。 +// 顺延同时保证了试探请求即使永不返回，下个周期也会再试，不会卡死。 +func (m *Member) tryAcquire(now time.Time, cooldown time.Duration) bool { + m.mu.Lock() + defer m.mu.Unlock() + if m.openUntil.IsZero() { + return true + } + if now.After(m.openUntil) { + m.openUntil = now.Add(cooldown) + return true + } + return false +} + +func (m *Member) markOK() { + m.okTotal.Add(1) + m.mu.Lock() + defer m.mu.Unlock() + m.fails = 0 + m.openUntil = time.Time{} +} + +func (m *Member) markFail(threshold int, cooldown time.Duration) { + m.failTotal.Add(1) + m.mu.Lock() + defer m.mu.Unlock() + m.fails++ + if m.fails < threshold { + return + } + // 只在熔断窗口尚未打开（或已到期）时开一个新窗口。 + // 熔断期内陆续返回的在途失败不再顺延窗口，否则实际冷却时间会远超配置值。 + now := time.Now() + if m.openUntil.IsZero() || now.After(m.openUntil) { + m.openUntil = now.Add(cooldown) + } +} + +// Healthy 报告成员当前是否可用（未处于熔断打开状态）。 +func (m *Member) Healthy() bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.openUntil.IsZero() || time.Now().After(m.openUntil) +} + +// Pool 代理池。 +type Pool struct { + cfg Config + members []*Member + ring []*Member // 按权重展开的轮询环 + cursor atomic.Uint64 + noProxy []string + stopCh chan struct{} + stopOne sync.Once +} + +// New 构建代理池。members 为空或 Enabled=false 时返回 nil，调用方据此回退到旧逻辑。 +func New(cfg Config) (*Pool, error) { + if !cfg.Enabled || len(cfg.Members) == 0 { + return nil, nil + } + if cfg.FailThreshold <= 0 { + cfg.FailThreshold = 3 + } + if cfg.Cooldown <= 0 { + cfg.Cooldown = 5 * time.Minute + } + if cfg.DialTimeout <= 0 { + cfg.DialTimeout = 10 * time.Second + } + if cfg.ProbeInterval <= 0 { + cfg.ProbeInterval = time.Minute + } + if cfg.ProbeTimeout <= 0 { + cfg.ProbeTimeout = 10 * time.Second + } + if cfg.ResponseHeaderTimeout <= 0 { + cfg.ResponseHeaderTimeout = 60 * time.Second + } + + p := &Pool{cfg: cfg, stopCh: make(chan struct{})} + for _, s := range cfg.NoProxy { + if s = strings.TrimSpace(strings.ToLower(s)); s != "" { + p.noProxy = append(p.noProxy, s) + } + } + + seen := make(map[string]struct{}, len(cfg.Members)) + for i, mc := range cfg.Members { + raw := strings.TrimSpace(mc.URL) + if raw == "" { + return nil, fmt.Errorf("proxypool: 第 %d 个成员缺少 url", i+1) + } + u, err := url.Parse(raw) + if err != nil { + return nil, fmt.Errorf("proxypool: 成员 %s 的 url 无法解析: %w", raw, err) + } + if u.Scheme == "" || u.Host == "" { + return nil, fmt.Errorf("proxypool: 成员 %s 的 url 需形如 http://host:port", raw) + } + name := strings.TrimSpace(mc.Name) + if name == "" { + name = u.Host + } + if _, dup := seen[name]; dup { + return nil, fmt.Errorf("proxypool: 成员名重复: %s", name) + } + seen[name] = struct{}{} + + weight := mc.Weight + if weight <= 0 { + weight = 1 + } + transport := newTransport(u, cfg) + m := &Member{ + name: name, + rawURL: raw, + weight: weight, + proxyURL: u, + client: &http.Client{Timeout: cfg.ReqTimeout, Transport: transport}, + headClient: &http.Client{ + Timeout: cfg.ReqTimeout, + Transport: transport, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + return http.ErrUseLastResponse // 阻止跟随重定向 + }, + }, + } + p.members = append(p.members, m) + for i := 0; i < weight; i++ { + p.ring = append(p.ring, m) + } + } + return p, nil +} + +// newTransport 每个出口一份 transport，成员之间连接池互不干扰。 +func newTransport(proxyURL *url.URL, cfg Config) *http.Transport { + return &http.Transport{ + Proxy: http.ProxyURL(proxyURL), + MaxIdleConns: 100, + MaxIdleConnsPerHost: 32, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: cfg.DialTimeout, + ForceAttemptHTTP2: false, + ResponseHeaderTimeout: cfg.ResponseHeaderTimeout, + ExpectContinueTimeout: time.Second, + DialContext: (&net.Dialer{ + Timeout: cfg.DialTimeout, + KeepAlive: 30 * time.Second, + }).DialContext, + } +} + +// Pick 取一个可用成员。每次调用都推进游标，因此同一请求的多次重试 +// 会自动落到不同出口——这正是修复 "#1 EOF #2 EOF #3 EOF" 的关键。 +// 全部成员都在熔断中时返回 nil，调用方应回退直连。 +func (p *Pool) Pick() *Member { + if p == nil { + return nil + } + n := len(p.ring) + if n == 0 { + return nil + } + now := time.Now() + start := p.cursor.Add(1) - 1 + for i := 0; i < n; i++ { + m := p.ring[(start+uint64(i))%uint64(n)] + if m.tryAcquire(now, p.cfg.Cooldown) { + return m + } + } + return nil +} + +// Report 上报一次请求结果。statusCode 为 0 表示没拿到响应。 +func (p *Pool) Report(m *Member, statusCode int, err error) { + if p == nil || m == nil { + return + } + // Report 位于每个数据块的下载热路径上，这里只做计数器自增（无锁）， + // 健康度 gauge 的刷新交给探活循环，不在热路径上遍历全部成员。 + if IsFailure(statusCode, err) { + m.markFail(p.cfg.FailThreshold, p.cfg.Cooldown) + observe(m, false) + } else { + m.markOK() + observe(m, true) + } +} + +// IsFailure 判定一次请求是否算出口故障。 +// +// 只认三类信号：传输层错误、429 限流、5xx 服务端错误。 +// +// 刻意不计入 4xx 中的鉴权类状态码：HF 对 gated 仓库返回 401/403， +// 这是「用户没有该资源的权限」而不是「出口坏了」（见 remote_task.go 对 +// 401/403 的处理）。把它算作故障会导致用户拉几次 gated 模型就熔断一个 +// 健康出口。被墙的出口同样可能返回 403，但这两种情况在响应里无法可靠 +// 区分，宁可漏判也不能误杀——漏判由传输错误和 5xx 兜底。 +func IsFailure(statusCode int, err error) bool { + if err != nil { + return true + } + return statusCode == http.StatusTooManyRequests || statusCode >= http.StatusInternalServerError +} + +// Available 返回当前未熔断的成员数。 +func (p *Pool) Available() int { + if p == nil { + return 0 + } + n := 0 + for _, m := range p.members { + if m.Healthy() { + n++ + } + } + return n +} + +// Size 返回成员总数。 +func (p *Pool) Size() int { + if p == nil { + return 0 + } + return len(p.members) +} + +// Members 返回成员列表（只读用途）。 +func (p *Pool) Members() []*Member { + if p == nil { + return nil + } + return p.members +} + +// ShouldBypass 判断目标是否应绕过代理直连。 +// 私有网段/回环必须旁路：兄弟节点互拉、gRPC 连 scheduler、 +// local-upload 回环这些流量走公网代理必然失败。 +func (p *Pool) ShouldBypass(rawURL string) bool { + host := hostOf(rawURL) + if host == "" { + return false + } + if ip := net.ParseIP(host); ip != nil { + return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsUnspecified() + } + if host == "localhost" || strings.HasSuffix(host, ".localhost") { + return true + } + if p == nil { + return false + } + for _, s := range p.noProxy { + if host == s || strings.HasSuffix(host, "."+strings.TrimPrefix(s, ".")) { + return true + } + } + return false +} + +func hostOf(rawURL string) string { + s := strings.TrimSpace(rawURL) + if s == "" { + return "" + } + if !strings.Contains(s, "//") { + s = "//" + s + } + u, err := url.Parse(s) + if err != nil { + return "" + } + host := u.Hostname() + if host == "" { + // 形如 "127.0.0.1:8091" 且解析失败时兜底 + if h, _, e := net.SplitHostPort(strings.TrimPrefix(rawURL, "//")); e == nil { + host = h + } + } + return strings.ToLower(host) +} + +// Close 停止后台探活。 +func (p *Pool) Close() { + if p == nil { + return + } + p.stopOne.Do(func() { close(p.stopCh) }) +} diff --git a/pkg/proxypool/pool_test.go b/pkg/proxypool/pool_test.go new file mode 100644 index 0000000..ad05e13 --- /dev/null +++ b/pkg/proxypool/pool_test.go @@ -0,0 +1,240 @@ +package proxypool + +import ( + "errors" + "net/http" + "testing" + "time" +) + +func newTestPool(t *testing.T, members []MemberConfig, failThreshold int, cooldown time.Duration) *Pool { + t.Helper() + p, err := New(Config{ + Enabled: true, + Members: members, + FailThreshold: failThreshold, + Cooldown: cooldown, + }) + if err != nil { + t.Fatalf("New() 失败: %v", err) + } + if p == nil { + t.Fatal("New() 返回 nil") + } + return p +} + +// 核心诉求：连续取用必须轮换出口，否则重试还是撞同一条死路。 +func TestPickRotatesAcrossMembers(t *testing.T) { + p := newTestPool(t, []MemberConfig{ + {Name: "a", URL: "http://10.0.0.1:8121"}, + {Name: "b", URL: "http://10.0.0.2:8121"}, + {Name: "c", URL: "http://10.0.0.3:8121"}, + }, 3, time.Minute) + + seen := map[string]bool{} + for i := 0; i < 3; i++ { + m := p.Pick() + if m == nil { + t.Fatalf("第 %d 次 Pick 返回 nil", i) + } + if seen[m.Name()] { + t.Fatalf("连续 3 次 Pick 重复命中 %s，重试不会换出口", m.Name()) + } + seen[m.Name()] = true + } +} + +func TestPickHonorsWeight(t *testing.T) { + p := newTestPool(t, []MemberConfig{ + {Name: "heavy", URL: "http://10.0.0.1:8121", Weight: 3}, + {Name: "light", URL: "http://10.0.0.2:8121", Weight: 1}, + }, 3, time.Minute) + + counts := map[string]int{} + for i := 0; i < 8; i++ { + counts[p.Pick().Name()]++ + } + if counts["heavy"] != 6 || counts["light"] != 2 { + t.Fatalf("权重未生效: %v", counts) + } +} + +// HEAD 必须拿到 302 本身而不是跟过去：上层要靠 Location 头解析 CDN 真实地址。 +func TestHeadClientDoesNotFollowRedirect(t *testing.T) { + p := newTestPool(t, []MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, 3, time.Minute) + m := p.Members()[0] + + head := m.Client(http.MethodHead) + if head.CheckRedirect == nil { + t.Fatal("HEAD 客户端必须阻止跟随重定向") + } + if err := head.CheckRedirect(nil, nil); err != http.ErrUseLastResponse { + t.Fatalf("CheckRedirect 应返回 ErrUseLastResponse，实际 %v", err) + } + if get := m.Client(http.MethodGet); get.CheckRedirect != nil { + t.Fatal("GET 客户端应保持默认跟随重定向") + } + // 两者共用同一个 transport，连接池不重复。 + if head.Transport != m.Client(http.MethodGet).Transport { + t.Fatal("HEAD 与 GET 客户端应共用 transport") + } +} + +func TestBreakerTripsAndExcludesMember(t *testing.T) { + p := newTestPool(t, []MemberConfig{ + {Name: "bad", URL: "http://10.0.0.1:8121"}, + {Name: "good", URL: "http://10.0.0.2:8121"}, + }, 2, time.Hour) + + bad := p.Members()[0] + p.Report(bad, 0, errors.New("EOF")) + if !bad.Healthy() { + t.Fatal("1 次失败不应触发熔断（阈值为 2）") + } + p.Report(bad, 0, errors.New("EOF")) + if bad.Healthy() { + t.Fatal("达到阈值后应熔断") + } + if got := p.Available(); got != 1 { + t.Fatalf("可用成员数应为 1，实际 %d", got) + } + for i := 0; i < 5; i++ { + if m := p.Pick(); m.Name() != "good" { + t.Fatalf("熔断成员仍被选中: %s", m.Name()) + } + } +} + +func TestSuccessResetsFailStreak(t *testing.T) { + p := newTestPool(t, []MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, 3, time.Hour) + m := p.Members()[0] + p.Report(m, 0, errors.New("boom")) + p.Report(m, 0, errors.New("boom")) + p.Report(m, http.StatusOK, nil) + p.Report(m, 0, errors.New("boom")) + if !m.Healthy() { + // 失败必须是「连续」的才熔断，中间成功一次就该清零。 + t.Fatal("中途成功后失败计数未清零") + } +} + +func TestHalfOpenAfterCooldown(t *testing.T) { + p := newTestPool(t, []MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, 1, 30*time.Millisecond) + m := p.Members()[0] + p.Report(m, 0, errors.New("boom")) + if p.Pick() != nil { + t.Fatal("熔断期内不应放行") + } + time.Sleep(50 * time.Millisecond) + if p.Pick() == nil { + t.Fatal("冷却结束后应放行半开试探请求") + } + // 半开试探成功即完全恢复。 + p.Report(m, http.StatusOK, nil) + if !m.Healthy() { + t.Fatal("半开试探成功后应恢复") + } +} + +func TestPoolExhaustedReturnsNil(t *testing.T) { + p := newTestPool(t, []MemberConfig{ + {Name: "a", URL: "http://10.0.0.1:8121"}, + {Name: "b", URL: "http://10.0.0.2:8121"}, + }, 1, time.Hour) + for _, m := range p.Members() { + p.Report(m, 0, errors.New("boom")) + } + if p.Pick() != nil { + t.Fatal("全池熔断时应返回 nil，交由调用方回退直连") + } +} + +func TestIsFailure(t *testing.T) { + cases := []struct { + name string + code int + err error + want bool + }{ + {"传输错误", 0, errors.New("EOF"), true}, + {"429 限流", http.StatusTooManyRequests, nil, true}, + {"502 网关错误", http.StatusBadGateway, nil, true}, + {"503 不可用", http.StatusServiceUnavailable, nil, true}, + {"200 正常", http.StatusOK, nil, false}, + {"302 跳转", http.StatusFound, nil, false}, + // 404 是内容不存在，跟出口好坏无关。 + {"404 内容不存在", http.StatusNotFound, nil, false}, + // 401/403 是 gated 仓库的权限拒绝。若计入失败， + // 用户拉几次 gated 模型就会熔断一个完全健康的出口。 + {"401 未授权", http.StatusUnauthorized, nil, false}, + {"403 gated 仓库", http.StatusForbidden, nil, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := IsFailure(c.code, c.err); got != c.want { + t.Fatalf("IsFailure(%d, %v) = %v, want %v", c.code, c.err, got, c.want) + } + }) + } +} + +func TestShouldBypass(t *testing.T) { + p, err := New(Config{ + Enabled: true, + Members: []MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, + NoProxy: []string{"svc.cluster.local"}, + }) + if err != nil { + t.Fatal(err) + } + cases := []struct { + in string + want bool + }{ + {"http://127.0.0.1:8091", true}, + {"http://10.201.146.65:8090", true}, + {"http://192.168.1.10:8090", true}, + {"http://172.16.5.4:8090", true}, + {"http://localhost:8091", true}, + {"http://dingospeed.svc.cluster.local:8090", true}, + {"https://huggingface.co", false}, + {"https://hf-mirror.com", false}, + {"https://cas-bridge.xethub.hf.co", false}, + {"", false}, + } + for _, c := range cases { + if got := p.ShouldBypass(c.in); got != c.want { + t.Fatalf("ShouldBypass(%q) = %v, want %v", c.in, got, c.want) + } + } +} + +// 池未启用时全部方法必须对 nil 安全，调用方不该到处判空。 +func TestNilPoolIsSafe(t *testing.T) { + var p *Pool + if p.Pick() != nil || p.Available() != 0 || p.Size() != 0 || p.ShouldBypass("https://huggingface.co") { + t.Fatal("nil 池的行为不符合预期") + } + p.Report(nil, 200, nil) + p.Close() +} + +func TestNewRejectsBadConfig(t *testing.T) { + if _, err := New(Config{Enabled: true, Members: []MemberConfig{{Name: "a", URL: ""}}}); err == nil { + t.Fatal("空 url 应报错") + } + if _, err := New(Config{Enabled: true, Members: []MemberConfig{{Name: "a", URL: "10.0.0.1:8121"}}}); err == nil { + t.Fatal("缺少 scheme 的 url 应报错") + } + if _, err := New(Config{Enabled: true, Members: []MemberConfig{ + {Name: "dup", URL: "http://10.0.0.1:8121"}, + {Name: "dup", URL: "http://10.0.0.2:8121"}, + }}); err == nil { + t.Fatal("重名成员应报错") + } + p, err := New(Config{Enabled: false, Members: []MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}}) + if err != nil || p != nil { + t.Fatal("未启用时应返回 (nil, nil)") + } +} diff --git a/pkg/proxypool/probe.go b/pkg/proxypool/probe.go new file mode 100644 index 0000000..53f9dc3 --- /dev/null +++ b/pkg/proxypool/probe.go @@ -0,0 +1,93 @@ +// Copyright (c) 2025 dingodb.com, Inc. All Rights Reserved +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http:www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package proxypool + +import ( + "context" + "io" + "net/http" + "sync" + "time" +) + +// defaultProbeTarget 用真实回源域名做探活。 +// 旧实现探 www.google.com —— 国内本就不通，探活必然失败，等于没探。 +// 探活目标必须是“这个代理实际要去的地方”。 +const defaultProbeTarget = "https://hf-mirror.com/api/models/bert-base-uncased" + +// StartProbe 启动后台探活循环，直到 Close 被调用。 +// 探活只做两件事：把恢复了的成员尽早放回轮转、把已经坏掉但还没被流量打中的成员提前摘掉。 +func (p *Pool) StartProbe() { + if p == nil || len(p.members) == 0 { + return + } + target := p.cfg.ProbeTarget + if target == "" { + target = defaultProbeTarget + } + go func() { + ticker := time.NewTicker(p.cfg.ProbeInterval) + defer ticker.Stop() + p.probeAll(target) + for { + select { + case <-p.stopCh: + return + case <-ticker.C: + p.probeAll(target) + } + } + }() +} + +func (p *Pool) probeAll(target string) { + var wg sync.WaitGroup + for _, m := range p.members { + wg.Add(1) + go func(m *Member) { + defer wg.Done() + code, err := probeOnce(m, target, p.cfg.ProbeTimeout) + // 探活结果与业务流量走同一套计分：探通即清零熔断，探不通累计失败。 + p.Report(m, code, err) + }(m) + } + wg.Wait() + // 健康度 gauge 统一在这里刷新，避免在下载热路径的 Report 里遍历全部成员。 + p.refreshGauges() +} + +func probeOnce(m *Member, target string, timeout time.Duration) (int, error) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return 0, err + } + req.Header.Set("User-Agent", "dingospeed-proxypool/1.0") + // 探活单独用短超时的客户端，不能复用成员那个 Timeout=0 的下载客户端。 + client := &http.Client{Timeout: timeout, Transport: m.client.Transport} + resp, err := client.Do(req) + if err != nil { + return 0, err + } + defer resp.Body.Close() + // 必须真读一段 body：CONNECT 建连成功、响应头也回来了、 + // 但隧道内传输被掐断的情况只有读 body 才暴露得出来。 + _, err = io.CopyN(io.Discard, resp.Body, 1024) + if err != nil && err != io.EOF { + return resp.StatusCode, err + } + return resp.StatusCode, nil +} diff --git a/pkg/proxypool/probe_test.go b/pkg/proxypool/probe_test.go new file mode 100644 index 0000000..a712878 --- /dev/null +++ b/pkg/proxypool/probe_test.go @@ -0,0 +1,112 @@ +package proxypool + +import ( + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +// fakeProxy 起一个最小的正向代理：对明文 http 请求，代理收到的是绝对 URI， +// 直接按 handler 回内容即可，足以覆盖 probeOnce 的真实路径。 +func fakeProxy(t *testing.T, h http.HandlerFunc) *httptest.Server { + t.Helper() + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + return srv +} + +func TestProbeMarksHealthyOnSuccess(t *testing.T) { + srv := fakeProxy(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, strings.Repeat("x", 2048)) + }) + p := newTestPool(t, []MemberConfig{{Name: "a", URL: srv.URL}}, 1, time.Hour) + m := p.Members()[0] + p.Report(m, 0, io.EOF) // 先打成熔断 + if m.Healthy() { + t.Fatal("前置条件不成立：应已熔断") + } + + code, err := probeOnce(m, "http://example.invalid/api/models/x", 5*time.Second) + if err != nil || code != http.StatusOK { + t.Fatalf("探活应成功, code=%d err=%v", code, err) + } + p.Report(m, code, err) + if !m.Healthy() { + t.Fatal("探活成功后应恢复可用") + } +} + +func TestProbeDetectsTruncatedBody(t *testing.T) { + // 模拟 gost 隧道建起来了、响应头也回来了，但正文传一半就断。 + // 这正是 TCP 层健康检查看不见、只有读 body 才暴露的故障。 + srv := fakeProxy(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", "4096") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("short")) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + // handler 返回但没写满 Content-Length，客户端读 body 时会拿到错误。 + }) + p := newTestPool(t, []MemberConfig{{Name: "a", URL: srv.URL}}, 1, time.Hour) + m := p.Members()[0] + + _, err := probeOnce(m, "http://example.invalid/api/models/x", 5*time.Second) + if err == nil { + t.Fatal("正文被截断时探活应判失败") + } +} + +func TestProbeFailsOnDeadProxy(t *testing.T) { + p := newTestPool(t, []MemberConfig{{Name: "dead", URL: "http://127.0.0.1:1"}}, 1, time.Hour) + m := p.Members()[0] + code, err := probeOnce(m, "http://example.invalid/x", 2*time.Second) + if err == nil { + t.Fatalf("连不上的代理应报错, code=%d", code) + } + p.Report(m, code, err) + if m.Healthy() { + t.Fatal("探活失败达阈值后应熔断") + } +} + +func TestProbeAllCoversEveryMember(t *testing.T) { + var hits atomic.Int32 + srv := fakeProxy(t, func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, "ok") + }) + p := newTestPool(t, []MemberConfig{ + {Name: "a", URL: srv.URL}, + {Name: "b", URL: srv.URL + "/"}, + }, 1, time.Hour) + p.cfg.ProbeTimeout = 5 * time.Second + + p.probeAll("http://example.invalid/x") + if got := hits.Load(); got != 2 { + t.Fatalf("每个成员都应被探一次，实际 %d", got) + } + if p.Available() != 2 { + t.Fatalf("探活全通后可用数应为 2，实际 %d", p.Available()) + } +} + +func TestStartProbeStopsOnClose(t *testing.T) { + srv := fakeProxy(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + p := newTestPool(t, []MemberConfig{{Name: "a", URL: srv.URL}}, 1, time.Hour) + p.cfg.ProbeInterval = 10 * time.Millisecond + p.cfg.ProbeTimeout = time.Second + p.StartProbe() + time.Sleep(30 * time.Millisecond) + p.Close() + p.Close() // 重复 Close 不应 panic +} diff --git a/pkg/util/http_util.go b/pkg/util/http_util.go index 53f8b3a..c6477f6 100644 --- a/pkg/util/http_util.go +++ b/pkg/util/http_util.go @@ -16,6 +16,8 @@ package util import ( "bytes" + "context" + "errors" "fmt" "io" "net" @@ -29,6 +31,7 @@ import ( "dingospeed/pkg/config" "dingospeed/pkg/consts" "dingospeed/pkg/prom" + "dingospeed/pkg/proxypool" "github.com/avast/retry-go" "github.com/labstack/echo/v4" @@ -130,13 +133,62 @@ func constructClient(method string) (string, *http.Client, error) { return domain, client, err } +// route 一次出站请求的选路结果。member 为 nil 表示这条请求没走池子（直连或旧逻辑）。 +type route struct { + domain string + client *http.Client + member *proxypool.Member +} + +// report 把本次请求结果回灌给代理池，驱动熔断与恢复。 +func (r route) report(statusCode int, err error) { + if r.member != nil { + ProxyPool().Report(r.member, statusCode, err) + } +} + +func (r route) reportResp(resp *common.Response, err error) { + code := 0 + if resp != nil { + code = resp.StatusCode + } + r.report(code, err) +} + +// constructRoute 为一次出站请求选路。 +// 关键语义：每调用一次就从池子里换一个出口，因此 RetryRequest 的三次重试 +// 会落在三个不同出口上，而不是把同一条死路撞三遍。 +func constructRoute(method string) (route, error) { + if pool := ProxyPool(); pool != nil { + if m := pool.Pick(); m != nil { + return route{ + domain: config.SysConfig.GetHFURLBase(), + client: m.Client(method), + member: m, + }, nil + } + // 全池熔断：所有出口都不可用，回退直连备用域名，至少保证有降级路径。 + if config.SysConfig.GetProxyPoolFallbackDirect() { + proxypool.FallbackDirectTotal.Inc() + // 故障时 QPS 不会下降，逐请求打日志会在最需要看日志的时候把日志刷爆。 + logFallbackThrottled() + client, err := NewHTTPClient(method) + return route{domain: config.SysConfig.GetBpHFURLBase(), client: client}, err + } + } + domain, client, err := constructClient(method) + return route{domain: domain, client: client}, err +} + func Head(requestUri string, headers map[string]string) (*common.Response, error) { - domain, client, err := constructClient(http.MethodHead) + r, err := constructRoute(http.MethodHead) if err != nil { return nil, fmt.Errorf("construct http client err: %v", err) } - requestURL := fmt.Sprintf("%s%s", domain, requestUri) - return doHead(client, requestURL, headers) + requestURL := fmt.Sprintf("%s%s", r.domain, requestUri) + resp, err := doHead(r.client, requestURL, headers) + r.reportResp(resp, err) + return resp, err } func doHead(client *http.Client, targetURL string, headers map[string]string) (*common.Response, error) { @@ -169,12 +221,14 @@ func doHead(client *http.Client, targetURL string, headers map[string]string) (* } func Get(requestUri string, headers map[string]string) (*common.Response, error) { - domain, client, err := constructClient(http.MethodGet) + r, err := constructRoute(http.MethodGet) if err != nil { return nil, fmt.Errorf("construct http client err: %v", err) } - requestURL := fmt.Sprintf("%s%s", domain, requestUri) - return doGet(client, requestURL, headers) + requestURL := fmt.Sprintf("%s%s", r.domain, requestUri) + resp, err := doGet(r.client, requestURL, headers) + r.reportResp(resp, err) + return resp, err } func doGet(client *http.Client, targetURL string, headers map[string]string) (*common.Response, error) { @@ -216,51 +270,69 @@ func doGet(client *http.Client, targetURL string, headers map[string]string) (*c } func GetStream(domain, uri string, headers map[string]string, f func(r *http.Response) error) error { - var ( - client *http.Client - err error - ) - if IsInnerDomain(domain) { - client, err = NewHTTPClient(http.MethodGet) + // 内网目标（兄弟节点互拉、回环上传口）必须旁路代理： + // 这些地址走公网出口必然失败，且会把好出口误判成坏出口。 + if IsInnerDomain(domain) || ProxyPool().ShouldBypass(domain) { + client, err := NewHTTPClient(http.MethodGet) + if err != nil { + return fmt.Errorf("construct http client err: %v", err) + } headers[consts.RequestSourceInner] = Itoa(1) - } else { - domain, client, err = constructClient(http.MethodGet) + _, err = doGetStream(client, fmt.Sprintf("%s%s", domain, uri), headers, f) + return err } + r, err := constructRoute(http.MethodGet) if err != nil { return fmt.Errorf("construct http client err: %v", err) } - requestURL := fmt.Sprintf("%s%s", domain, uri) - return doGetStream(client, requestURL, headers, f) + requestURL := fmt.Sprintf("%s%s", r.domain, uri) + code, err := doGetStream(r.client, requestURL, headers, f) + // 这里上报的 err 包含了流式传输中途的失败（f 回调里读 body 断掉）， + // 而这正是 gost 的 TCP 层健康检查看不见的那一类故障。 + // 但客户端主动取消不能算在出口头上，否则用户中断下载会误伤健康出口。 + if !isClientCanceled(err) { + r.report(code, err) + } + return err } -func doGetStream(client *http.Client, targetURL string, headers map[string]string, f func(r *http.Response) error) error { +// isClientCanceled 判断错误是否源自本地取消（用户中断下载、下游关闭）， +// 这类错误与出口健康无关。 +func isClientCanceled(err error) bool { + return err != nil && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) +} + +// doGetStream 返回响应状态码与错误，状态码供代理池计分使用；未拿到响应时返回 0。 +func doGetStream(client *http.Client, targetURL string, headers map[string]string, f func(r *http.Response) error) (int, error) { escapedURL := strings.ReplaceAll(targetURL, "#", "%23") req, err := http.NewRequest("GET", escapedURL, nil) if err != nil { - return fmt.Errorf("创建GET请求失败: %v", err) + return 0, fmt.Errorf("创建GET请求失败: %v", err) } for key, value := range headers { req.Header.Set(key, value) } resp, err := client.Do(req) if err != nil { - return err + return 0, err } defer resp.Body.Close() respHeaders := make(map[string]interface{}) for key, value := range resp.Header { respHeaders[strings.ToLower(key)] = value } - return f(resp) + return resp.StatusCode, f(resp) } func Post(requestUri string, contentType string, data []byte, headers map[string]string) (*common.Response, error) { - domain, client, err := constructClient(http.MethodPost) + r, err := constructRoute(http.MethodPost) if err != nil { return nil, fmt.Errorf("construct http client err: %v", err) } - requestURL := fmt.Sprintf("%s%s", domain, requestUri) - return doPost(client, requestURL, contentType, data, headers) + requestURL := fmt.Sprintf("%s%s", r.domain, requestUri) + resp, err := doPost(r.client, requestURL, contentType, data, headers) + r.reportResp(resp, err) + return resp, err } func doPost(client *http.Client, targetURL string, contentType string, data []byte, headers map[string]string) (*common.Response, error) { @@ -348,12 +420,13 @@ func ResponseStream(c echo.Context, fileName string, headers map[string]string, } func ForwardRequest(originalReq echo.Context) (*http.Response, error) { - domain, client, err := constructClient(http.MethodGet) + r, err := constructRoute(http.MethodGet) if err != nil { return nil, fmt.Errorf("construct http client err: %v", err) } + client := r.client reqUri := originalReq.Request().URL.Path - targetURL, err := url.Parse(domain) + targetURL, err := url.Parse(r.domain) if err != nil { return nil, fmt.Errorf("url.Parse err: %v", err) } @@ -375,9 +448,11 @@ func ForwardRequest(originalReq echo.Context) (*http.Response, error) { } resp, err := client.Do(proxyReq) if err != nil { + r.report(0, err) zap.S().Warnf("转发请求失败: %s, 错误: %v", targetURL, err) return nil, fmt.Errorf("执行转发请求失败: %v", err) } + r.report(resp.StatusCode, nil) return resp, nil } diff --git a/pkg/util/proxy.go b/pkg/util/proxy.go new file mode 100644 index 0000000..aae8f03 --- /dev/null +++ b/pkg/util/proxy.go @@ -0,0 +1,129 @@ +// Copyright (c) 2025 dingodb.com, Inc. All Rights Reserved +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http:www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package util + +import ( + "fmt" + "sync" + "sync/atomic" + "time" + + "go.uber.org/zap" + + "dingospeed/pkg/config" + "dingospeed/pkg/proxypool" +) + +var globalPool atomic.Pointer[proxypool.Pool] + +// InitProxyPool 依据配置构建全局代理池并启动探活。进程启动时调用一次。 +// 配置有误时返回 error 让进程启动失败：代理池是回源的唯一通路， +// 带着一个「看起来配了、实际没生效」的池子跑起来比起不来更难排查。 +func InitProxyPool() error { + cfg, ok := buildPoolConfig() + if !ok { + zap.S().Info("代理池未启用，回源走直连或旧的单代理逻辑") + return nil + } + p, err := proxypool.New(cfg) + if err != nil { + return fmt.Errorf("代理池初始化失败: %w", err) + } + if p == nil { + return nil + } + globalPool.Store(p) + p.StartProbe() + names := make([]string, 0, p.Size()) + for _, m := range p.Members() { + names = append(names, m.Name()) + } + zap.S().Infof("代理池已启用，成员 %d 个: %v", p.Size(), names) + return nil +} + +// ProxyPool 返回全局代理池，未启用时返回 nil。 +func ProxyPool() *proxypool.Pool { + return globalPool.Load() +} + +var ( + fallbackLogMu sync.Mutex + fallbackLogLast time.Time +) + +// logFallbackThrottled 全池熔断的告警日志每分钟最多一条， +// 精确次数由 proxypool_fallback_direct_total 指标承载。 +func logFallbackThrottled() { + fallbackLogMu.Lock() + defer fallbackLogMu.Unlock() + if time.Since(fallbackLogLast) < time.Minute { + return + } + fallbackLogLast = time.Now() + zap.S().Warnf("代理池全部熔断，请求回退直连 %s（本条日志每分钟最多一次）", + config.SysConfig.GetBpHFURLBase()) +} + +func buildPoolConfig() (proxypool.Config, bool) { + pc := config.SysConfig.ProxyPool + members := make([]proxypool.MemberConfig, 0, len(pc.Members)) + for _, m := range pc.Members { + members = append(members, proxypool.MemberConfig{Name: m.Name, URL: m.URL, Weight: m.Weight}) + } + if !pc.Enabled || len(members) == 0 { + // 兼容旧配置：单个 httpProxy 退化成单成员池。 + if config.SysConfig.GetHttpProxy() == "" { + return proxypool.Config{}, false + } + name := config.SysConfig.GetHttpProxyName() + if name == "" { + name = "legacy" + } + members = []proxypool.MemberConfig{{Name: name, URL: config.SysConfig.GetHttpProxy(), Weight: 1}} + } + + failThreshold := pc.FailThreshold + if failThreshold <= 0 { + // 旧配置里的 maxContinuousFails 语义一致，直接继承。 + failThreshold = config.SysConfig.GetMaxContinuousFails() + } + + return proxypool.Config{ + Enabled: true, + Members: members, + ProbeTarget: pc.ProbeTarget, + ProbeInterval: secOr(pc.ProbeInterval, config.SysConfig.GetDynamicProxyTimePeriod()), + ProbeTimeout: secOr(pc.ProbeTimeout, 10*time.Second), + FailThreshold: failThreshold, + Cooldown: secOr(pc.Cooldown, 5*time.Minute), + DialTimeout: secOr(pc.DialTimeout, 10*time.Second), + // 沿用既有 download.reqTimeout 语义（默认 0 = 不限）， + // 大文件流式下载靠它保持不被整体超时硬砍。 + ReqTimeout: config.SysConfig.GetReqTimeOut(), + ResponseHeaderTimeout: secOr(pc.ResponseHeaderTimeout, 60*time.Second), + NoProxy: pc.NoProxy, + }, true +} + +func secOr(seconds int, fallback time.Duration) time.Duration { + if seconds > 0 { + return time.Duration(seconds) * time.Second + } + if fallback > 0 { + return fallback + } + return 0 +} diff --git a/pkg/util/proxy_test.go b/pkg/util/proxy_test.go new file mode 100644 index 0000000..eb86bd4 --- /dev/null +++ b/pkg/util/proxy_test.go @@ -0,0 +1,328 @@ +// Copyright (c) 2025 dingodb.com, Inc. All Rights Reserved +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http:www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package util + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "dingospeed/pkg/config" + "dingospeed/pkg/proxypool" +) + +// 这些用例覆盖的是「池子接进下载链路」这一层，而不是池子本身： +// 选路、重试换出口、结果回灌、内网旁路、全池熔断回退。 +// 池内部的熔断/权重/探活逻辑在 pkg/proxypool 里单独测。 + +func setupPool(t *testing.T, members []proxypool.MemberConfig, failThreshold int) *proxypool.Pool { + t.Helper() + p, err := proxypool.New(proxypool.Config{ + Enabled: true, + Members: members, + FailThreshold: failThreshold, + Cooldown: time.Hour, + }) + if err != nil { + t.Fatalf("构建代理池失败: %v", err) + } + prev := globalPool.Load() + globalPool.Store(p) + t.Cleanup(func() { globalPool.Store(prev) }) + return p +} + +func setupConfig(t *testing.T, hfNetLoc, bpNetLoc string) { + t.Helper() + prev := config.SysConfig + config.SysConfig = &config.Config{} + config.SysConfig.Server.HfNetLoc = hfNetLoc + config.SysConfig.Server.BpHfNetLoc = bpNetLoc + config.SysConfig.Server.HfScheme = "http" + t.Cleanup(func() { config.SysConfig = prev }) +} + +// 核心回归：一次请求的多次重试必须落在不同出口上。 +// 修复前 RetryRequest 三次重试复用同一个代理，生产日志表现为 #1 EOF #2 EOF #3 EOF。 +func TestConstructRouteRotatesOnRetry(t *testing.T) { + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + setupPool(t, []proxypool.MemberConfig{ + {Name: "a", URL: "http://10.0.0.1:8121"}, + {Name: "b", URL: "http://10.0.0.2:8121"}, + {Name: "c", URL: "http://10.0.0.3:8121"}, + }, 3) + + seen := map[string]bool{} + for i := 0; i < 3; i++ { + r, err := constructRoute(http.MethodGet) + if err != nil { + t.Fatalf("第 %d 次选路失败: %v", i, err) + } + if r.member == nil { + t.Fatalf("第 %d 次选路没走池子", i) + } + if seen[r.member.Name()] { + t.Fatalf("第 %d 次重试又落回 %s，重试没换出口", i, r.member.Name()) + } + seen[r.member.Name()] = true + } +} + +// HEAD 必须拿到不跟随重定向的客户端。 +func TestConstructRouteUsesHeadClient(t *testing.T) { + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, 3) + + r, err := constructRoute(http.MethodHead) + if err != nil { + t.Fatal(err) + } + if r.client.CheckRedirect == nil { + t.Fatal("HEAD 应拿到阻止重定向的客户端") + } + g, _ := constructRoute(http.MethodGet) + if g.client.CheckRedirect != nil { + t.Fatal("GET 应保持默认跟随重定向") + } +} + +// 全池熔断后必须回退直连备用域名，而不是把请求丢掉。 +func TestConstructRouteFallsBackToDirect(t *testing.T) { + setupConfig(t, "hf-mirror.com", "backup.example.com") + p := setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, 1) + p.Report(p.Members()[0], http.StatusBadGateway, nil) + + r, err := constructRoute(http.MethodGet) + if err != nil { + t.Fatal(err) + } + if r.member != nil { + t.Fatal("全池熔断后不应再选中成员") + } + if r.domain != "http://backup.example.com" { + t.Fatalf("应回退到备用域名，实际 %s", r.domain) + } +} + +// gated 仓库返回 403 不能熔断健康出口。 +func TestReportDoesNotTripOnGatedRepo(t *testing.T) { + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + p := setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, 3) + m := p.Members()[0] + + r := route{member: m} + for i := 0; i < 5; i++ { + r.report(http.StatusForbidden, nil) + } + if !m.Healthy() { + t.Fatal("连续 5 次 403（gated 仓库无权限）不应熔断出口") + } +} + +func TestReportTripsOnServerErrors(t *testing.T) { + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + p := setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: "http://10.0.0.1:8121"}}, 3) + m := p.Members()[0] + + r := route{member: m} + for i := 0; i < 3; i++ { + r.report(http.StatusBadGateway, nil) + } + if m.Healthy() { + t.Fatal("连续 3 次 502 应熔断出口") + } +} + +// 客户端主动取消不能记到出口头上。 +func TestIsClientCanceled(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + // remote_task 在 ctx done 时返回的正是这个包装形态。 + {"包装的 ctx 取消", fmt.Errorf("form remote ctx done: %w", context.Canceled), true}, + {"包装的 ctx 超时", fmt.Errorf("wrapped: %w", context.DeadlineExceeded), true}, + {"真实传输错误", fmt.Errorf("premature EOF: expected 100 bytes, got 30"), false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := isClientCanceled(c.err); got != c.want { + t.Fatalf("isClientCanceled(%v) = %v, want %v", c.err, got, c.want) + } + }) + } +} + +// 端到端：GetStream 打真实 HTTP 服务，成功后出口保持健康、状态码正确回灌。 +func TestGetStreamReportsThroughProxy(t *testing.T) { + var viaProxy int + var mu sync.Mutex + proxySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + viaProxy++ + mu.Unlock() + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, "hello") + })) + defer proxySrv.Close() + + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + p := setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: proxySrv.URL}}, 2) + + var gotCode int + err := GetStream("http://hf-mirror.com", "/api/models/x", map[string]string{}, + func(resp *http.Response) error { + gotCode = resp.StatusCode + return nil + }) + if err != nil { + t.Fatalf("GetStream 失败: %v", err) + } + if gotCode != http.StatusOK { + t.Fatalf("状态码 %d", gotCode) + } + mu.Lock() + n := viaProxy + mu.Unlock() + if n != 1 { + t.Fatalf("请求应经过代理出口，实际经过 %d 次", n) + } + if !p.Members()[0].Healthy() { + t.Fatal("成功请求后出口应保持健康") + } +} + +// 端到端：流式传输中途断掉（gost 的 TCP 层健康检查看不见的那类故障）应计入失败。 +func TestGetStreamReportsMidStreamFailure(t *testing.T) { + proxySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, "partial") + })) + defer proxySrv.Close() + + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + p := setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: proxySrv.URL}}, 1) + + err := GetStream("http://hf-mirror.com", "/x", map[string]string{}, + func(resp *http.Response) error { + return fmt.Errorf("premature EOF: expected 100 bytes, got 7") + }) + if err == nil { + t.Fatal("应把回调错误透出") + } + if p.Members()[0].Healthy() { + t.Fatal("流式中途失败应熔断出口") + } +} + +// 端到端：客户端取消不应熔断出口。 +func TestGetStreamIgnoresClientCancel(t *testing.T) { + proxySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, "data") + })) + defer proxySrv.Close() + + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + p := setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: proxySrv.URL}}, 1) + + err := GetStream("http://hf-mirror.com", "/x", map[string]string{}, + func(resp *http.Response) error { + return fmt.Errorf("form remote ctx done: %w", context.Canceled) + }) + if err == nil { + t.Fatal("应把取消错误透出给调用方") + } + if !p.Members()[0].Healthy() { + t.Fatal("客户端取消不应熔断出口") + } +} + +// 内网目标必须旁路代理：兄弟节点互拉走公网出口必然失败， +// 还会把健康出口误判成坏出口。 +func TestGetStreamBypassesInnerDomain(t *testing.T) { + var viaProxy int + proxySrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + viaProxy++ + w.WriteHeader(http.StatusOK) + })) + defer proxySrv.Close() + + peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("inner") == "" { + t.Error("内网请求应带上 inner 标记头") + } + w.WriteHeader(http.StatusOK) + fmt.Fprint(w, "peer-data") + })) + defer peer.Close() + + setupConfig(t, "hf-mirror.com", "hf-mirror.com") + setupPool(t, []proxypool.MemberConfig{{Name: "a", URL: proxySrv.URL}}, 2) + + headers := map[string]string{} + err := GetStream(peer.URL, "/data", headers, func(resp *http.Response) error { + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("status %d", resp.StatusCode) + } + return nil + }) + if err != nil { + t.Fatalf("内网直连失败: %v", err) + } + if viaProxy != 0 { + t.Fatalf("内网请求不应经过代理，实际经过 %d 次", viaProxy) + } +} + +func TestBuildPoolConfigFallsBackToLegacyProxy(t *testing.T) { + prev := config.SysConfig + config.SysConfig = &config.Config{} + config.SysConfig.DynamicProxy.HttpProxy = "http://10.0.0.9:1080" + config.SysConfig.DynamicProxy.HttpProxyName = "旧代理" + config.SysConfig.DynamicProxy.MaxContinuousFails = 4 + defer func() { config.SysConfig = prev }() + + cfg, ok := buildPoolConfig() + if !ok { + t.Fatal("只配了 dynamicProxy 时也应构建单成员池") + } + if len(cfg.Members) != 1 || cfg.Members[0].URL != "http://10.0.0.9:1080" { + t.Fatalf("单成员池内容不对: %+v", cfg.Members) + } + if cfg.Members[0].Name != "旧代理" { + t.Fatalf("应沿用 httpProxyName: %s", cfg.Members[0].Name) + } + // 旧配置的 maxContinuousFails 语义一致，应被继承。 + if cfg.FailThreshold != 4 { + t.Fatalf("应继承 maxContinuousFails，实际 %d", cfg.FailThreshold) + } +} + +func TestBuildPoolConfigDisabled(t *testing.T) { + prev := config.SysConfig + config.SysConfig = &config.Config{} + defer func() { config.SysConfig = prev }() + + if _, ok := buildPoolConfig(); ok { + t.Fatal("既没配 proxyPool 也没配 httpProxy 时不应启用") + } +} From 77fbc3c696f8547bc6b2dcdb91461ac53dea2482 Mon Sep 17 00:00:00 2001 From: mangoknight Date: Wed, 2 Sep 2026 14:43:16 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(upload):=20=E5=A0=B5=E4=BD=8F=E4=B8=8A?= =?UTF-8?q?=E4=BC=A0=E5=8F=A3=E7=9A=84=E6=B5=8F=E8=A7=88=E5=99=A8=E8=B7=A8?= =?UTF-8?q?=E7=AB=99=E5=86=99=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 上传口没有任何身份校验，安全性建立在“绑 127.0.0.1，只有本机 ingest agent 会调它”这个假设上。但绑回环挡不住浏览器——用户的浏览器就在回环上。 可利用路径：上传口的写接口全部读裸 body、不校验 Content-Type，所以一个 Content-Type: text/plain 的 POST 就能带 JSON 打进来；而 text/plain 的 POST 属于 CORS 简单请求，不触发预检，浏览器直接发出。用户访问任意恶意页面， 该页面即可 POST /api/cache/orphans/delete 删除缓存 blob（含回源镜像缓存， 不只是本地上传内容）。 改动： - 上传引擎不再挂 CORSMiddleware，改挂新增的 UploadGuardMiddleware：拒绝带 Origin 头（或 Sec-Fetch-Site 表明跨站）的写请求。浏览器跨站请求必带 Origin， 服务端调用方不带，因此调用方无需任何改动。spinfield 侧是控制面后端调 ingest agent、agent 再调本口，无浏览器直连，不受影响。 - 摘掉上传口的 CORSMiddleware 另有一层收益：去掉 Access-Control-Allow-Origin: * 后，恶意页面无法再读取本口响应（例如 GET /api/cache/repos 枚举缓存清单）。 但要说清楚——CORS 头只决定浏览器允不允许页面读取响应，从不阻止请求到达 服务端，真正挡住写操作的是上面那个 Origin 判断。 - 清掉 docker/config/config.yaml 里已随 upload token 一起删除的 token 字段。 下载引擎的 CORSMiddleware 保持原样：收紧 Allow-Methods 拦不住跨域 POST （简单请求不发预检），只会打断跨域用 JSON 调 /api/cacheJob/* 的调用方， 是纯代价无收益。该口若要限制跨站写入，需要的同样是按 Origin 判断的中间件， 那是另一件事。 这不是鉴权。任何能直接发 HTTP 的进程仍可访问上传口，那需要单独给上传口 加回身份校验；本提交只把浏览器这条路堵死。 Co-Authored-By: Claude Opus 5 (1M context) --- docker/config/config.yaml | 1 - internal/server/upload.go | 12 +++- pkg/middleware/queue_limit.go | 5 ++ pkg/middleware/upload_guard.go | 77 +++++++++++++++++++++ pkg/middleware/upload_guard_test.go | 101 ++++++++++++++++++++++++++++ 5 files changed, 194 insertions(+), 2 deletions(-) create mode 100644 pkg/middleware/upload_guard.go create mode 100644 pkg/middleware/upload_guard_test.go diff --git a/docker/config/config.yaml b/docker/config/config.yaml index ee9a144..5be053b 100644 --- a/docker/config/config.yaml +++ b/docker/config/config.yaml @@ -18,7 +18,6 @@ server: upload: host: 127.0.0.1 port: 8091 - token: "" namespace: dingo-local concurrentLimit: 4 stagingRetentionHours: 168 diff --git a/internal/server/upload.go b/internal/server/upload.go index e3ef656..f184ae5 100644 --- a/internal/server/upload.go +++ b/internal/server/upload.go @@ -71,7 +71,17 @@ func (s *UploadServer) Stop(ctx context.Context) error { func NewUploadEngine() router.UploadEcho { e := echo.New() - e.Use(middleware.CORSMiddleware()) + // 这里刻意不挂 CORSMiddleware，换成 UploadGuardMiddleware。 + // + // 两者作用不同，不要混淆：CORS 响应头只决定「浏览器允不允许页面读取响应」， + // 它从不阻止请求到达服务端。摘掉 CORSMiddleware 的收益只有一条——去掉 + // Access-Control-Allow-Origin: *，使恶意页面无法再读取本口的响应 + // （例如 GET /api/cache/repos 枚举缓存清单）。真正挡住写操作的是 + // UploadGuardMiddleware 按 Origin 做的拒绝。 + // + // 上传口是机器对机器的接口：spinfield 控制面后端调 ingest agent，agent 再调这里， + // 没有浏览器直连，因此去掉 CORS 头不影响任何现有调用方。 + e.Use(middleware.UploadGuardMiddleware()) return router.UploadEcho{Echo: e} } diff --git a/pkg/middleware/queue_limit.go b/pkg/middleware/queue_limit.go index 1eefb7a..f3a8f46 100644 --- a/pkg/middleware/queue_limit.go +++ b/pkg/middleware/queue_limit.go @@ -83,6 +83,11 @@ func CORSMiddleware() echo.MiddlewareFunc { return func(c echo.Context) error { // 设置跨域头 c.Response().Header().Set("Access-Control-Allow-Origin", "*") + // 注意 Allow-Methods 只影响预检请求，删掉这里的 POST 并不能阻止跨域 POST： + // text/plain 的 POST 属于简单请求，压根不发预检。所以这一行不是访问控制， + // 收紧它只会打断跨域用 JSON 调 /api/cacheJob/* 的调用方（那种请求要预检）， + // 拦不住任何攻击。下载口若要真正限制跨站写入，需要的是按 Origin 判断的 + // 中间件，而不是改这里。 c.Response().Header().Set("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS, HEAD") c.Response().Header().Set("Access-Control-Allow-Headers", "*") c.Response().Header().Set("Access-Control-Expose-Headers", "*") diff --git a/pkg/middleware/upload_guard.go b/pkg/middleware/upload_guard.go new file mode 100644 index 0000000..f67aa14 --- /dev/null +++ b/pkg/middleware/upload_guard.go @@ -0,0 +1,77 @@ +// Copyright (c) 2025 dingodb.com, Inc. All Rights Reserved +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http:www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package middleware + +import ( + "net/http" + + "github.com/labstack/echo/v4" + "go.uber.org/zap" +) + +// UploadGuardMiddleware 挡住浏览器发起的跨站请求。 +// +// 上传口没有任何身份校验，安全性完全建立在“只有本机的 ingest agent 会调它” +// 这个假设上。但绑在 127.0.0.1 挡不住浏览器——用户的浏览器就在回环上。 +// +// 具体的攻击路径：上传口的写接口全部读裸 body、不校验 Content-Type， +// 于是一个 text/plain 的 POST 就能带着 JSON 打进来。而 text/plain 的 POST +// 属于 CORS 简单请求，不触发预检，浏览器会直接送出去。用户访问任意一个恶意 +// 页面，那个页面就能 POST /api/cache/orphans/delete 把缓存删掉。 +// +// 这里用 Origin 头来区分：浏览器发起的跨站请求一定带 Origin（简单请求也带）， +// 而 ingest agent、curl 这类服务端调用方不会带。因此拒掉“带 Origin 的写请求” +// 既能挡住浏览器，又不需要任何调用方改代码。Sec-Fetch-Site 是同一判断的补强， +// 现代浏览器都会发，且是禁止修改的头。 +// +// 这不能替代真正的鉴权，只是把浏览器这条路堵死。任何能直接发 HTTP 的进程 +// 仍然可以访问上传口——那个问题要靠给上传口加回身份校验来解决。 +func UploadGuardMiddleware() echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + req := c.Request() + if !isStateChanging(req.Method) { + return next(c) + } + if origin := req.Header.Get("Origin"); origin != "" { + zap.S().Warnf("[UPLOAD] 拒绝跨站写请求: method=%s path=%s origin=%s", + req.Method, req.URL.Path, origin) + return c.JSON(http.StatusForbidden, map[string]string{ + "code": "UPLOAD_CROSS_ORIGIN_DENIED", + "error": "cross-origin requests are not allowed on the upload port", + }) + } + // none = 用户直接输地址；same-origin = 同源页面。其余（cross-site、 + // same-site）都是别的站点发起的。 + if site := req.Header.Get("Sec-Fetch-Site"); site != "" && site != "none" && site != "same-origin" { + zap.S().Warnf("[UPLOAD] 拒绝跨站写请求: method=%s path=%s sec-fetch-site=%s", + req.Method, req.URL.Path, site) + return c.JSON(http.StatusForbidden, map[string]string{ + "code": "UPLOAD_CROSS_ORIGIN_DENIED", + "error": "cross-origin requests are not allowed on the upload port", + }) + } + return next(c) + } + } +} + +func isStateChanging(method string) bool { + switch method { + case http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete: + return true + } + return false +} diff --git a/pkg/middleware/upload_guard_test.go b/pkg/middleware/upload_guard_test.go new file mode 100644 index 0000000..e33890c --- /dev/null +++ b/pkg/middleware/upload_guard_test.go @@ -0,0 +1,101 @@ +// Copyright (c) 2025 dingodb.com, Inc. All Rights Reserved +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http:www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package middleware + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/labstack/echo/v4" +) + +func runGuard(t *testing.T, method, path string, headers map[string]string) (int, bool) { + t.Helper() + e := echo.New() + reached := false + h := UploadGuardMiddleware()(func(c echo.Context) error { + reached = true + return c.NoContent(http.StatusOK) + }) + req := httptest.NewRequest(method, path, strings.NewReader(`{"items":[]}`)) + for k, v := range headers { + req.Header.Set(k, v) + } + rec := httptest.NewRecorder() + if err := h(e.NewContext(req, rec)); err != nil { + t.Fatalf("middleware err: %v", err) + } + return rec.Code, reached +} + +// 核心回归：恶意网页用 text/plain 的 POST 发起简单请求（不触发预检）， +// 试图删掉缓存。浏览器一定会带上 Origin，据此拦截。 +func TestGuardBlocksCrossOriginDelete(t *testing.T) { + code, reached := runGuard(t, http.MethodPost, "/api/cache/orphans/delete", map[string]string{ + "Origin": "https://evil.example.com", + "Content-Type": "text/plain", + }) + if reached { + t.Fatal("跨站写请求不应到达 handler") + } + if code != http.StatusForbidden { + t.Fatalf("应返回 403，实际 %d", code) + } +} + +// 没有 Origin 但 Sec-Fetch-Site 表明是跨站，同样拦掉。 +func TestGuardBlocksBySecFetchSite(t *testing.T) { + for _, site := range []string{"cross-site", "same-site"} { + code, reached := runGuard(t, http.MethodPost, "/api/local-publish/models/a/b/main", + map[string]string{"Sec-Fetch-Site": site}) + if reached || code != http.StatusForbidden { + t.Fatalf("Sec-Fetch-Site=%s 应被拦截，code=%d reached=%v", site, code, reached) + } + } +} + +// ingest agent、curl 这类服务端调用方不带 Origin，必须放行。 +func TestGuardAllowsServerToServer(t *testing.T) { + for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} { + code, reached := runGuard(t, method, "/api/cache/files/delete", map[string]string{ + "Content-Type": "application/json", + }) + if !reached { + t.Fatalf("%s 无 Origin 的服务端调用应放行，code=%d", method, code) + } + } +} + +// 用户在地址栏直接访问（Sec-Fetch-Site: none）不算跨站。 +func TestGuardAllowsSameOriginAndNone(t *testing.T) { + for _, site := range []string{"none", "same-origin"} { + if _, reached := runGuard(t, http.MethodPost, "/api/cache/files/delete", + map[string]string{"Sec-Fetch-Site": site}); !reached { + t.Fatalf("Sec-Fetch-Site=%s 应放行", site) + } + } +} + +// 只读方法不受影响：跨站也读不到响应，因为上传引擎不再发 CORS 头。 +func TestGuardIgnoresReadMethods(t *testing.T) { + for _, method := range []string{http.MethodGet, http.MethodHead, http.MethodOptions} { + if _, reached := runGuard(t, method, "/api/cache/repos", + map[string]string{"Origin": "https://evil.example.com"}); !reached { + t.Fatalf("%s 不应被拦截", method) + } + } +}