feat(proxy): 出口代理池，重试自动换出口并按应用层信号熔断

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)

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的数值结果

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 分钟持续出现全池熔断后的直连回退。"

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

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 // 原始数量

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}
 }
 

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()
 			}
 		})

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
 }

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")
+	}
+}

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", "*")

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
+}