Initial commit

agent/jboard-agent/internal/probe/latency.go

@@ -0,0 +1,171 @@
+package probe
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"log"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/jboard/jboard-agent/internal/config"
+)
+
+// Three-carrier TCP ping targets (Chinese ISP backbone nodes)
+var latencyTargets = []struct {
+	Carrier string
+	Host    string
+	Port    string
+}{
+	{"mobile", "js-cm-v4.ip.zstaticcdn.com", "80"},
+	{"unicom", "js-cu-v4.ip.zstaticcdn.com", "80"},
+	{"telecom", "js-ct-v4.ip.zstaticcdn.com", "80"},
+}
+
+type latencyEntry struct {
+	Carrier   string `json:"carrier"`
+	LatencyMs int    `json:"latencyMs"`
+}
+
+type latencyPayload struct {
+	Latencies []latencyEntry `json:"latencies"`
+}
+
+// LatencyLoop periodically measures TCP ping latency to three carriers and pushes to J-Board.
+func LatencyLoop(ctx context.Context, cfg *config.Config) {
+	ticker := time.NewTicker(cfg.LatencyInterval)
+	defer ticker.Stop()
+
+	// Run immediately
+	measureAndPush(cfg)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			measureAndPush(cfg)
+		}
+	}
+}
+
+func measureAndPush(cfg *config.Config) {
+	var entries []latencyEntry
+
+	for _, target := range latencyTargets {
+		ms := tcpPing(target.Host, target.Port)
+		if ms >= 0 {
+			entries = append(entries, latencyEntry{
+				Carrier:   target.Carrier,
+				LatencyMs: ms,
+			})
+			log.Printf("[latency] %s: %dms", target.Carrier, ms)
+		} else {
+			log.Printf("[latency] %s: timeout", target.Carrier)
+		}
+	}
+
+	if len(entries) == 0 {
+		return
+	}
+
+	payload := latencyPayload{Latencies: entries}
+	body, _ := json.Marshal(payload)
+
+	if err := postToServer(cfg, "/api/agent/latency", body); err != nil {
+		log.Printf("[latency] push error: %v", err)
+	}
+}
+
+// tcpPing measures TCP handshake latency in milliseconds. Returns -1 on failure.
+// The DNS lookup is intentionally performed before timing starts, matching
+// classic probe panels such as Komari, so DNS jitter is not mixed into latency.
+func tcpPing(host, port string) int {
+	const (
+		timeout              = 3 * time.Second
+		highLatencyThreshold = 1000
+		highLatencyRetries   = 3
+	)
+
+	ip, err := resolveIP(host)
+	if err != nil {
+		return -1
+	}
+
+	latency, err := measureTCPConnect(ip, port, timeout)
+	if err != nil {
+		return -1
+	}
+	best := latency
+
+	if latency > highLatencyThreshold {
+		for i := 0; i < highLatencyRetries; i++ {
+			retryLatency, retryErr := measureTCPConnect(ip, port, timeout)
+			if retryErr != nil {
+				continue
+			}
+			if retryLatency < best {
+				best = retryLatency
+			}
+			if retryLatency <= highLatencyThreshold {
+				break
+			}
+		}
+	}
+
+	return best
+}
+
+func resolveIP(host string) (string, error) {
+	if ip := net.ParseIP(host); ip != nil {
+		return host, nil
+	}
+
+	addrs, err := net.LookupHost(host)
+	if err != nil || len(addrs) == 0 {
+		return "", errors.New("failed to resolve target")
+	}
+
+	return addrs[0], nil
+}
+
+func measureTCPConnect(ip string, port string, timeout time.Duration) (int, error) {
+	start := time.Now()
+	conn, err := net.DialTimeout("tcp", net.JoinHostPort(ip, port), timeout)
+	if err != nil {
+		return -1, err
+	}
+	conn.Close()
+	return int(time.Since(start).Milliseconds()), nil
+}
+
+func postToServer(cfg *config.Config, path string, body []byte) error {
+	req, err := http.NewRequest("POST", cfg.ServerURL+path, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+cfg.AuthToken)
+
+	client := &http.Client{Timeout: 15 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return &httpError{StatusCode: resp.StatusCode}
+	}
+	return nil
+}
+
+type httpError struct {
+	StatusCode int
+}
+
+func (e *httpError) Error() string {
+	return "server returned " + http.StatusText(e.StatusCode)
+}