From a47121f99b2c7e3ceb0249aded853beb56058b29 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Sat, 13 Jun 2026 17:21:43 +0800 Subject: [PATCH] feat(probe): add third-party dial-test adapter for 15C (tsk_-WC9smQK7YRt) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements ProberAgent interface + Alibaba Cloud CloudMonitor Synthetic adapter: - ProbeTarget / VantageResult types aligned with 15A schema (L3 always nil) - AliyunSyntheticAgent: CreateSiteMonitor → poll → map to VantageResult - ISP vantage uses "3rd-{ISP}" prefix; Redis key probe:{node}:CN:{region}:3rd-* - Degradation: API error/rate-limit/timeout → skip vantage, log + counter (no write) - RunOnce(ctx, targets) shape for 15H assembly wiring - Credentials via cfg only (env-var injection, zero hardcoded secrets) - 6 new tests (mock vendor server): normalization, rate-limit, timeout, Redis keys, failed-target L1, no-credentials check; all 22 probe-package tests pass Co-Authored-By: Claude Sonnet 4.6 --- .../internal/scheduler/probe/prober_agent.go | 551 ++++++++++++++++++ .../scheduler/probe/prober_agent_test.go | 337 +++++++++++ 2 files changed, 888 insertions(+) create mode 100644 server/internal/scheduler/probe/prober_agent.go create mode 100644 server/internal/scheduler/probe/prober_agent_test.go diff --git a/server/internal/scheduler/probe/prober_agent.go b/server/internal/scheduler/probe/prober_agent.go new file mode 100644 index 0000000..c0079b2 --- /dev/null +++ b/server/internal/scheduler/probe/prober_agent.go @@ -0,0 +1,551 @@ +// Package probe – prober_agent.go +// +// # 供应商选型：阿里云云监控·云拨测 (Alibaba Cloud CloudMonitor Synthetic Monitoring) +// +// 选型理由： +// 1. 官方支持电信/联通/移动运营商节点筛选（IspCode 参数：246/247/248）。 +// 2. 支持 TCP 连通性探测（TaskType="TCP"），可获取 RTT + 失败原因。 +// 3. 覆盖全国 30+ 省份，节点元数据含 ISP 编码和城市。 +// 4. RPC 风格 REST API，HMAC-SHA1 签名，无需官方 SDK，依赖只有标准库。 +// 5. 按次计费，支持 RAM 子账号（权限最小化，满足运营安全 §06 红线 §2 身份隔离）。 +// +// 身份隔离结论： +// 阿里云 RAM 支持独立子账号 + Action 级别授权 +// （cloudmonitor:CreateSiteMonitor + cloudmonitor:DescribeSiteMonitorData）， +// 可与运营身份完全隔离，无 KYC 障碍。满足红线要求，无需退纯自建。 +// +// API 版本：2019-01-01，RPC 风格 +// 参考文档：https://help.aliyun.com/zh/cms/developer-reference/api-cms-2019-01-01-createsitemonitor +// +// # 降级语义 +// +// API 报错/限流/超时 → 该 vantage 本周期不写 Redis（= 无数据，绝不写失败）。 +// 连续不可用：incrementFailCount + slog.Warn，留给 15G 事件口消费。 +package probe + +import ( + "context" + "crypto/hmac" + "crypto/sha1" //nolint:gosec // Alibaba Cloud ACS v1 signing mandates HMAC-SHA1 + "encoding/base64" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "sort" + "strings" + "sync/atomic" + "time" +) + +// -------------------------------------------------------------------------- +// Public types +// -------------------------------------------------------------------------- + +// ProbeTarget describes a single node endpoint to be dial-tested. +type ProbeTarget struct { + // NodeID identifies the node in the database (matches nodes.id). + NodeID string + + // Host is the hostname or IP address of the node endpoint. + Host string + + // Port is the TCP port to probe (e.g. 443). + Port int +} + +// VantageResult is a normalized dial-test result from a single ISP vantage point. +// +// L3 is always absent for third-party dial-tests. Commercial synthetic-monitoring +// APIs probe TCP/TLS reachability only; they cannot validate the application +// protocol (Hysteria2/REALITY) end-to-end. 15D aggregation treats L3==nil from +// a 3rd-party vantage as "unknown at L3", not as a failure. +type VantageResult struct { + // Vantage is the network vantage point from which the probe was conducted. + // For Alibaba Cloud probes the Country is always "CN"; ISP uses the "3rd-" + // prefix convention (e.g. "3rd-ChinaTelecom") to distinguish from first-party + // probe agents. 15D may apply a configurable weight multiplier on this prefix. + Vantage VantagePoint + + // L1 is the TCP connectivity result. + L1 L1Result + + // L2 is the TLS handshake result. Nil when L1 failed or the vendor did not + // perform TLS probing. + L2 *L2Result + + // L3 is intentionally absent — see package doc above. +} + +// ProberAgent abstracts a third-party dial-test provider. +// Implementations must be safe for concurrent use. +type ProberAgent interface { + // Probe issues multi-ISP dial-tests for the given target and returns one + // VantageResult per ISP vantage point. + // + // Degradation contract: + // - On any provider error (rate-limit, timeout, API error) the + // implementation MUST return (nil, nil) and handle the error internally + // (log + counter). Callers treat an empty slice as "no data this cycle". + // - The implementation MUST NOT return a VantageResult whose L1.OK==false + // solely because the API call failed; that would conflate infra errors + // with real reachability failures. + Probe(ctx context.Context, target ProbeTarget) ([]VantageResult, error) +} + +// -------------------------------------------------------------------------- +// AliyunSyntheticAgent +// -------------------------------------------------------------------------- + +// aliyunISP maps human-readable ISP names (used in VantageResult) to the +// Alibaba Cloud CloudMonitor ISP code strings. +var aliyunISP = []struct { + name string // becomes VantagePoint.ISP with "3rd-" prefix + ispCode string // Alibaba Cloud IspCode +}{ + {"ChinaTelecom", "246"}, + {"ChinaUnicom", "247"}, + {"ChinaMobile", "248"}, +} + +// aliyunEndpoint is the CloudMonitor API base URL. +// Overridable in tests via AliyunSyntheticAgent.baseURL. +const aliyunEndpoint = "https://cloudmonitor.cn-hangzhou.aliyuncs.com/" + +// AliyunSyntheticAgentConfig holds configuration for AliyunSyntheticAgent. +// The AccessKeyID and AccessKeySecret must belong to a RAM sub-account with +// minimal permissions (cloudmonitor:CreateSiteMonitor + +// cloudmonitor:DescribeSiteMonitorData only). They must be injected via +// environment variables; never stored in the database or committed to the repo. +type AliyunSyntheticAgentConfig struct { + // AccessKeyID is the Alibaba Cloud RAM AccessKey ID. + // Source: env var ALIYUN_PROBE_ACCESS_KEY_ID. + AccessKeyID string + + // AccessKeySecret is the Alibaba Cloud RAM AccessKey Secret. + // Source: env var ALIYUN_PROBE_ACCESS_KEY_SECRET. + AccessKeySecret string + + // PollInterval is how long to wait between polling for task results. + // Default: 5 seconds. + PollInterval time.Duration + + // PollTimeout is the maximum time to wait for a single task to complete. + // Default: 60 seconds. + PollTimeout time.Duration + + // ProbeCity is the Alibaba Cloud city code used for all ISP probes. + // Default: "563" (Beijing). Only one city is needed — the ISP is the + // relevant dimension for Pangolin's determination logic. + ProbeCity string + + // BaseURL overrides the default CloudMonitor API endpoint. + // Leave empty for production; set to a mock server URL in tests. + BaseURL string + + // HTTPTimeout overrides the HTTP client timeout per call. + // Zero uses the default of 30 seconds. + HTTPTimeout time.Duration +} + +// AliyunSyntheticAgent implements ProberAgent using Alibaba Cloud CloudMonitor +// Synthetic Monitoring. +type AliyunSyntheticAgent struct { + cfg AliyunSyntheticAgentConfig + store *Store + httpClient *http.Client + baseURL string // overridable in tests + failCount atomic.Int64 + logger *slog.Logger +} + +// NewAliyunSyntheticAgent creates an AliyunSyntheticAgent. +// logger may be nil (falls back to slog.Default()). +func NewAliyunSyntheticAgent(cfg AliyunSyntheticAgentConfig, store *Store, logger *slog.Logger) *AliyunSyntheticAgent { + if cfg.PollInterval == 0 { + cfg.PollInterval = 5 * time.Second + } + if cfg.PollTimeout == 0 { + cfg.PollTimeout = 60 * time.Second + } + if cfg.ProbeCity == "" { + cfg.ProbeCity = "563" // Beijing + } + if logger == nil { + logger = slog.Default() + } + httpTimeout := 30 * time.Second + if cfg.HTTPTimeout > 0 { + httpTimeout = cfg.HTTPTimeout + } + baseURL := aliyunEndpoint + if cfg.BaseURL != "" { + baseURL = cfg.BaseURL + } + return &AliyunSyntheticAgent{ + cfg: cfg, + store: store, + httpClient: &http.Client{ + Timeout: httpTimeout, + }, + baseURL: baseURL, + logger: logger, + } +} + +// NewAliyunSyntheticAgentWithOptions is a convenience constructor for tests that +// need to inject a mock server URL without embedding it in the main config type. +// Production code should use NewAliyunSyntheticAgent. +func NewAliyunSyntheticAgentWithOptions(cfg AliyunSyntheticAgentConfig, store *Store, logger *slog.Logger, baseURL string) *AliyunSyntheticAgent { + cfg.BaseURL = baseURL + return NewAliyunSyntheticAgent(cfg, store, logger) +} + +// SetHTTPClientTimeout replaces the agent's HTTP client timeout. +// Exported only for use in tests; production code should set cfg.HTTPTimeout. +func SetHTTPClientTimeout(a *AliyunSyntheticAgent, d time.Duration) *AliyunSyntheticAgent { + a.httpClient = &http.Client{Timeout: d} + return a +} + +// RunOnce issues a full multi-ISP probe cycle for all given targets and writes +// results to Redis via store. Designed for use in an independent goroutine +// controlled by 15H assembly. +// +// Per-target, per-ISP errors are handled internally (degraded silently); +// the method only returns a non-nil error when the context is cancelled. +func (a *AliyunSyntheticAgent) RunOnce(ctx context.Context, targets []ProbeTarget) error { + for _, t := range targets { + if ctx.Err() != nil { + return ctx.Err() + } + results, err := a.Probe(ctx, t) + if err != nil { + // Probe already logged internally; continue to next target. + continue + } + if len(results) == 0 { + continue + } + // Convert VantageResult slice into the store's NodeReport format. + // Each ISP vantage is a separate SaveReports call (distinct vantage key). + for _, vr := range results { + report := NodeReport{ + NodeID: t.NodeID, + Ts: time.Now().Unix(), + L1: vr.L1, + L2: vr.L2, + // L3: intentionally nil + } + // probeID for 3rd-party: "aliyun-synthetic" (no HMAC secret required; + // writes come from the server process, not an external agent). + if wErr := a.store.SaveReports(ctx, "aliyun-synthetic", vr.Vantage, []NodeReport{report}); wErr != nil { + a.logger.Warn("prober_agent: store write failed", + "node", t.NodeID, "isp", vr.Vantage.ISP, "error", wErr) + } + } + } + return nil +} + +// Probe implements ProberAgent. It creates one TCP synthetic-monitoring task +// per ISP (telecom/unicom/mobile), polls for completion, then maps vendor +// results to VantageResult. +// +// On any API or polling error the ISP vantage is skipped (no result returned, +// no Redis write) per the degradation contract. +func (a *AliyunSyntheticAgent) Probe(ctx context.Context, target ProbeTarget) ([]VantageResult, error) { + var out []VantageResult + for _, isp := range aliyunISP { + vr, err := a.probeISP(ctx, target, isp.ispCode, isp.name) + if err != nil { + cnt := a.failCount.Add(1) + a.logger.Warn("prober_agent: ISP probe failed (degraded, no data written)", + "node", target.NodeID, "isp", isp.name, "error", err, + "consecutive_failures", cnt) + // Degradation: skip this vantage this cycle. + continue + } + a.failCount.Store(0) // reset on success + out = append(out, vr) + } + return out, nil +} + +// probeISP creates a single-ISP task, polls for results, and maps to VantageResult. +func (a *AliyunSyntheticAgent) probeISP(ctx context.Context, target ProbeTarget, ispCode, ispName string) (VantageResult, error) { + taskID, err := a.createTask(ctx, target, ispCode) + if err != nil { + return VantageResult{}, fmt.Errorf("create task: %w", err) + } + + result, err := a.pollResult(ctx, taskID) + if err != nil { + return VantageResult{}, fmt.Errorf("poll result: %w", err) + } + + vp := VantagePoint{ + Country: "CN", + Region: result.Province, + ISP: "3rd-" + ispName, + } + + vr := VantageResult{Vantage: vp} + + if result.Availability > 0 && result.ResponseTime > 0 { + vr.L1 = L1Result{OK: true, RttMs: int(result.ResponseTime)} + } else { + vr.L1 = L1Result{OK: false, RttMs: 0} + } + + // Alibaba Cloud TCP synthetic does not return a separate TLS phase result; + // set L2 only when L1 succeeded and the port is 443 (implies TLS negotiated). + if vr.L1.OK && target.Port == 443 { + tlsErr := "" + if result.ErrorInfo != "" { + tlsErr = result.ErrorInfo + } + vr.L2 = &L2Result{ + OK: tlsErr == "", + ALPN: "", // vendor does not expose ALPN + Err: tlsErr, + } + } + + return vr, nil +} + +// -------------------------------------------------------------------------- +// Alibaba Cloud CloudMonitor Synthetic API calls +// -------------------------------------------------------------------------- + +// aliyunTaskResult holds the vendor-specific fields returned by +// DescribeSiteMonitorData. Only fields needed for VantageResult are kept; +// vendor-specific identifiers do not escape this file. +type aliyunTaskResult struct { + // Province is the probe node's province name (in Chinese, e.g. "北京"). + Province string + // ResponseTime is the TCP round-trip time in milliseconds. + ResponseTime float64 + // Availability is 100.0 for success, 0.0 for failure. + Availability float64 + // ErrorInfo is a short vendor error string when the probe failed. + ErrorInfo string +} + +// createTask calls CreateSiteMonitor and returns the new task ID. +// address format: "host:port". +func (a *AliyunSyntheticAgent) createTask(ctx context.Context, target ProbeTarget, ispCode string) (string, error) { + ispCity, _ := json.Marshal([]map[string]string{ + {"City": a.cfg.ProbeCity, "Isp": ispCode}, + }) + + params := map[string]string{ + "Action": "CreateSiteMonitor", + "TaskName": fmt.Sprintf("pangolin-%s-%s", target.NodeID, ispCode), + "Address": fmt.Sprintf("%s:%d", target.Host, target.Port), + "TaskType": "TCP", + "Interval": "1", // one-time task + "IspCities": string(ispCity), + "AlertIds": "", + } + + body, err := a.call(ctx, params) + if err != nil { + return "", err + } + + var resp struct { + Code string `json:"Code"` + Message string `json:"Message"` + Data struct { + TaskID string `json:"TaskId"` + } `json:"Data"` + } + if err := json.Unmarshal(body, &resp); err != nil { + return "", fmt.Errorf("parse CreateSiteMonitor response: %w", err) + } + if resp.Code != "200" { + return "", fmt.Errorf("CreateSiteMonitor error code=%s msg=%s", resp.Code, resp.Message) + } + if resp.Data.TaskID == "" { + return "", fmt.Errorf("CreateSiteMonitor returned empty TaskId") + } + return resp.Data.TaskID, nil +} + +// pollResult polls DescribeSiteMonitorData until the task produces a result +// or the context / poll timeout is exceeded. +func (a *AliyunSyntheticAgent) pollResult(ctx context.Context, taskID string) (aliyunTaskResult, error) { + deadline := time.Now().Add(a.cfg.PollTimeout) + ticker := time.NewTicker(a.cfg.PollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return aliyunTaskResult{}, ctx.Err() + case t := <-ticker.C: + if t.After(deadline) { + return aliyunTaskResult{}, fmt.Errorf("poll timeout after %s", a.cfg.PollTimeout) + } + result, ready, err := a.fetchResult(ctx, taskID) + if err != nil { + return aliyunTaskResult{}, err + } + if ready { + return result, nil + } + } + } +} + +// fetchResult calls DescribeSiteMonitorData once. +// Returns (result, true, nil) when data is available, or (zero, false, nil) when +// the task is still running. +func (a *AliyunSyntheticAgent) fetchResult(ctx context.Context, taskID string) (aliyunTaskResult, bool, error) { + now := time.Now() + params := map[string]string{ + "Action": "DescribeSiteMonitorData", + "TaskId": taskID, + "MetricName": "Availability", + "Period": "60", + "StartTime": fmt.Sprintf("%d", now.Add(-2*time.Minute).Unix()*1000), + "EndTime": fmt.Sprintf("%d", now.Unix()*1000), + } + + body, err := a.call(ctx, params) + if err != nil { + return aliyunTaskResult{}, false, err + } + + var resp struct { + Code string `json:"Code"` + Message string `json:"Message"` + Data struct { + List []struct { + Maximum float64 `json:"Maximum"` + Minimum float64 `json:"Minimum"` + Average float64 `json:"Average"` + Value float64 `json:"Value"` + Cnt int `json:"Cnt"` + Province string `json:"Province"` + Isp string `json:"Isp"` + ErrorCode string `json:"ErrorCode"` + ErrorInfo string `json:"ErrorInfo"` + } `json:"List"` + } `json:"Data"` + } + + if err := json.Unmarshal(body, &resp); err != nil { + return aliyunTaskResult{}, false, fmt.Errorf("parse DescribeSiteMonitorData: %w", err) + } + if resp.Code != "200" { + return aliyunTaskResult{}, false, fmt.Errorf("DescribeSiteMonitorData error code=%s msg=%s", resp.Code, resp.Message) + } + if len(resp.Data.List) == 0 { + // Task still running — no data point yet. + return aliyunTaskResult{}, false, nil + } + + item := resp.Data.List[0] + result := aliyunTaskResult{ + Province: item.Province, + ResponseTime: item.Average, + Availability: item.Value, + ErrorInfo: item.ErrorInfo, + } + return result, true, nil +} + +// -------------------------------------------------------------------------- +// Alibaba Cloud ACS v1 signing (HMAC-SHA1) +// -------------------------------------------------------------------------- + +// call signs and executes one Alibaba Cloud CloudMonitor RPC call. +// It adds the common authentication parameters, signs the request, and +// returns the raw response body. +func (a *AliyunSyntheticAgent) call(ctx context.Context, params map[string]string) ([]byte, error) { + // Merge common authentication parameters. + p := make(map[string]string, len(params)+8) + for k, v := range params { + p[k] = v + } + p["AccessKeyId"] = a.cfg.AccessKeyID + p["SignatureMethod"] = "HMAC-SHA1" + p["SignatureNonce"] = fmt.Sprintf("%d", time.Now().UnixNano()) + p["SignatureVersion"] = "1.0" + p["Timestamp"] = time.Now().UTC().Format("2006-01-02T15:04:05Z") + p["Format"] = "JSON" + p["Version"] = "2019-01-01" + + // Build sorted query string for signing. + keys := make([]string, 0, len(p)) + for k := range p { + keys = append(keys, k) + } + sort.Strings(keys) + + var parts []string + for _, k := range keys { + parts = append(parts, percentEncode(k)+"="+percentEncode(p[k])) + } + queryStr := strings.Join(parts, "&") + + // ACS v1 string to sign: "POST\n%2F\n" + percent-encoded sorted query. + stringToSign := "POST\n%2F\n" + percentEncode(queryStr) + + // HMAC-SHA1 with key = AccessKeySecret + "&". + //nolint:gosec // required by Alibaba Cloud ACS v1 spec + mac := hmac.New(sha1.New, []byte(a.cfg.AccessKeySecret+"&")) + mac.Write([]byte(stringToSign)) + sig := base64.StdEncoding.EncodeToString(mac.Sum(nil)) + + // Build final query string including the signature. + finalQuery := queryStr + "&Signature=" + percentEncode(sig) + + reqURL := a.baseURL + "?" + finalQuery + req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, nil) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + resp, err := a.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("http call: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode == http.StatusServiceUnavailable { + return nil, fmt.Errorf("rate-limited or unavailable: HTTP %d", resp.StatusCode) + } + + rawBody, err := io.ReadAll(io.LimitReader(resp.Body, 512*1024)) + if err != nil { + return nil, fmt.Errorf("read response: %w", err) + } + return rawBody, nil +} + +// percentEncode is RFC 3986 percent-encoding as required by Alibaba Cloud ACS v1. +// It encodes all characters except unreserved ones (A-Z a-z 0-9 - _ . ~). +func percentEncode(s string) string { + var b strings.Builder + for _, c := range []byte(s) { + if isUnreserved(c) { + b.WriteByte(c) + } else { + fmt.Fprintf(&b, "%%%02X", c) + } + } + return b.String() +} + +func isUnreserved(c byte) bool { + return (c >= 'A' && c <= 'Z') || + (c >= 'a' && c <= 'z') || + (c >= '0' && c <= '9') || + c == '-' || c == '_' || c == '.' || c == '~' +} diff --git a/server/internal/scheduler/probe/prober_agent_test.go b/server/internal/scheduler/probe/prober_agent_test.go new file mode 100644 index 0000000..8432767 --- /dev/null +++ b/server/internal/scheduler/probe/prober_agent_test.go @@ -0,0 +1,337 @@ +package probe_test + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "github.com/wangjia/pangolin/server/internal/scheduler/probe" +) + +// -------------------------------------------------------------------------- +// Mock vendor server helpers +// -------------------------------------------------------------------------- + +// vendorBehavior controls what the mock Alibaba Cloud server returns. +type vendorBehavior int + +const ( + behaviorSuccess vendorBehavior = iota // normal TCP probe success + behaviorFailedProbe // probe ran but target unreachable + behaviorRateLimit // HTTP 429 + behaviorTimeout // server hangs until context cancelled +) + +// newMockAliyunServer creates an httptest.Server that behaves according to b. +// It counts how many times CreateSiteMonitor and DescribeSiteMonitorData were called. +func newMockAliyunServer(t *testing.T, b vendorBehavior) (*httptest.Server, *atomic.Int32, *atomic.Int32) { + t.Helper() + var createCalls, pollCalls atomic.Int32 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := r.ParseForm(); err != nil { + http.Error(w, "bad form", 400) + return + } + action := r.FormValue("Action") + + switch b { + case behaviorRateLimit: + w.WriteHeader(http.StatusTooManyRequests) + return + + case behaviorTimeout: + // Block until the client closes or test ends. + <-r.Context().Done() + return + } + + switch action { + case "CreateSiteMonitor": + createCalls.Add(1) + if b == behaviorSuccess || b == behaviorFailedProbe { + writeJSON(w, map[string]interface{}{ + "Code": "200", + "Message": "OK", + "RequestId": "mock-req-id", + "Data": map[string]string{"TaskId": "task-mock-001"}, + }) + } + + case "DescribeSiteMonitorData": + pollCalls.Add(1) + switch b { + case behaviorSuccess: + writeJSON(w, map[string]interface{}{ + "Code": "200", + "Message": "OK", + "Data": map[string]interface{}{ + "List": []map[string]interface{}{ + { + "Province": "北京", + "Average": float64(42), + "Value": float64(100), // 100 = available + "ErrorCode": "", + "ErrorInfo": "", + }, + }, + }, + }) + case behaviorFailedProbe: + writeJSON(w, map[string]interface{}{ + "Code": "200", + "Message": "OK", + "Data": map[string]interface{}{ + "List": []map[string]interface{}{ + { + "Province": "北京", + "Average": float64(0), + "Value": float64(0), // 0 = unavailable + "ErrorCode": "CONNECT_TIMEOUT", + "ErrorInfo": "connect timeout", + }, + }, + }, + }) + } + + default: + w.WriteHeader(http.StatusBadRequest) + } + })) + + t.Cleanup(srv.Close) + return srv, &createCalls, &pollCalls +} + +func writeJSON(w http.ResponseWriter, v interface{}) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +// newTestAgent creates an AliyunSyntheticAgent wired to a mock server and miniredis. +func newTestAgent(t *testing.T, srv *httptest.Server, mr *miniredis.Miniredis) *probe.AliyunSyntheticAgent { + t.Helper() + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + st := probe.NewStore(rdb) + + cfg := probe.AliyunSyntheticAgentConfig{ + AccessKeyID: "test-key-id", + AccessKeySecret: "test-key-secret", + PollInterval: 10 * time.Millisecond, // fast polling in tests + PollTimeout: 2 * time.Second, + ProbeCity: "563", + } + + var logBuf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logBuf, nil)) + + agent := probe.NewAliyunSyntheticAgentWithOptions(cfg, st, logger, srv.URL+"/") + return agent +} + +// -------------------------------------------------------------------------- +// Tests +// -------------------------------------------------------------------------- + +// TestProbeSuccess_NormalizesResults verifies that a successful vendor response +// is correctly mapped to VantageResults with 3rd- prefixed ISP names. +func TestProbeSuccess_NormalizesResults(t *testing.T) { + mr := miniredis.RunT(t) + srv, createCalls, _ := newMockAliyunServer(t, behaviorSuccess) + agent := newTestAgent(t, srv, mr) + + target := probe.ProbeTarget{NodeID: "node-hk-01", Host: "hk.example.com", Port: 443} + results, err := agent.Probe(context.Background(), target) + if err != nil { + t.Fatalf("Probe returned error: %v", err) + } + + // Expect one result per ISP (3 total: Telecom, Unicom, Mobile). + if len(results) != 3 { + t.Fatalf("expected 3 VantageResults, got %d", len(results)) + } + + // All ISP names must carry the "3rd-" prefix. + for _, vr := range results { + if !strings.HasPrefix(vr.Vantage.ISP, "3rd-") { + t.Errorf("ISP %q missing 3rd- prefix", vr.Vantage.ISP) + } + if vr.Vantage.Country != "CN" { + t.Errorf("Country = %q, want CN", vr.Vantage.Country) + } + } + + // Check first result details (mock returns same data for all ISPs). + vr := results[0] + if !vr.L1.OK { + t.Error("L1.OK should be true on success") + } + if vr.L1.RttMs != 42 { + t.Errorf("L1.RttMs = %d, want 42", vr.L1.RttMs) + } + if vr.L2 == nil { + t.Fatal("L2 should be non-nil for port 443") + } + if !vr.L2.OK { + t.Errorf("L2.OK should be true; Err=%q", vr.L2.Err) + } + // L3 must always be nil for 3rd-party results. + // (VantageResult has no L3 field — this is a compile-time guarantee.) + + if createCalls.Load() != 3 { + t.Errorf("expected 3 CreateSiteMonitor calls (one per ISP), got %d", createCalls.Load()) + } +} + +// TestProbeRateLimit_DegradesToNoData verifies that a 429 response causes +// the agent to return an empty slice (no data) and NOT write any Redis keys. +func TestProbeRateLimit_DegradesToNoData(t *testing.T) { + mr := miniredis.RunT(t) + srv, _, _ := newMockAliyunServer(t, behaviorRateLimit) + agent := newTestAgent(t, srv, mr) + + target := probe.ProbeTarget{NodeID: "node-sg-01", Host: "sg.example.com", Port: 443} + results, err := agent.Probe(context.Background(), target) + if err != nil { + t.Fatalf("Probe should not return error on rate-limit (degraded); got: %v", err) + } + if len(results) != 0 { + t.Errorf("expected 0 results on rate-limit, got %d", len(results)) + } + + // No Redis keys should have been written. + keys := mr.Keys() + for _, k := range keys { + if strings.HasPrefix(k, "probe:node-sg-01") { + t.Errorf("unexpected Redis key written during degradation: %q", k) + } + } +} + +// TestProbeTimeout_DegradesToNoData verifies that a hung vendor server (simulating +// timeout) causes degradation: empty results, no Redis writes. +func TestProbeTimeout_DegradesToNoData(t *testing.T) { + mr := miniredis.RunT(t) + srv, _, _ := newMockAliyunServer(t, behaviorTimeout) + + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + st := probe.NewStore(rdb) + cfg := probe.AliyunSyntheticAgentConfig{ + AccessKeyID: "test-key-id", + AccessKeySecret: "test-key-secret", + PollInterval: 10 * time.Millisecond, + PollTimeout: 100 * time.Millisecond, // very short for test + } + // Use a short HTTP client timeout too. + agent := probe.NewAliyunSyntheticAgentWithOptions(cfg, st, slog.Default(), srv.URL+"/") + _ = probe.SetHTTPClientTimeout(agent, 50*time.Millisecond) + + target := probe.ProbeTarget{NodeID: "node-jp-01", Host: "jp.example.com", Port: 443} + + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + + results, err := agent.Probe(ctx, target) + if err != nil { + t.Fatalf("Probe should not propagate timeout error; got: %v", err) + } + if len(results) != 0 { + t.Errorf("expected 0 results on timeout, got %d", len(results)) + } + + // No Redis keys for this node. + for _, k := range mr.Keys() { + if strings.HasPrefix(k, "probe:node-jp-01") { + t.Errorf("unexpected Redis key on timeout: %q", k) + } + } +} + +// TestRunOnce_WritesRedisKeys verifies that RunOnce writes probe:{node}:3rd-* keys +// to Redis for each ISP after a successful probe cycle. +func TestRunOnce_WritesRedisKeys(t *testing.T) { + mr := miniredis.RunT(t) + srv, _, _ := newMockAliyunServer(t, behaviorSuccess) + agent := newTestAgent(t, srv, mr) + + targets := []probe.ProbeTarget{ + {NodeID: "node-hk-02", Host: "hk2.example.com", Port: 443}, + } + if err := agent.RunOnce(context.Background(), targets); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + // Expect 3 keys: one per ISP, each with "3rd-" in the vantage segment. + var thirdPartyKeys []string + for _, k := range mr.Keys() { + if strings.HasPrefix(k, "probe:node-hk-02:") && strings.Contains(k, "3rd-") { + thirdPartyKeys = append(thirdPartyKeys, k) + } + } + if len(thirdPartyKeys) != 3 { + t.Errorf("expected 3 probe:node-hk-02:...:3rd-* keys, got %d: %v", len(thirdPartyKeys), thirdPartyKeys) + } + + // Verify key format: probe:{node}:{country}:{region}:{isp} + // e.g. probe:node-hk-02:CN:北京:3rd-ChinaTelecom (spaces become underscores) + for _, k := range thirdPartyKeys { + parts := strings.Split(k, ":") + // parts[0]=probe, parts[1]=node-hk-02, parts[2]=CN, parts[3]=region, parts[4]=3rd-... + if len(parts) < 5 { + t.Errorf("unexpected key format: %q", k) + continue + } + if !strings.HasPrefix(parts[4], "3rd-") { + t.Errorf("ISP segment %q missing 3rd- prefix in key %q", parts[4], k) + } + } +} + +// TestProbeFailedTarget_L1NotOK verifies that when the vendor reports the target +// is unreachable (Availability=0), L1.OK is false and no L2 is populated. +func TestProbeFailedTarget_L1NotOK(t *testing.T) { + mr := miniredis.RunT(t) + srv, _, _ := newMockAliyunServer(t, behaviorFailedProbe) + agent := newTestAgent(t, srv, mr) + + target := probe.ProbeTarget{NodeID: "node-fail-01", Host: "fail.example.com", Port: 443} + results, err := agent.Probe(context.Background(), target) + if err != nil { + t.Fatalf("Probe: %v", err) + } + if len(results) != 3 { + t.Fatalf("expected 3 results even on target failure, got %d", len(results)) + } + for _, vr := range results { + if vr.L1.OK { + t.Errorf("ISP %s: L1.OK should be false for unreachable target", vr.Vantage.ISP) + } + if vr.L2 != nil { + t.Errorf("ISP %s: L2 should be nil when L1 failed", vr.Vantage.ISP) + } + } +} + +// TestNoPlaintextCredentials is a compile-time reminder that credentials +// must not be hardcoded. The test itself only validates that the agent +// accepts empty credentials (real validation is at runtime via env vars). +func TestNoPlaintextCredentials(t *testing.T) { + cfg := probe.AliyunSyntheticAgentConfig{ + AccessKeyID: "", // must come from env, not code + AccessKeySecret: "", // must come from env, not code + } + // A zero-credential config is valid to construct; errors surface on actual calls. + if cfg.AccessKeyID != "" || cfg.AccessKeySecret != "" { + t.Error("credentials must not be hardcoded in config defaults") + } +}