Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 nomdn
|
981b2fec9c |
No files matched your search
+28
-1
@@ -29,8 +29,11 @@
|
||||
- `node-id`：WS 节点 id（与中间件 `api-keys`/`ws-keys` 键、前端配置的节点 `id` 一致；建议用 UUID 唯一标识）
|
||||
- `node-key`：WS 注册 key（与中间件 `ws-keys[节点id]` 一致；**必填**——节点未配置该 key 时中间件拒绝注册并返回 401）
|
||||
- `node-ota`：**OTA 自更新开关**（`"true"` 启用）。启用后节点定期检查本仓库 GitHub Release，发现新版本时下载与当前平台匹配的二进制、**替换自身并重启**（旧二进制保留为 `<程序名>.old` 便于回滚）；默认关闭
|
||||
- `report-url`：**数据上报收集中心** HTTP 基址（如 `"https://collector.example.com"`，即 ipw-boce 中间件）。节点把自身观测的统计/拨测按 `report-interval-seconds` 周期上报；`ws-url` 已启用时优先走 WS `report` 消息（所有已连接中间件广播），WS 未启用或全部掉线时 HTTP POST 到 `<report-url>/report` 兜底；远端配置可覆盖（`report-ignore-config` 加 `report-url` 可禁止）
|
||||
- `report-token`：收集中心 `/report` 鉴权 token（与中间件 `report-token` 配置一致）；属敏感凭据，**不随远端配置覆盖**
|
||||
- `report-interval-seconds`：上报周期秒，缺省 15
|
||||
|
||||
以上字段均可用环境变量覆盖（`PORTS` / `DNS_SERVER` / `DNSSEC_DNS_SERVER` / `BLOCK_PRIVATE_IPS` / `IPDB` / `CORS` / `TRUSTED_PROXIES` / `ACCESS_TOKEN` / `WS_URL` / `NODE_ID` / `NODE_KEY` / `NODE_OTA`）。需要从远端拉取配置时，设置 `remote-config-url` 或环境变量 `REMOTE_CONFIG_URL`（优先级：远端 > 环境变量 > setting.json）。
|
||||
以上字段均可用环境变量覆盖（`PORTS` / `DNS_SERVER` / `DNSSEC_DNS_SERVER` / `BLOCK_PRIVATE_IPS` / `IPDB` / `CORS` / `TRUSTED_PROXIES` / `ACCESS_TOKEN` / `WS_URL` / `NODE_ID` / `NODE_KEY` / `NODE_OTA` / `REPORT_URL` / `REPORT_TOKEN` / `REPORT_INTERVAL_SECONDS`）。需要从远端拉取配置时，设置 `remote-config-url` 或环境变量 `REMOTE_CONFIG_URL`（优先级：远端 > 环境变量 > setting.json）。
|
||||
|
||||
---
|
||||
|
||||
@@ -141,6 +144,22 @@ APIBaseURL: {
|
||||
|
||||
也可以在 `nuxt.config.ts` 的 `runtimeConfig.apiKeys` 里直接写死（不推荐，密钥不应进仓库）。
|
||||
|
||||
### 内置中间件数据上报（调一次上报一次）
|
||||
|
||||
内置中间件可作为数据上报入口：每转发一次请求，向收集中心（ipw-boce 中间件）`POST /report` 上报一次统计与拨测明细。**配置全部经环境变量注入**（敏感项不进仓库）：
|
||||
|
||||
| 环境变量 | 说明 |
|
||||
|------|------|
|
||||
| `NUXT_BOCE_REPORT_URL` | 收集中心基址（如 `https://collector.example.com`）；**留空 = 不上报**（此时转发也不带标记头，由节点兜底记账） |
|
||||
| `NUXT_BOCE_REPORT_TOKEN` | 收集中心 `/report` 鉴权 token（敏感，与收集中心 `report-token` 配置一致） |
|
||||
| `NUXT_BOCE_REPORT_INSTANCE` | 上报方标识（存收集库 `probe_results.origin`）；默认 `frontend-ssr` |
|
||||
|
||||
行为约定：
|
||||
|
||||
- 配置了上报后，内置中间件转发请求会带 `X-Boce-Reporter` 标记头——后端节点据此跳过这些请求的统计（防双算，见后端节点"数据上报"一节）
|
||||
- 上报为 fire-and-forget：失败静默丢弃、不重试（统计是增量累加语义，重试会双算），不影响转发响应
|
||||
- 上报失败不影响主链路；超时 5 秒
|
||||
|
||||
---
|
||||
|
||||
## 独立中间件（middleware-go）
|
||||
@@ -193,6 +212,14 @@ WS 通道涉及**中间件（服务端）**与**后端节点（客户端）**两
|
||||
| 节点（客户端） | `node-id` | `NODE_ID` | 节点 id，与中间件 `api-keys`/`ws-keys` 键一致（建议 UUID） |
|
||||
| 节点（客户端） | `node-key` | `NODE_KEY` | 注册 key，与中间件 `ws-keys[节点id]` 一致，**必填** |
|
||||
|
||||
**数据上报通道**（节点把自身观测的统计/拨测上报到收集中心，归属规则防双算——中间件转发的请求由中间件上报，节点只报直连/前端/边缘流量）：
|
||||
|
||||
| 角色 | 配置项 | 环境变量 | 说明 |
|
||||
|------|--------|----------|------|
|
||||
| 节点（客户端） | `report-url` | `REPORT_URL` | 收集中心（ipw-boce 中间件）HTTP 基址；WS 在线时优先走 WS 广播，掉线时 POST `<report-url>/report` 兜底 |
|
||||
| 节点（客户端） | `report-token` | `REPORT_TOKEN` | `/report` 鉴权 token，与收集中心 `report-token` 一致；**不随远端配置覆盖** |
|
||||
| 节点（客户端） | `report-interval-seconds` | `REPORT_INTERVAL_SECONDS` | 上报周期秒，缺省 15 |
|
||||
|
||||
**链路**：中间件收到 `ws:true` 节点的拨测请求 → 经 WS 发 `probe` → 节点执行探针（直调 webtest 函数 + 缓存）→ `probe_result` 回传 → 中间件返回 HTTP 响应（`Content-Type: application/json`）。节点侧接入步骤见 [后端节点部署 - WS 通道接入](/guide/deploy-node#ws-通道接入可选)。
|
||||
|
||||
---
|
||||
|
||||
@@ -64,6 +64,13 @@ export default defineNuxtConfig({
|
||||
runtimeConfig: {
|
||||
indexnowKey: '',
|
||||
apiKeys: '',
|
||||
// 数据上报（协议见收集中心中间件 report.go）。敏感项 token 只从环境变量注入，不进仓库。
|
||||
// NUXT_BOCE_REPORT_URL：收集中心基址（如 https://collector.example.com）；空 = 不上报
|
||||
boceReportUrl: '',
|
||||
// NUXT_BOCE_REPORT_TOKEN：收集中心 /report 鉴权 token
|
||||
boceReportToken: '',
|
||||
// NUXT_BOCE_REPORT_INSTANCE：上报方标识（存收集库 origin）；空 = 默认 frontend-ssr
|
||||
boceReportInstance: '',
|
||||
public: {
|
||||
siteUrl: config.siteUrl,
|
||||
docConfig: docConfig,
|
||||
|
||||
@@ -21,6 +21,54 @@ function checkRateLimit(ip: string, limit: number): boolean {
|
||||
return entry.count <= limit
|
||||
}
|
||||
|
||||
// ---- 数据上报（协议见收集中心中间件 ipw-boce 的 report.go）：调一次上报一次 ----
|
||||
// 配置经 runtimeConfig 从环境变量注入：
|
||||
// NUXT_BOCE_REPORT_URL 收集中心基址（如 https://collector.example.com）；空 = 不上报
|
||||
// NUXT_BOCE_REPORT_TOKEN /report 鉴权 token（敏感，仅环境变量，不进仓库）
|
||||
// NUXT_BOCE_REPORT_INSTANCE 上报方标识（存收集库 probe_results.origin）；空 = 默认 frontend-ssr
|
||||
// 语义：本入口只报"自己转发的请求"（fire-and-forget，失败丢弃不影响主链路）；
|
||||
// 配置了上报时，转发请求会带 X-Boce-Reporter 标记头，节点据此跳过，避免双算。
|
||||
const BOCE_PROBE_TYPES = new Set(['tcping', 'udping', 'speed'])
|
||||
const BOCE_INSTANCE_DEFAULT = 'frontend-ssr'
|
||||
|
||||
function boceReportOnce(rt: any, backendID: string, apiType: string, raw: string, status: number, latencyMs: number, body?: any) {
|
||||
const base: string = rt.boceReportUrl || ''
|
||||
if (!base) return
|
||||
const payload: Record<string, any> = {
|
||||
instance: rt.boceReportInstance || BOCE_INSTANCE_DEFAULT,
|
||||
stats: [{
|
||||
nodeId: backendID,
|
||||
apiType,
|
||||
total: 1,
|
||||
errors: status >= 500 ? 1 : 0,
|
||||
latencySumMs: latencyMs,
|
||||
latencyMaxMs: latencyMs,
|
||||
minute: 0, // 0 = 收集器按自己时钟入桶（前端时钟不可信也不影响聚合口径）
|
||||
}],
|
||||
}
|
||||
if (BOCE_PROBE_TYPES.has(apiType)) {
|
||||
payload.probes = [{
|
||||
nodeId: backendID,
|
||||
apiType,
|
||||
raw,
|
||||
status,
|
||||
latencyMs,
|
||||
source: 'http',
|
||||
body: body === undefined || body === null ? undefined : body,
|
||||
}]
|
||||
}
|
||||
const base2 = base.replace(new RegExp('/+$'), '')
|
||||
$fetch(`${base2}/report`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
authorization: `Bearer ${rt.boceReportToken || ''}`,
|
||||
'content-type': 'application/json',
|
||||
},
|
||||
body: payload,
|
||||
timeout: 5_000,
|
||||
}).catch(() => {}) // 上报失败静默：不影响转发响应
|
||||
}
|
||||
|
||||
export default defineEventHandler(async (event) => {
|
||||
// 1) Origin / Referer 校验：跨域浏览器调用一律拒绝。
|
||||
// 注意：无 Origin/Referer 的服务端 SSR 调用与命令行请求无法区分，仅靠下方限流约束。
|
||||
@@ -77,9 +125,10 @@ export default defineEventHandler(async (event) => {
|
||||
throw createError({ statusCode: 400, statusMessage: 'Invalid path' })
|
||||
}
|
||||
const query = getQuery(event)
|
||||
const runtimeConfig = useRuntimeConfig(event)
|
||||
|
||||
// 从运行时变量读取 APIKEYS (JSON字符串)，解析后按 backendID 查找 token
|
||||
const runtimeConfig = useRuntimeConfig(event)
|
||||
// （runtimeConfig 同时提供数据上报配置 boceReport*，见文件头说明）
|
||||
let apiKey: string | undefined
|
||||
try {
|
||||
const apiKeysMap: Record<string, string> = (runtimeConfig.apiKeys || {}) as Record<string, string>
|
||||
@@ -90,7 +139,9 @@ export default defineEventHandler(async (event) => {
|
||||
} catch (e) {
|
||||
console.debug('[middleware debug] JSON parse FAILED:', e)
|
||||
}
|
||||
// 转发头：不透传 Origin，避免上游 CORS 误判；仅按需注入 Authorization
|
||||
// 转发头：不透传 Origin，避免上游 CORS 误判；仅按需注入 Authorization。
|
||||
// 配置了数据上报时注入 X-Boce-Reporter 标记头（本入口会计数上报，节点据此跳过防双算；
|
||||
// 未配置上报则不打标记，由节点兜底记账）
|
||||
const authHeaders: Record<string, string> = {}
|
||||
if (apiKey) {
|
||||
authHeaders['Authorization'] = `Bearer ${apiKey}`
|
||||
@@ -98,6 +149,10 @@ export default defineEventHandler(async (event) => {
|
||||
} else {
|
||||
console.debug('[middleware debug] authHeaders NO Authorization header')
|
||||
}
|
||||
const boceInstance = (runtimeConfig.boceReportInstance || BOCE_INSTANCE_DEFAULT) as string
|
||||
if (runtimeConfig.boceReportUrl) {
|
||||
authHeaders['X-Boce-Reporter'] = boceInstance
|
||||
}
|
||||
|
||||
if (apiType === 'whois' || apiType === 'dns' || apiType === 'location' || apiType === 'ssl' || apiType === 'asn' || apiType === 'dnssec' || apiType === 'detail') {
|
||||
let apiBaseUrls: Array<{ id: string, url: string ,label: string}> = []
|
||||
@@ -132,6 +187,8 @@ export default defineEventHandler(async (event) => {
|
||||
if (apiBaseUrl.slice(-1) != '/') {
|
||||
apiBaseUrl = `${apiBaseUrl}/`
|
||||
}
|
||||
const boceStart = Date.now()
|
||||
let upstreamStatus = 200
|
||||
data = await $fetch(`${apiBaseUrl}v1/${apiType}/${raw}`, {
|
||||
method: 'GET',
|
||||
headers: authHeaders,
|
||||
@@ -141,13 +198,16 @@ export default defineEventHandler(async (event) => {
|
||||
const errStatus = error?.status ?? error?.statusCode
|
||||
if (errStatus) {
|
||||
// 上游返回的错误响应（4xx/5xx），直接转发其状态码与错误体，不包装成 500
|
||||
upstreamStatus = errStatus
|
||||
setResponseStatus(event, errStatus)
|
||||
return error.data ?? {}
|
||||
}
|
||||
// 网络层错误（无法连接上游）
|
||||
upstreamStatus = 502
|
||||
setResponseStatus(event, 502)
|
||||
return { statusCode: 502, statusMessage: 'Backend unreachable' }
|
||||
})
|
||||
boceReportOnce(runtimeConfig, backendID, apiType, raw, upstreamStatus, Date.now() - boceStart)
|
||||
return data
|
||||
}else if (apiType === 'tcping' || apiType === 'udping' || apiType === 'speed') {
|
||||
// tcping/udping/speed 统一走 APIBaseURL 节点池（平铺三栈）
|
||||
@@ -174,6 +234,8 @@ export default defineEventHandler(async (event) => {
|
||||
Object.entries(query).filter(([_, v]) => v !== undefined) as [string, string][]
|
||||
).toString()
|
||||
|
||||
const boceStart = Date.now()
|
||||
let upstreamStatus = 200
|
||||
data = await $fetch(`${apiBaseUrl}v1/${apiType}/${raw}${queryString ? '?' + queryString : ''}`, {
|
||||
method: 'GET',
|
||||
headers: authHeaders,
|
||||
@@ -183,13 +245,16 @@ export default defineEventHandler(async (event) => {
|
||||
const errStatus = error?.status ?? error?.statusCode
|
||||
if (errStatus) {
|
||||
// 上游返回的错误响应（4xx/5xx），直接转发其状态码与错误体，不包装成 500
|
||||
upstreamStatus = errStatus
|
||||
setResponseStatus(event, errStatus)
|
||||
return error.data ?? {}
|
||||
}
|
||||
// 网络层错误（无法连接上游）
|
||||
upstreamStatus = 502
|
||||
setResponseStatus(event, 502)
|
||||
return { statusCode: 502, statusMessage: 'Backend unreachable' }
|
||||
})
|
||||
boceReportOnce(runtimeConfig, backendID, apiType, raw, upstreamStatus, Date.now() - boceStart, data)
|
||||
return data
|
||||
|
||||
|
||||
|
||||
+37
@@ -15,10 +15,12 @@ import (
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/signal"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/AdguardTeam/golibs/netutil/sysresolv"
|
||||
@@ -241,6 +243,9 @@ var (
|
||||
WS_URL string // WS 客户端：中间件 WS 地址（ws://host:port/ws）
|
||||
WS_NODE_ID string // WS 客户端：节点 id（与中间件 ws-keys 键一致）
|
||||
WS_NODE_KEY string // WS 客户端：注册 key（与中间件 ws-keys[节点id] 一致，可空）
|
||||
REPORT_URL string // 数据上报：收集中心 HTTP 接口基址（report-url / REPORT_URL），WS 全部掉线时的兜底通道
|
||||
REPORT_TOKEN string // 数据上报：/report 鉴权 token（report-token / REPORT_TOKEN）
|
||||
REPORT_INTERVAL int // 数据上报间隔秒（report-interval-seconds / REPORT_INTERVAL_SECONDS），缺省 15
|
||||
NODE_OTA string // OTA 自更新开关（node-ota / NODE_OTA），"true" 时启用
|
||||
TRUSTED_PROXIES string // 可信代理列表（trusted-proxies / TRUSTED_PROXIES），逗号分隔 IP/CIDR，供 ClientIP 判定
|
||||
httpServer *http.Server // 显式持有的 HTTP 服务（Windows OTA 优雅重启需要先 Shutdown）
|
||||
@@ -1070,6 +1075,10 @@ func applyRemoteConfig() {
|
||||
if v := configValue(CONFIG, "node-key"); v != "" && !ignored("node-key") {
|
||||
WS_NODE_KEY = v
|
||||
}
|
||||
// 数据上报收集中心（远端可统一下发；report-token 属凭据，不随远端覆盖）
|
||||
if v := configValue(CONFIG, "report-url"); v != "" && !ignored("report-url") {
|
||||
REPORT_URL = v
|
||||
}
|
||||
// OTA 自更新开关（远端可统一开启/关闭节点的自更新）
|
||||
if v := configValue(CONFIG, "node-ota"); v != "" && !ignored("node-ota") {
|
||||
NODE_OTA = v
|
||||
@@ -1159,6 +1168,23 @@ func readConfig() {
|
||||
if WS_NODE_KEY == "" {
|
||||
WS_NODE_KEY = viper.GetString("node-key")
|
||||
}
|
||||
// 数据上报（report-url / report-token / report-interval-seconds），env 优先；详见 report.go
|
||||
REPORT_URL = os.Getenv("REPORT_URL")
|
||||
if REPORT_URL == "" {
|
||||
REPORT_URL = viper.GetString("report-url")
|
||||
}
|
||||
REPORT_TOKEN = os.Getenv("REPORT_TOKEN")
|
||||
if REPORT_TOKEN == "" {
|
||||
REPORT_TOKEN = viper.GetString("report-token")
|
||||
}
|
||||
REPORT_INTERVAL = viper.GetInt("report-interval-seconds")
|
||||
if v := os.Getenv("REPORT_INTERVAL_SECONDS"); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil && n > 0 {
|
||||
REPORT_INTERVAL = n
|
||||
}
|
||||
} else if REPORT_INTERVAL <= 0 {
|
||||
REPORT_INTERVAL = 15
|
||||
}
|
||||
// NODE_OTA：OTA 自更新开关（环境变量 NODE_OTA 或 setting.json 的 node-ota）
|
||||
NODE_OTA = os.Getenv("NODE_OTA")
|
||||
if NODE_OTA == "" {
|
||||
@@ -1276,6 +1302,9 @@ func main() {
|
||||
slog.Info("WS client enabled", "url", WS_URL, "nodeId", WS_NODE_ID)
|
||||
}
|
||||
|
||||
// 数据上报：WS 在线走 WS 广播，否则 HTTP POST 收集中心 /report（详见 report.go）
|
||||
startNodeReporter()
|
||||
|
||||
// OTA 自更新：定期检查 GitHub Release 并替换二进制（配置 NODE_OTA / node-ota 启用）
|
||||
initOTA(GH_PROXY)
|
||||
|
||||
@@ -1315,6 +1344,7 @@ func main() {
|
||||
if ACCESS_TOKEN != "" {
|
||||
v1.Use(tokenCheck()) // Apply token check middleware to the v1 group
|
||||
}
|
||||
v1.Use(nodeReportMiddleware()) // 数据上报统计：归属规则过滤后计数（详见 report.go）
|
||||
{
|
||||
v1.GET("/detail/*url", checkWebsiteHandler)
|
||||
v1.GET("/ssl/*url", sslCheckHandler)
|
||||
@@ -1354,7 +1384,14 @@ func main() {
|
||||
if otaEnabled() && runtime.GOOS == "windows" {
|
||||
otaHandoverWait = make(chan struct{})
|
||||
<-otaHandoverWait
|
||||
return
|
||||
}
|
||||
// 非 OTA 交接模式：main 必须阻塞，否则 Server goroutine 随 main 返回而消亡，
|
||||
// 进程启动后立即退出。阻塞在退出信号上，收到后走优雅停机。
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
|
||||
<-sigCh
|
||||
gracefulShutdown()
|
||||
}
|
||||
|
||||
// gracefulShutdown 优雅停止 HTTP 服务：停止接收新请求，等待在途请求完成（上限 30s）。
|
||||
|
||||
+324
@@ -0,0 +1,324 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/coder/websocket"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// ==================== 数据上报（节点视角 → 收集中心中间件） ====================
|
||||
//
|
||||
// 与中间件（ipw-boce）的上报协议对接（协议定义见 ipw-boce report.go）：
|
||||
// - WS 客户端启用（WS_URL 配置且至少一条连接在线）→ 经 WS 发 {"type":"report","data":...}
|
||||
// - WS 未启用或全部连接掉线 → HTTP POST 到收集中心 /report（REPORT_URL）
|
||||
//
|
||||
// 归属规则（防双算）：每条请求只由一个入口上报。
|
||||
// - 转发方（Go 中间件）转发时带 X-Boce-Reporter 标记头并自行上报 → 本节点跳过这些请求
|
||||
// - WS 下发的拨测（probe 消息带 requestId）→ 下发方中间件已记录上报 → 本节点跳过
|
||||
// - 其余流量（直连 / 前端内置 TS 中间件 / 边缘函数）→ 本节点上报（唯一记录者）
|
||||
// 注意：TS 中间件/边缘函数若也做"调一次上报一次"，其转发请求必须同样携带
|
||||
// X-Boce-Reporter 头，否则会与本节点的上报双算。
|
||||
|
||||
const (
|
||||
reportMarkerHeader = "X-Boce-Reporter" // 与 ipw-boce 转发标记头一致
|
||||
reportMaxProbes = 500
|
||||
reportMaxStats = 500
|
||||
)
|
||||
|
||||
var (
|
||||
nodeStatsMu sync.Mutex
|
||||
nodeStats = map[string]*nodeStatVal{} // apiType → 计数
|
||||
nodeProbeCh chan nodeProbeRec
|
||||
nodeProbeDropN int64
|
||||
nodeDropMu sync.Mutex
|
||||
)
|
||||
|
||||
type nodeStatVal struct {
|
||||
total, errs, latSum, latMax int64
|
||||
}
|
||||
|
||||
// nodeProbeRec 一条拨测明细（结构与 ipw-boce reportProbe 对齐）
|
||||
type nodeProbeRec struct {
|
||||
RequestID string `json:"requestId,omitempty"`
|
||||
NodeID string `json:"nodeId"`
|
||||
APIType string `json:"apiType"`
|
||||
Raw string `json:"raw"`
|
||||
Query string `json:"query,omitempty"`
|
||||
Status int `json:"status"`
|
||||
LatencyMs int64 `json:"latencyMs"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Source string `json:"source,omitempty"`
|
||||
Body string `json:"body,omitempty"`
|
||||
CreatedAt int64 `json:"createdAt,omitempty"`
|
||||
}
|
||||
|
||||
// nodeIsProbeType 拨测类接口（结果需明细上报）
|
||||
func nodeIsProbeType(apiType string) bool {
|
||||
switch apiType {
|
||||
case "tcping", "speed", "udping":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// startNodeReporter 启动上报循环（WS 启用或配置 REPORT_URL 时；main() 调用）
|
||||
func startNodeReporter() {
|
||||
wsOn := WS_URL != ""
|
||||
if !wsOn && REPORT_URL == "" {
|
||||
return
|
||||
}
|
||||
if REPORT_INTERVAL <= 0 {
|
||||
REPORT_INTERVAL = 15
|
||||
}
|
||||
nodeProbeCh = make(chan nodeProbeRec, 2048)
|
||||
go nodeReportLoop()
|
||||
slog.Info("node reporter enabled", "ws", wsOn, "httpFallback", REPORT_URL, "interval", REPORT_INTERVAL)
|
||||
}
|
||||
|
||||
// nodeReportMiddleware /v1 组统计中间件：归属规则过滤 → 计数 → 拨测类补明细
|
||||
func nodeReportMiddleware() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
// 中间件已带标记：它自己会计数上报，本节点跳过（防双算）
|
||||
if c.Request.Header.Get(reportMarkerHeader) != "" {
|
||||
c.Next()
|
||||
return
|
||||
}
|
||||
start := time.Now()
|
||||
c.Next()
|
||||
|
||||
apiType, raw := nodeAPIFromPath(c)
|
||||
if apiType == "" {
|
||||
return
|
||||
}
|
||||
latency := time.Since(start)
|
||||
status := c.Writer.Status()
|
||||
nodeRecordAPI(apiType, status, latency)
|
||||
if nodeIsProbeType(apiType) {
|
||||
nodeRecordProbe(nodeProbeRec{
|
||||
NodeID: nodeReportNodeID(), APIType: apiType, Raw: raw,
|
||||
Query: c.Request.URL.RawQuery, Status: status, LatencyMs: latency.Milliseconds(),
|
||||
Source: "http", CreatedAt: time.Now().Unix(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// nodeRecordWSProbe WS 拨测出口（ws.go wsHandleProbe 调用）。
|
||||
// requestId 非空 = 中间件下发的请求，中间件侧已计数上报，本节点跳过（防双算）
|
||||
func nodeRecordWSProbe(requestID, apiType, raw string, query map[string]string, status int, latency time.Duration) {
|
||||
if requestID != "" {
|
||||
return
|
||||
}
|
||||
if apiType == "" {
|
||||
return
|
||||
}
|
||||
nodeRecordAPI(apiType, status, latency)
|
||||
if nodeIsProbeType(apiType) {
|
||||
q := ""
|
||||
for k, v := range query {
|
||||
if q != "" {
|
||||
q += "&"
|
||||
}
|
||||
q += k + "=" + v
|
||||
}
|
||||
nodeRecordProbe(nodeProbeRec{
|
||||
NodeID: nodeReportNodeID(), APIType: apiType, Raw: raw, Query: q,
|
||||
Status: status, LatencyMs: latency.Milliseconds(), Source: "ws",
|
||||
CreatedAt: time.Now().Unix(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func nodeRecordAPI(apiType string, status int, latency time.Duration) {
|
||||
ms := latency.Milliseconds()
|
||||
nodeStatsMu.Lock()
|
||||
v, ok := nodeStats[apiType]
|
||||
if !ok {
|
||||
v = &nodeStatVal{}
|
||||
nodeStats[apiType] = v
|
||||
}
|
||||
v.total++
|
||||
if status == 0 || status >= 500 {
|
||||
v.errs++
|
||||
}
|
||||
v.latSum += ms
|
||||
if ms > v.latMax {
|
||||
v.latMax = ms
|
||||
}
|
||||
nodeStatsMu.Unlock()
|
||||
}
|
||||
|
||||
func nodeRecordProbe(rec nodeProbeRec) {
|
||||
if nodeProbeCh == nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case nodeProbeCh <- rec:
|
||||
default:
|
||||
nodeDropMu.Lock()
|
||||
nodeProbeDropN++
|
||||
n := nodeProbeDropN
|
||||
nodeDropMu.Unlock()
|
||||
if n%100 == 1 {
|
||||
slog.Warn("node report probe queue full", "dropped", n)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// nodeReportNodeID 上报身份：优先 node-id（WS_NODE_ID），否则 hostname
|
||||
func nodeReportNodeID() string {
|
||||
if WS_NODE_ID != "" {
|
||||
return WS_NODE_ID
|
||||
}
|
||||
if h, err := os.Hostname(); err == nil && h != "" {
|
||||
return h
|
||||
}
|
||||
return "unknown"
|
||||
}
|
||||
|
||||
// nodeAPIFromPath 从 gin 路由模板提取 apiType 与拨测目标（对齐 WS probe 的 raw 格式）
|
||||
func nodeAPIFromPath(c *gin.Context) (apiType, raw string) {
|
||||
path := c.FullPath() // 形如 /v1/tcping/:ip
|
||||
if !strings.HasPrefix(path, "/v1/") {
|
||||
return "", ""
|
||||
}
|
||||
seg := strings.SplitN(strings.TrimPrefix(path, "/v1/"), "/", 2)
|
||||
if len(seg) == 0 || seg[0] == "" {
|
||||
return "", ""
|
||||
}
|
||||
apiType = seg[0]
|
||||
switch apiType {
|
||||
case "tcping", "dnssec", "whois":
|
||||
raw = c.Param("ip")
|
||||
if raw == "" {
|
||||
raw = c.Param("domain")
|
||||
}
|
||||
case "speed":
|
||||
raw = c.Param("version") + c.Param("url") // v4/example.com（对齐 WS probe raw 格式）
|
||||
case "dns":
|
||||
raw = c.Param("type") + c.Param("domain") // a/example.com
|
||||
case "detail", "ssl":
|
||||
raw = strings.TrimPrefix(c.Param("url"), "/")
|
||||
case "location", "asn":
|
||||
raw = c.Param("ip")
|
||||
}
|
||||
return apiType, raw
|
||||
}
|
||||
|
||||
// nodeReportLoop 周期上报：WS 在线走 WS 广播（所有已注册中间件），否则 HTTP POST 收集中心
|
||||
func nodeReportLoop() {
|
||||
ticker := time.NewTicker(time.Duration(REPORT_INTERVAL) * time.Second)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
payload := nodeBuildPayload()
|
||||
if payload == nil {
|
||||
continue
|
||||
}
|
||||
sent := wsBroadcastReport(payload)
|
||||
if !sent && REPORT_URL != "" {
|
||||
if err := nodeHTTPReport(payload); err != nil {
|
||||
slog.Warn("node http report failed", "error", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// nodeBuildPayload 取走当前计数与拨测缓冲，组装上报报文；无数据返回 nil。
|
||||
// minute 传 0 = 收集器按自己时钟入桶（节点时钟不可信时不影响聚合口径）
|
||||
func nodeBuildPayload() map[string]any {
|
||||
nodeStatsMu.Lock()
|
||||
snap := nodeStats
|
||||
nodeStats = make(map[string]*nodeStatVal, len(snap))
|
||||
nodeStatsMu.Unlock()
|
||||
|
||||
probes := make([]nodeProbeRec, 0, reportMaxProbes)
|
||||
if nodeProbeCh != nil {
|
||||
drain:
|
||||
for len(probes) < reportMaxProbes {
|
||||
select {
|
||||
case r := <-nodeProbeCh:
|
||||
probes = append(probes, r)
|
||||
default:
|
||||
break drain // 队列已空，本轮结束
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(snap) == 0 && len(probes) == 0 {
|
||||
return nil
|
||||
}
|
||||
minute := time.Now().UTC().Unix() / 60
|
||||
stats := make([]map[string]any, 0, len(snap))
|
||||
for apiType, v := range snap {
|
||||
stats = append(stats, map[string]any{
|
||||
"minute": minute, "nodeId": nodeReportNodeID(), "apiType": apiType,
|
||||
"total": v.total, "errors": v.errs, "latencySumMs": v.latSum, "latencyMaxMs": v.latMax,
|
||||
})
|
||||
}
|
||||
payload := map[string]any{
|
||||
"instance": nodeReportNodeID(),
|
||||
"stats": stats,
|
||||
}
|
||||
if len(probes) > 0 {
|
||||
payload["probes"] = probes
|
||||
}
|
||||
return payload
|
||||
}
|
||||
|
||||
// wsBroadcastReport 经 WS 通道向所有在线中间件广播 report 消息；无在线连接返回 false（触发 HTTP 兜底）
|
||||
func wsBroadcastReport(payload map[string]any) bool {
|
||||
sent := false
|
||||
data, _ := json.Marshal(payload)
|
||||
wsActiveConns.Range(func(_, v any) bool {
|
||||
if c, ok := v.(*websocket.Conn); ok && c != nil {
|
||||
if err := wsSendErr(c, wsMsg{Type: "report", NodeID: WS_NODE_ID, Data: data}); err != nil {
|
||||
slog.Warn("node ws report send failed", "error", err)
|
||||
} else {
|
||||
sent = true
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
return sent
|
||||
}
|
||||
|
||||
// nodeHTTPReport HTTP 兜底上报（WS 未启用/全部掉线时）
|
||||
func nodeHTTPReport(payload map[string]any) error {
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
url := strings.TrimSuffix(REPORT_URL, "/") + "/report"
|
||||
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if REPORT_TOKEN != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+REPORT_TOKEN)
|
||||
}
|
||||
client := &http.Client{Timeout: 10 * time.Second}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return &httpStatusError{status: resp.StatusCode}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type httpStatusError struct{ status int }
|
||||
|
||||
func (e *httpStatusError) Error() string {
|
||||
return fmt.Sprintf("collector returned status %d", e.status)
|
||||
}
|
||||
@@ -14,5 +14,8 @@
|
||||
"ws-url": "",
|
||||
"node-id": "",
|
||||
"node-key": "",
|
||||
"node-ota": "false"
|
||||
}
|
||||
"node-ota": "false",
|
||||
"report-url": "",
|
||||
"report-token": "",
|
||||
"report-interval-seconds": 15
|
||||
}
|
||||
@@ -133,7 +133,9 @@ func wsProbe(apiType, raw string, query map[string]string) (int, any) {
|
||||
return webtest.TCPingRun(host, port, n, v, 10*time.Second, 100*time.Millisecond)
|
||||
},
|
||||
func(err error) *webtest.TCPingStats { return &webtest.TCPingStats{IP: "Error: " + err.Error()} },
|
||||
func(v string) *webtest.TCPingStats { return &webtest.TCPingStats{IP: "Skipped due to SINGLE_STACK=" + v} },
|
||||
func(v string) *webtest.TCPingStats {
|
||||
return &webtest.TCPingStats{IP: "Skipped due to SINGLE_STACK=" + v}
|
||||
},
|
||||
)
|
||||
result := &TCPingResult{IPv4: ipv4, IPv6: ipv6}
|
||||
pingCache.Store(cacheKey, pingCacheEntry{result: result, timestamp: time.Now()})
|
||||
@@ -334,7 +336,11 @@ func wsRaw(v any) json.RawMessage {
|
||||
return b
|
||||
}
|
||||
|
||||
// wsHandleProbe 处理中间件下发的拨测请求（并发执行）：取数 → probe_result
|
||||
// wsActiveConns 当前已注册的中间件连接（url → conn）：注册成功写入、断开移除，数据上报广播用（report.go）
|
||||
var wsActiveConns sync.Map
|
||||
|
||||
// wsHandleProbe 处理中间件下发的拨测请求（并发执行）：取数 → probe_result。
|
||||
// 同时按归属规则喂给数据上报：requestId 非空 = 中间件下发（中间件侧已计数上报），本节点跳过
|
||||
func wsHandleProbe(c *websocket.Conn, data json.RawMessage) {
|
||||
var req struct {
|
||||
RequestID string `json:"requestId"`
|
||||
@@ -346,7 +352,9 @@ func wsHandleProbe(c *websocket.Conn, data json.RawMessage) {
|
||||
slog.Warn("ws client bad probe message")
|
||||
return
|
||||
}
|
||||
start := time.Now()
|
||||
status, body := wsProbe(req.APIType, req.Raw, req.Query)
|
||||
nodeRecordWSProbe(req.RequestID, req.APIType, req.Raw, req.Query, status, time.Since(start))
|
||||
payload := struct {
|
||||
RequestID string `json:"requestId"`
|
||||
Status int `json:"status"`
|
||||
@@ -384,7 +392,10 @@ func wsClientOnce(url string) time.Duration {
|
||||
slog.Warn("ws client dial failed", "url", url, "error", err)
|
||||
return 3 * time.Second
|
||||
}
|
||||
defer c.CloseNow()
|
||||
defer func() {
|
||||
wsActiveConns.Delete(url)
|
||||
c.CloseNow()
|
||||
}()
|
||||
|
||||
reg := map[string]string{"nodeId": WS_NODE_ID}
|
||||
if WS_NODE_KEY != "" {
|
||||
@@ -413,6 +424,7 @@ func wsClientOnce(url string) time.Duration {
|
||||
return 3 * time.Second
|
||||
}
|
||||
slog.Info("ws client registered", "nodeId", WS_NODE_ID, "url", url)
|
||||
wsActiveConns.Store(url, c)
|
||||
|
||||
// 心跳 goroutine：每 10s 发 ping，失败重试，连续 3 次失败主动断开（per-URL 循环重连同一中间件）
|
||||
stopHeartbeat := make(chan struct{})
|
||||
|
||||
Reference in new issue
Block a user