Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 nomdn
|
56d6bc6c08 |
No files matched your search
@@ -64,13 +64,6 @@ 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,54 +21,8 @@ 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(() => {}) // 上报失败静默：不影响转发响应
|
||||
}
|
||||
|
||||
// 数据上报已移除：统计与拨测明细统一由下游节点（lemon-ipw/src）自行上报，
|
||||
// 本转发入口不再计数、不再 POST /report，转发请求也不再携带 X-Boce-Reporter 归属头（防双算协议已废除）。
|
||||
export default defineEventHandler(async (event) => {
|
||||
// 1) Origin / Referer 校验：跨域浏览器调用一律拒绝。
|
||||
// 注意：无 Origin/Referer 的服务端 SSR 调用与命令行请求无法区分，仅靠下方限流约束。
|
||||
@@ -128,7 +82,6 @@ export default defineEventHandler(async (event) => {
|
||||
const runtimeConfig = useRuntimeConfig(event)
|
||||
|
||||
// 从运行时变量读取 APIKEYS (JSON字符串)，解析后按 backendID 查找 token
|
||||
// （runtimeConfig 同时提供数据上报配置 boceReport*，见文件头说明）
|
||||
let apiKey: string | undefined
|
||||
try {
|
||||
const apiKeysMap: Record<string, string> = (runtimeConfig.apiKeys || {}) as Record<string, string>
|
||||
@@ -140,8 +93,6 @@ export default defineEventHandler(async (event) => {
|
||||
console.debug('[middleware debug] JSON parse FAILED:', e)
|
||||
}
|
||||
// 转发头：不透传 Origin，避免上游 CORS 误判；仅按需注入 Authorization。
|
||||
// 配置了数据上报时注入 X-Boce-Reporter 标记头（本入口会计数上报，节点据此跳过防双算；
|
||||
// 未配置上报则不打标记，由节点兜底记账）
|
||||
const authHeaders: Record<string, string> = {}
|
||||
if (apiKey) {
|
||||
authHeaders['Authorization'] = `Bearer ${apiKey}`
|
||||
@@ -149,11 +100,6 @@ 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}> = []
|
||||
switch (apiType) {
|
||||
@@ -187,8 +133,6 @@ 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,
|
||||
@@ -198,16 +142,13 @@ 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 节点池（平铺三栈）
|
||||
@@ -234,8 +175,6 @@ 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,
|
||||
@@ -245,16 +184,13 @@ 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
|
||||
|
||||
|
||||
|
||||
+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=
|
||||
|
||||
+23
-23
@@ -15,23 +15,20 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// ==================== 数据上报（节点视角 → 收集中心中间件） ====================
|
||||
// ==================== 数据上报（节点 → 收集中心 ipw-boce） ====================
|
||||
//
|
||||
// 与中间件（ipw-boce）的上报协议对接（协议定义见 ipw-boce report.go）：
|
||||
// 节点是唯一记录者：统计与拨测明细都由本节点上报，中间件不在转发路径上计数，不存在双算。
|
||||
// 上报通道（报文同构，协议定义见 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 头，否则会与本节点的上报双算。
|
||||
// 两条来源都要上报：
|
||||
// - HTTP 接口收到的请求（/v1/*）→ nodeReportMiddleware 计数
|
||||
// - 中间件经 WS 下发的拨测（probe）→ nodeRecordWSProbe 计数
|
||||
|
||||
const (
|
||||
reportMarkerHeader = "X-Boce-Reporter" // 与 ipw-boce 转发标记头一致
|
||||
reportMaxProbes = 500
|
||||
reportMaxStats = 500
|
||||
reportMaxProbes = 500
|
||||
reportMaxStats = 500
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -61,10 +58,12 @@ type nodeProbeRec struct {
|
||||
CreatedAt int64 `json:"createdAt,omitempty"`
|
||||
}
|
||||
|
||||
// nodeIsProbeType 拨测类接口（结果需明细上报）
|
||||
// nodeIsProbeType 拨测类接口（结果需明细上报）。
|
||||
// 明细覆盖：连通性/证书类 detail、ssl，DNS 查询类 dns，直连类 tcping/speed。
|
||||
// dnssec 属功能性校验（结果是否签名/可信），不逐条进明细，只进统计聚合。
|
||||
func nodeIsProbeType(apiType string) bool {
|
||||
switch apiType {
|
||||
case "tcping", "speed", "udping":
|
||||
case "tcping", "speed", "detail", "ssl", "dns":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
@@ -87,14 +86,16 @@ func startNodeReporter() {
|
||||
// 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()
|
||||
|
||||
// 收集中心(ipw-boce)主动调度下发的拨测带 X-Scheduler-Probe 头：
|
||||
// 该次执行由收集中心侧本地落库(source=sched/biz)，本节点跳过计数与明细上报，避免双算。
|
||||
// 真实业务请求（用户/中间件转发）不带此头，仍按"节点是唯一记录者"正常上报。
|
||||
if c.Request.Header.Get("X-Scheduler-Probe") != "" {
|
||||
return
|
||||
}
|
||||
|
||||
apiType, raw := nodeAPIFromPath(c)
|
||||
if apiType == "" {
|
||||
return
|
||||
@@ -113,11 +114,9 @@ func nodeReportMiddleware() gin.HandlerFunc {
|
||||
}
|
||||
|
||||
// nodeRecordWSProbe WS 拨测出口（ws.go wsHandleProbe 调用）。
|
||||
// requestId 非空 = 中间件下发的请求，中间件侧已计数上报，本节点跳过（防双算）
|
||||
// 中间件已不在转发路径上计数，WS 下发的拨测同样由本节点上报（唯一记录者）；
|
||||
// requestId 只作为明细的关联标识带出，不参与归属判断。
|
||||
func nodeRecordWSProbe(requestID, apiType, raw string, query map[string]string, status int, latency time.Duration) {
|
||||
if requestID != "" {
|
||||
return
|
||||
}
|
||||
if apiType == "" {
|
||||
return
|
||||
}
|
||||
@@ -131,7 +130,8 @@ func nodeRecordWSProbe(requestID, apiType, raw string, query map[string]string,
|
||||
q += k + "=" + v
|
||||
}
|
||||
nodeRecordProbe(nodeProbeRec{
|
||||
NodeID: nodeReportNodeID(), APIType: apiType, Raw: raw, Query: q,
|
||||
RequestID: requestID,
|
||||
NodeID: nodeReportNodeID(), APIType: apiType, Raw: raw, Query: q,
|
||||
Status: status, LatencyMs: latency.Milliseconds(), Source: "ws",
|
||||
CreatedAt: time.Now().Unix(),
|
||||
})
|
||||
|
||||
@@ -339,14 +339,17 @@ func wsRaw(v any) json.RawMessage {
|
||||
// wsActiveConns 当前已注册的中间件连接（url → conn）：注册成功写入、断开移除，数据上报广播用（report.go）
|
||||
var wsActiveConns sync.Map
|
||||
|
||||
// wsHandleProbe 处理中间件下发的拨测请求（并发执行）：取数 → probe_result。
|
||||
// 同时按归属规则喂给数据上报：requestId 非空 = 中间件下发（中间件侧已计数上报），本节点跳过
|
||||
// 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")
|
||||
@@ -354,7 +357,9 @@ func wsHandleProbe(c *websocket.Conn, data json.RawMessage) {
|
||||
}
|
||||
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))
|
||||
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"`
|
||||
|
||||
Reference in new issue
Block a user