Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 nomdn
|
56d6bc6c08 | ||
|
nomdn
|
981b2fec9c | ||
|
nomdn
|
f1790742a3 |
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-通道接入可选)。
|
||||
|
||||
---
|
||||
|
||||
@@ -21,6 +21,8 @@ function checkRateLimit(ip: string, limit: number): boolean {
|
||||
return entry.count <= limit
|
||||
}
|
||||
|
||||
// 数据上报已移除：统计与拨测明细统一由下游节点（lemon-ipw/src）自行上报，
|
||||
// 本转发入口不再计数、不再 POST /report，转发请求也不再携带 X-Boce-Reporter 归属头（防双算协议已废除）。
|
||||
export default defineEventHandler(async (event) => {
|
||||
// 1) Origin / Referer 校验：跨域浏览器调用一律拒绝。
|
||||
// 注意：无 Origin/Referer 的服务端 SSR 调用与命令行请求无法区分，仅靠下方限流约束。
|
||||
@@ -77,9 +79,9 @@ 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)
|
||||
let apiKey: string | undefined
|
||||
try {
|
||||
const apiKeysMap: Record<string, string> = (runtimeConfig.apiKeys || {}) as Record<string, string>
|
||||
@@ -90,7 +92,7 @@ export default defineEventHandler(async (event) => {
|
||||
} catch (e) {
|
||||
console.debug('[middleware debug] JSON parse FAILED:', e)
|
||||
}
|
||||
// 转发头：不透传 Origin，避免上游 CORS 误判；仅按需注入 Authorization
|
||||
// 转发头：不透传 Origin，避免上游 CORS 误判；仅按需注入 Authorization。
|
||||
const authHeaders: Record<string, string> = {}
|
||||
if (apiKey) {
|
||||
authHeaders['Authorization'] = `Bearer ${apiKey}`
|
||||
@@ -98,7 +100,6 @@ export default defineEventHandler(async (event) => {
|
||||
} else {
|
||||
console.debug('[middleware debug] authHeaders NO Authorization header')
|
||||
}
|
||||
|
||||
if (apiType === 'whois' || apiType === 'dns' || apiType === 'location' || apiType === 'ssl' || apiType === 'asn' || apiType === 'dnssec' || apiType === 'detail') {
|
||||
let apiBaseUrls: Array<{ id: string, url: string ,label: string}> = []
|
||||
switch (apiType) {
|
||||
|
||||
@@ -799,9 +799,9 @@ func main() {
|
||||
app.Get("/v1/*", middlewareHandler)
|
||||
app.Get("/middleware/*", middlewareHandler)
|
||||
|
||||
// 监听地址，优先级: 环境变量 PORT > setting.json port > 默认 8080。
|
||||
// 写回全局 PORT：waitChildReady 的健康检查地址要用同一个端口
|
||||
PORT = os.Getenv("PORT")
|
||||
// 监听地址：readConfig 已按 env PORT > setting.json port 填好全局 PORT，
|
||||
// 这里只在为空时补默认 8080——不要再用 Getenv 覆盖，否则 setting.json 的 port 失效。
|
||||
// waitChildReady 的健康检查地址要用同一个端口
|
||||
if PORT == "" {
|
||||
PORT = "8080"
|
||||
}
|
||||
|
||||
+1
-1
@@ -3,6 +3,7 @@ module lemon-ipw
|
||||
go 1.26.6
|
||||
|
||||
require (
|
||||
github.com/AdguardTeam/golibs v0.35.16
|
||||
github.com/coder/websocket v1.8.15
|
||||
github.com/gin-contrib/cors v1.7.7
|
||||
github.com/gin-gonic/gin v1.12.0
|
||||
@@ -20,7 +21,6 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/AdguardTeam/golibs v0.35.16 // indirect
|
||||
github.com/bytedance/gopkg v0.1.3 // indirect
|
||||
github.com/bytedance/sonic v1.15.0 // indirect
|
||||
github.com/bytedance/sonic/loader v0.5.0 // indirect
|
||||
|
||||
+3
-10
@@ -11,9 +11,9 @@ github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gE
|
||||
github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
|
||||
github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
|
||||
github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
|
||||
github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
|
||||
@@ -80,16 +80,15 @@ github.com/oschwald/maxminddb-golang/v2 v2.3.0 h1:PnXjMGjkSQlwOBSyZ7hk6Fd75t7erk
|
||||
github.com/oschwald/maxminddb-golang/v2 v2.3.0/go.mod h1:NSQvgFwPxODpBTJI5+5Ns1AAucnx7ggW9PSRRifAT1s=
|
||||
github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
|
||||
github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8=
|
||||
github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII=
|
||||
github.com/quic-go/quic-go v0.59.1 h1:0Gmua0HW1Tv7ANR7hUYwRyD0MG5OJfgvYSZasGZzBic=
|
||||
github.com/quic-go/quic-go v0.59.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
|
||||
github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
|
||||
github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
|
||||
github.com/rogpeppe/go-internal v1.15.0 h1:D0RCU5rMAp+SpgkiNdrjfJ+LX4J1M32V2NeCY7EJ6hc=
|
||||
github.com/rogpeppe/go-internal v1.15.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs=
|
||||
github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
|
||||
github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik=
|
||||
github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw=
|
||||
@@ -129,8 +128,6 @@ golang.org/x/arch v0.23.0 h1:lKF64A2jF6Zd8L0knGltUnegD62JMFBiCPBmQpToHhg=
|
||||
golang.org/x/arch v0.23.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A=
|
||||
golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
|
||||
golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
|
||||
golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ=
|
||||
golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0=
|
||||
golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk=
|
||||
golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40=
|
||||
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
|
||||
@@ -142,12 +139,8 @@ golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
|
||||
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
|
||||
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
|
||||
golang.org/x/tools v0.47.0 h1:7Kn5x/d1svx/PzryTsqeoZN4TZwqeH5pGWjefhLi/1Q=
|
||||
golang.org/x/tools v0.47.0/go.mod h1:dFHnyTvFWY212G+h7ZY4Vsp/K3U4/7W9TyVaAul8uCA=
|
||||
golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE=
|
||||
golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk=
|
||||
google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
|
||||
google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
|
||||
+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）
|
||||
//
|
||||
// 两条来源都要上报：
|
||||
// - HTTP 接口收到的请求（/v1/*）→ nodeReportMiddleware 计数
|
||||
// - 中间件经 WS 下发的拨测（probe）→ nodeRecordWSProbe 计数
|
||||
|
||||
const (
|
||||
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 拨测类接口（结果需明细上报）。
|
||||
// 明细覆盖：连通性/证书类 detail、ssl，DNS 查询类 dns，直连类 tcping/speed。
|
||||
// dnssec 属功能性校验（结果是否签名/可信），不逐条进明细，只进统计聚合。
|
||||
func nodeIsProbeType(apiType string) bool {
|
||||
switch apiType {
|
||||
case "tcping", "speed", "detail", "ssl", "dns":
|
||||
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) {
|
||||
start := time.Now()
|
||||
c.Next()
|
||||
|
||||
// 收集中心(ipw-boce)主动调度下发的拨测带 X-Scheduler-Probe 头：
|
||||
// 该次执行由收集中心侧本地落库(source=sched/biz)，本节点跳过计数与明细上报，避免双算。
|
||||
// 真实业务请求（用户/中间件转发）不带此头，仍按"节点是唯一记录者"正常上报。
|
||||
if c.Request.Header.Get("X-Scheduler-Probe") != "" {
|
||||
return
|
||||
}
|
||||
|
||||
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 调用）。
|
||||
// 中间件已不在转发路径上计数，WS 下发的拨测同样由本节点上报（唯一记录者）；
|
||||
// requestId 只作为明细的关联标识带出，不参与归属判断。
|
||||
func nodeRecordWSProbe(requestID, apiType, raw string, query map[string]string, status int, latency time.Duration) {
|
||||
if apiType == "" {
|
||||
return
|
||||
}
|
||||
nodeRecordAPI(apiType, status, latency)
|
||||
if nodeIsProbeType(apiType) {
|
||||
q := ""
|
||||
for k, v := range query {
|
||||
if q != "" {
|
||||
q += "&"
|
||||
}
|
||||
q += k + "=" + v
|
||||
}
|
||||
nodeRecordProbe(nodeProbeRec{
|
||||
RequestID: requestID,
|
||||
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,19 +336,30 @@ func wsRaw(v any) json.RawMessage {
|
||||
return b
|
||||
}
|
||||
|
||||
// wsHandleProbe 处理中间件下发的拨测请求（并发执行）：取数 → probe_result
|
||||
// wsActiveConns 当前已注册的中间件连接（url → conn）：注册成功写入、断开移除，数据上报广播用（report.go）
|
||||
var wsActiveConns sync.Map
|
||||
|
||||
// wsHandleProbe 处理中间件/收集中心经 WS 下发的拨测请求（并发执行）：取数 → probe_result。
|
||||
// 归属规则：报文带 Scheduler=true = 收集中心主动调度拨测（收集中心侧已本地落库 sched/biz），
|
||||
// 本节点跳过 data 上报，避免双算；仅回 probe_result。不带 Scheduler 的 WS probe
|
||||
// （如中间件转发真实业务）仍由本节点上报（节点是唯一记录者，见 report.go）。
|
||||
func wsHandleProbe(c *websocket.Conn, data json.RawMessage) {
|
||||
var req struct {
|
||||
RequestID string `json:"requestId"`
|
||||
APIType string `json:"apiType"`
|
||||
Raw string `json:"raw"`
|
||||
Query map[string]string `json:"query"`
|
||||
Scheduler bool `json:"scheduler"`
|
||||
}
|
||||
if err := json.Unmarshal(data, &req); err != nil || req.RequestID == "" || req.APIType == "" {
|
||||
slog.Warn("ws client bad probe message")
|
||||
return
|
||||
}
|
||||
start := time.Now()
|
||||
status, body := wsProbe(req.APIType, req.Raw, req.Query)
|
||||
if !req.Scheduler {
|
||||
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 +397,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 +429,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