luobei/mmw-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

格式化

Browse Source
This commit is contained in:
iluobei
2026-04-10 15:25:21 +08:00
parent e13c81ce7a
commit b3202edb49
17 changed files with 633 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/agent/client.go
+183 -110
View File
@@ -21,11 +21,12 @@ import (
"mmw-agent/internal/collector"
"mmw-agent/internal/config"
"mmw-agent/internal/constants"
"github.com/gorilla/websocket"
)
// ConnectionMode represents the current connection mode
// ConnectionMode 表示当前连接模式。
type ConnectionMode string
const (
@@ -35,7 +36,7 @@ const (
ModeAuto ConnectionMode = "auto"
)
// Client represents an agent client that connects to a master server
// Client 表示连接主控端的 agent 客户端。
type Client struct {
config *config.Config
collector *collector.Collector
@@ -47,20 +48,20 @@ type Client struct {
stopCh chan struct{}
wg sync.WaitGroup
// Connection state
// 连接状态
currentMode ConnectionMode
httpClient *http.Client
httpAvailable bool
modeMu sync.RWMutex
// Speed calculation (from system network interface)
// 速率计算（基于系统网卡统计）
lastRxBytes int64
lastTxBytes int64
lastSampleTime time.Time
speedMu sync.Mutex
}
// NewClient creates a new agent client
// 创建 agent 客户端。
func NewClient(cfg *config.Config) *Client {
return &Client{
config: cfg,
@@ -68,20 +69,20 @@ func NewClient(cfg *config.Config) *Client {
xrayServers: cfg.XrayServers,
stopCh: make(chan struct{}),
httpClient: &http.Client{
Timeout: 10 * time.Second,
Timeout: constants.DefaultHTTPClientTimeout,
},
currentMode: ModePull, // Default to pull mode
currentMode: ModePull, // 默认使用拉取模式
}
}
// wsHeaders returns HTTP headers for WebSocket handshake
// 生成 WebSocket 握手请求头。
func (c *Client) wsHeaders() http.Header {
h := http.Header{}
h.Set("User-Agent", config.AgentUserAgent)
h.Set(constants.HeaderUserAgent, constants.AgentUserAgent)
return h
}
// newRequest creates an HTTP request with standard headers (Content-Type, Authorization, User-Agent)
// 创建带标准请求头的 HTTP 请求。
func (c *Client) newRequest(ctx context.Context, method, urlStr string, body []byte) (*http.Request, error) {
var req *http.Request
var err error
@@ -93,13 +94,13 @@ func (c *Client) newRequest(ctx context.Context, method, urlStr string, body []b
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+c.config.Token)
req.Header.Set("User-Agent", config.AgentUserAgent)
req.Header.Set(constants.HeaderContentType, constants.ContentTypeJSON)
req.Header.Set(constants.HeaderAuthorization, constants.BearerPrefix+c.config.Token)
req.Header.Set(constants.HeaderUserAgent, constants.AgentUserAgent)
return req, nil
}
// Start starts the agent client with automatic mode selection
// 按配置启动客户端。
func (c *Client) Start(ctx context.Context) {
log.Printf("[Agent] Starting in %s mode", c.config.ConnectionMode)
@@ -117,7 +118,7 @@ func (c *Client) Start(ctx context.Context) {
case ModePull:
c.setCurrentMode(ModePull)
log.Printf("[Agent] Pull mode enabled - API will be served at /api/child/traffic and /api/child/speed")
// Report agent info immediately via HTTP heartbeat
// 启动后先通过 HTTP 上报一次心跳信息
if err := c.sendHeartbeatHTTP(ctx); err != nil {
log.Printf("[Agent] Failed to send initial heartbeat in pull mode: %v", err)
}
@@ -130,7 +131,7 @@ func (c *Client) Start(ctx context.Context) {
}
}
// Stop stops the agent client
// 停止客户端。
func (c *Client) Stop() {
close(c.stopCh)
c.wg.Wait()
@@ -144,33 +145,33 @@ func (c *Client) Stop() {
log.Printf("[Agent] Stopped")
}
// IsConnected returns whether the WebSocket is connected
// 返回 WebSocket 连接状态。
func (c *Client) IsConnected() bool {
c.wsMu.Lock()
defer c.wsMu.Unlock()
return c.connected
}
// GetCurrentMode returns the current connection mode
// 返回当前连接模式。
func (c *Client) GetCurrentMode() ConnectionMode {
c.modeMu.RLock()
defer c.modeMu.RUnlock()
return c.currentMode
}
// setCurrentMode sets the current connection mode
// 设置当前连接模式。
func (c *Client) setCurrentMode(mode ConnectionMode) {
c.modeMu.Lock()
defer c.modeMu.Unlock()
c.currentMode = mode
}
// runWebSocket manages the WebSocket connection lifecycle with fallback to auto mode
// 维护 WebSocket 连接，并在失败时回退自动模式。
func (c *Client) runWebSocket(ctx context.Context) {
defer c.wg.Done()
maxConsecutiveFailures := 5
maxAuthFailures := 10
maxConsecutiveFailures := constants.WebSocketMaxConsecutiveFailures
maxAuthFailures := constants.WebSocketMaxAuthFailures
consecutiveFailures := 0
authFailures := 0
@@ -190,22 +191,22 @@ func (c *Client) runWebSocket(ctx context.Context) {
return
}
// Check if this is an authentication error
// 判断是否为鉴权错误
if authErr, ok := err.(*AuthError); ok {
authFailures++
if authErr.IsTokenInvalid() {
log.Printf("[Agent] Authentication failed (invalid token): %v", err)
if authFailures >= maxAuthFailures {
log.Printf("[Agent] Too many auth failures (%d), entering sleep mode (30 min backoff)", authFailures)
c.waitWithTrafficReport(ctx, 30*time.Minute)
c.waitWithTrafficReport(ctx, constants.AuthFailureSleepBackoff)
authFailures = 0
continue
}
}
// Use longer backoff for auth errors
backoff := time.Duration(authFailures) * 30 * time.Second
if backoff > 10*time.Minute {
backoff = 10 * time.Minute
// 鉴权错误使用更长退避时间
backoff := time.Duration(authFailures) * constants.AuthFailureBackoffStep
if backoff > constants.AuthFailureMaxBackoff {
backoff = constants.AuthFailureMaxBackoff
}
log.Printf("[Agent] Auth error, reconnecting in %v...", backoff)
c.waitWithTrafficReport(ctx, backoff)
@@ -233,21 +234,21 @@ func (c *Client) runWebSocket(ctx context.Context) {
}
}
// calculateBackoff calculates the reconnection backoff duration with exponential increase
// 计算重连退避时长。
func (c *Client) calculateBackoff() time.Duration {
c.reconnects++
// Exponential backoff: 5s, 10s, 20s, 40s, 80s, 160s, 300s(cap)
backoff := 5 * time.Second
for i := 1; i < c.reconnects && backoff < 5*time.Minute; i++ {
// 指数退避: 5s, 10s, 20s, 40s, 80s, 160s, 300s(上限)
backoff := constants.ReconnectBaseBackoff
for i := 1; i < c.reconnects && backoff < constants.ReconnectMaxBackoff; i++ {
backoff *= 2
}
if backoff > 5*time.Minute {
backoff = 5 * time.Minute
if backoff > constants.ReconnectMaxBackoff {
backoff = constants.ReconnectMaxBackoff
}
return backoff
}
// connectAndRun establishes and maintains a WebSocket connection
// 建立并维持 WebSocket 连接。
func (c *Client) connectAndRun(ctx context.Context) error {
masterURL := c.config.MasterURL
u, err := url.Parse(masterURL)
@@ -262,12 +263,12 @@ func (c *Client) connectAndRun(ctx context.Context) error {
u.Scheme = "wss"
}
u.Path = "/api/remote/ws"
u.Path = constants.PathRemoteWebSocket
log.Printf("[Agent] Connecting to %s", u.String())
dialer := websocket.Dialer{
HandshakeTimeout: 10 * time.Second,
HandshakeTimeout: constants.WebSocketHandshakeTimeout,
}
conn, _, err := dialer.DialContext(ctx, u.String(), c.wsHeaders())
@@ -298,18 +299,18 @@ func (c *Client) connectAndRun(ctx context.Context) error {
log.Printf("[Agent] Connected and authenticated")
// Report agent info (listen_port) immediately after connection
// 连接成功后立即上报 agent 信息（listen_port）
if err := c.sendHeartbeat(conn); err != nil {
log.Printf("[Agent] Failed to send initial heartbeat: %v", err)
}
// Send scan result to master for auto-sync
// 异步上报扫描结果，供主控端自动同步
go c.sendScanResult(conn)
return c.runMessageLoop(ctx, conn)
}
// authenticate sends the authentication message
// 发送鉴权消息。
func (c *Client) authenticate(conn *websocket.Conn) error {
authPayload, _ := json.Marshal(map[string]string{
"token": c.config.Token,
@@ -324,7 +325,7 @@ func (c *Client) authenticate(conn *websocket.Conn) error {
return err
}
conn.SetReadDeadline(time.Now().Add(10 * time.Second))
conn.SetReadDeadline(time.Now().Add(constants.WebSocketReadDeadline))
_, message, err := conn.ReadMessage()
if err != nil {
return err
@@ -349,11 +350,11 @@ func (c *Client) authenticate(conn *websocket.Conn) error {
return nil
}
// runMessageLoop handles sending traffic data, speed data, and heartbeats
// 处理流量、速率和心跳上报。
func (c *Client) runMessageLoop(ctx context.Context, conn *websocket.Conn) error {
trafficTicker := time.NewTicker(c.config.TrafficReportInterval)
speedTicker := time.NewTicker(c.config.SpeedReportInterval)
heartbeatTicker := time.NewTicker(30 * time.Second)
heartbeatTicker := time.NewTicker(constants.WebSocketHeartbeatInterval)
defer trafficTicker.Stop()
defer speedTicker.Stop()
defer heartbeatTicker.Stop()
@@ -362,13 +363,13 @@ func (c *Client) runMessageLoop(ctx context.Context, conn *websocket.Conn) error
errCh := make(chan error, 1)
go func() {
for {
conn.SetReadDeadline(time.Now().Add(5 * time.Minute))
conn.SetReadDeadline(time.Now().Add(constants.WebSocketIdleDeadline))
_, message, err := conn.ReadMessage()
if err != nil {
errCh <- err
return
}
// Send message to processing channel
// 投递到消息处理通道
select {
case msgCh <- message:
default:
@@ -406,7 +407,7 @@ func (c *Client) runMessageLoop(ctx context.Context, conn *websocket.Conn) error
}
}
// sendTrafficData collects and sends traffic data to the master
// 采集并发送流量数据。
func (c *Client) sendTrafficData(conn *websocket.Conn) error {
stats, err := c.collectLocalMetrics()
if err != nil {
@@ -437,7 +438,7 @@ func (c *Client) sendTrafficData(conn *websocket.Conn) error {
return nil
}
// sendHeartbeat sends a heartbeat message
// 发送心跳消息。
func (c *Client) sendHeartbeat(conn *websocket.Conn) error {
now := time.Now()
listenPort, _ := strconv.Atoi(c.config.ListenPort)
@@ -458,7 +459,7 @@ func (c *Client) sendHeartbeat(conn *websocket.Conn) error {
return err
}
// collectLocalMetrics collects traffic metrics from local Xray servers
// 采集本机 Xray 流量指标。
func (c *Client) collectLocalMetrics() (*collector.XrayStats, error) {
stats := &collector.XrayStats{
Inbound: make(map[string]collector.TrafficData),
@@ -487,23 +488,23 @@ func (c *Client) collectLocalMetrics() (*collector.XrayStats, error) {
return stats, nil
}
// GetStats returns the current traffic stats (for pull mode)
// 返回当前流量统计（拉取模式）。
func (c *Client) GetStats() (*collector.XrayStats, error) {
return c.collectLocalMetrics()
}
// GetSpeed returns the current speed data (for pull mode)
// 返回当前速率（拉取模式）。
func (c *Client) GetSpeed() (uploadSpeed, downloadSpeed int64) {
return c.collectSpeed()
}
// runAutoMode implements the three-tier fallback: WebSocket -> HTTP -> Pull
// 使用三层回退：WebSocket -> HTTP -> Pull。
func (c *Client) runAutoMode(ctx context.Context) {
defer c.wg.Done()
c.runAutoModeLoop(ctx)
}
// runAutoModeLoop is the internal loop for auto mode fallback
// 是自动模式的内部循环。
func (c *Client) runAutoModeLoop(ctx context.Context) {
autoRetries := 0
for {
@@ -548,14 +549,14 @@ func (c *Client) runAutoModeLoop(ctx context.Context) {
log.Printf("[Agent] Falling back to pull mode - API available at /api/child/traffic and /api/child/speed")
c.sendHeartbeatHTTP(ctx)
// Exponential backoff for pull mode: 30s, 60s, 120s, 240s, 300s(cap)
// 拉取模式退避: 30s, 60s, 120s, 240s, 300s(上限)
autoRetries++
pullDuration := 30 * time.Second
for i := 1; i < autoRetries && pullDuration < 5*time.Minute; i++ {
pullDuration := constants.AutoModePullFallbackBackoff
for i := 1; i < autoRetries && pullDuration < constants.ReconnectMaxBackoff; i++ {
pullDuration *= 2
}
if pullDuration > 5*time.Minute {
pullDuration = 5 * time.Minute
if pullDuration > constants.ReconnectMaxBackoff {
pullDuration = constants.ReconnectMaxBackoff
}
c.runPullModeWithTrafficReport(ctx, pullDuration)
@@ -567,7 +568,7 @@ func (c *Client) runAutoModeLoop(ctx context.Context) {
}
}
// tryWebSocketOnce attempts a single WebSocket connection test
// 执行一次 WebSocket 可用性探测。
func (c *Client) tryWebSocketOnce(ctx context.Context) error {
masterURL := c.config.MasterURL
u, err := url.Parse(masterURL)
@@ -581,10 +582,10 @@ func (c *Client) tryWebSocketOnce(ctx context.Context) error {
case "https":
u.Scheme = "wss"
}
u.Path = "/api/remote/ws"
u.Path = constants.PathRemoteWebSocket
dialer := websocket.Dialer{
HandshakeTimeout: 10 * time.Second,
HandshakeTimeout: constants.WebSocketHandshakeTimeout,
}
conn, _, err := dialer.DialContext(ctx, u.String(), c.wsHeaders())
@@ -595,13 +596,13 @@ func (c *Client) tryWebSocketOnce(ctx context.Context) error {
return nil
}
// tryHTTPOnce tests if HTTP push is available
// 探测 HTTP 推送是否可用。
func (c *Client) tryHTTPOnce(ctx context.Context) bool {
u, err := url.Parse(c.config.MasterURL)
if err != nil {
return false
}
u.Path = "/api/remote/heartbeat"
u.Path = constants.PathRemoteHeartbeat
req, err := c.newRequest(ctx, http.MethodPost, u.String(), []byte("{}"))
if err != nil {
@@ -619,18 +620,18 @@ func (c *Client) tryHTTPOnce(ctx context.Context) bool {
return c.httpAvailable
}
// runHTTPReporter runs the HTTP push reporter
// 运行 HTTP 推送上报器。
func (c *Client) runHTTPReporter(ctx context.Context) {
defer c.wg.Done()
c.setCurrentMode(ModeHTTP)
c.runHTTPReporterLoop(ctx)
}
// runHTTPReporterLoop runs the HTTP reporting loop
// 执行 HTTP 上报循环。
func (c *Client) runHTTPReporterLoop(ctx context.Context) {
trafficTicker := time.NewTicker(c.config.TrafficReportInterval)
speedTicker := time.NewTicker(c.config.SpeedReportInterval)
heartbeatTicker := time.NewTicker(30 * time.Second)
heartbeatTicker := time.NewTicker(constants.WebSocketHeartbeatInterval)
defer trafficTicker.Stop()
defer speedTicker.Stop()
defer heartbeatTicker.Stop()
@@ -676,7 +677,7 @@ func (c *Client) runHTTPReporterLoop(ctx context.Context) {
}
}
// sendTrafficHTTP sends traffic data via HTTP POST
// 通过 HTTP POST 发送流量数据。
func (c *Client) sendTrafficHTTP(ctx context.Context) error {
stats, err := c.collectLocalMetrics()
if err != nil {
@@ -691,7 +692,7 @@ func (c *Client) sendTrafficHTTP(ctx context.Context) error {
if err != nil {
return err
}
u.Path = "/api/remote/traffic"
u.Path = constants.PathRemoteTraffic
req, err := c.newRequest(ctx, http.MethodPost, u.String(), payload)
if err != nil {
@@ -714,7 +715,7 @@ func (c *Client) sendTrafficHTTP(ctx context.Context) error {
return nil
}
// sendSpeedHTTP sends speed data via HTTP POST
// 通过 HTTP POST 发送速率数据。
func (c *Client) sendSpeedHTTP(ctx context.Context) error {
uploadSpeed, downloadSpeed := c.collectSpeed()
@@ -727,7 +728,7 @@ func (c *Client) sendSpeedHTTP(ctx context.Context) error {
if err != nil {
return err
}
u.Path = "/api/remote/speed"
u.Path = constants.PathRemoteSpeed
req, err := c.newRequest(ctx, http.MethodPost, u.String(), payload)
if err != nil {
@@ -749,7 +750,7 @@ func (c *Client) sendSpeedHTTP(ctx context.Context) error {
return nil
}
// sendHeartbeatHTTP sends heartbeat via HTTP POST
// 通过 HTTP POST 发送心跳。
func (c *Client) sendHeartbeatHTTP(ctx context.Context) error {
now := time.Now()
listenPort, _ := strconv.Atoi(c.config.ListenPort)
@@ -762,7 +763,7 @@ func (c *Client) sendHeartbeatHTTP(ctx context.Context) error {
if err != nil {
return err
}
u.Path = "/api/remote/heartbeat"
u.Path = constants.PathRemoteHeartbeat
req, err := c.newRequest(ctx, http.MethodPost, u.String(), payload)
if err != nil {
@@ -783,7 +784,7 @@ func (c *Client) sendHeartbeatHTTP(ctx context.Context) error {
return nil
}
// runPullModeWithTrafficReport runs pull mode while sending traffic data to keep server online
// 在拉取模式下持续上报流量，保持在线状态。
func (c *Client) runPullModeWithTrafficReport(ctx context.Context, duration time.Duration) {
trafficTicker := time.NewTicker(c.config.TrafficReportInterval)
defer trafficTicker.Stop()
@@ -810,13 +811,13 @@ func (c *Client) runPullModeWithTrafficReport(ctx context.Context, duration time
}
}
// waitWithTrafficReport waits for the specified duration while sending traffic data
// 在等待期间继续上报流量。
func (c *Client) waitWithTrafficReport(ctx context.Context, duration time.Duration) {
if duration <= 0 {
return
}
if duration > 30*time.Second {
if duration > constants.PullModeTrafficReportThreshold {
if err := c.sendTrafficHTTP(ctx); err != nil {
log.Printf("[Agent] Traffic report during backoff failed: %v", err)
}
@@ -843,7 +844,7 @@ func (c *Client) waitWithTrafficReport(ctx context.Context, duration time.Durati
}
}
// sendSpeedData sends speed data via WebSocket
// 通过 WebSocket 发送速率数据。
func (c *Client) sendSpeedData(conn *websocket.Conn) error {
uploadSpeed, downloadSpeed := c.collectSpeed()
@@ -869,7 +870,7 @@ func (c *Client) sendSpeedData(conn *websocket.Conn) error {
return nil
}
// collectSpeed calculates the current upload and download speed from system network interface
// 基于系统网卡统计计算当前上下行速率。
func (c *Client) collectSpeed() (uploadSpeed, downloadSpeed int64) {
c.speedMu.Lock()
defer c.speedMu.Unlock()
@@ -900,7 +901,7 @@ func (c *Client) collectSpeed() (uploadSpeed, downloadSpeed int64) {
return uploadSpeed, downloadSpeed
}
// getSystemNetworkStats reads network statistics from /proc/net/dev
// 从 /proc/net/dev 读取网卡统计。
func (c *Client) getSystemNetworkStats() (rxBytes, txBytes int64) {
data, err := os.ReadFile("/proc/net/dev")
if err != nil {
@@ -936,7 +937,7 @@ func (c *Client) getSystemNetworkStats() (rxBytes, txBytes int64) {
return rxBytes, txBytes
}
// AuthError represents an authentication error
// AuthError 表示鉴权失败错误。
type AuthError struct {
Message string
Code string // "token_expired", "token_invalid", "server_error"
@@ -946,12 +947,12 @@ func (e *AuthError) Error() string {
return "authentication failed: " + e.Message
}
// IsTokenInvalid returns true if the error indicates an invalid token
// 判断是否为 token 无效错误。
func (e *AuthError) IsTokenInvalid() bool {
return e.Code == "token_invalid" || e.Message == "Invalid token"
}
// WebSocket message types
// WebSocket 消息类型
const (
WSMsgTypeCertDeploy = "cert_deploy"
WSMsgTypeTokenUpdate = "token_update"
@@ -960,7 +961,7 @@ const (
WSMsgTypeDomainLatencyResult = "domain_latency_result"
)
// WSCertDeployPayload represents a certificate deploy command from master
// WSCertDeployPayload 是主控端下发的证书部署指令。
type WSCertDeployPayload struct {
Domain string `json:"domain"`
CertPEM string `json:"cert_pem"`
@@ -970,20 +971,20 @@ type WSCertDeployPayload struct {
Reload string `json:"reload"`
}
// WSTokenUpdatePayload represents a token update from master
// WSTokenUpdatePayload 是主控端下发的 token 更新指令。
type WSTokenUpdatePayload struct {
ServerToken string `json:"server_token"`
ExpiresAt time.Time `json:"expires_at"`
}
// WSDomainLatencyProbePayload is received from master
// WSDomainLatencyProbePayload 是主控端下发的域名延迟探测请求。
type WSDomainLatencyProbePayload struct {
RequestID string `json:"request_id"`
Domains []string `json:"domains"`
TimeoutMs int `json:"timeout_ms"`
}
// handleMessage processes incoming messages from master
// 处理主控端下发的消息。
func (c *Client) handleMessage(conn *websocket.Conn, message []byte) {
var msg struct {
Type string `json:"type"`
@@ -1018,11 +1019,11 @@ func (c *Client) handleMessage(conn *websocket.Conn, message []byte) {
}
go c.handleDomainLatencyProbe(conn, payload)
default:
// Ignore unknown message types
// 忽略未知消息类型
}
}
// handleCertDeploy deploys a certificate received from master
// 处理主控端下发的证书部署。
func (c *Client) handleCertDeploy(payload WSCertDeployPayload) {
log.Printf("[Agent] Received cert_deploy for domain: %s, target: %s", payload.Domain, payload.Reload)
@@ -1065,7 +1066,7 @@ func deployCert(certPEM, keyPEM, certPath, keyPath, reloadTarget string) error {
}
func reloadNginxCmd() error {
for _, bin := range []string{"/usr/local/nginx/sbin/nginx", "nginx"} {
for _, bin := range constants.NginxBinarySearchPaths {
if path, err := exec.LookPath(bin); err == nil {
return runCmd(path, "-s", "reload")
}
@@ -1080,43 +1081,47 @@ func runCmd(name string, args ...string) error {
return nil
}
// handleTokenUpdate processes a token update from master
// 处理主控端下发的 token 更新。
func (c *Client) handleTokenUpdate(payload WSTokenUpdatePayload) {
log.Printf("[Agent] Received token update from master, new token expires at %s", payload.ExpiresAt.Format(time.RFC3339))
// Update the token in memory
// 更新内存中的 token
c.config.Token = payload.ServerToken
log.Printf("[Agent] Token updated successfully in memory")
}
// handleDomainLatencyProbe probes domain latency locally and sends results back via WebSocket
// 在本机探测域名延迟并回传结果。
func (c *Client) handleDomainLatencyProbe(conn *websocket.Conn, payload WSDomainLatencyProbePayload) {
log.Printf("[Agent] Received domain_latency_probe: %d domains, timeout=%dms", len(payload.Domains), payload.TimeoutMs)
timeoutMs := payload.TimeoutMs
if timeoutMs <= 0 {
timeoutMs = 2000
timeoutMs = constants.DomainProbeDefaultTimeoutMS
}
if timeoutMs < 200 {
timeoutMs = 200
if timeoutMs < constants.DomainProbeMinTimeoutMS {
timeoutMs = constants.DomainProbeMinTimeoutMS
}
if timeoutMs > 10000 {
timeoutMs = 10000
if timeoutMs > constants.DomainProbeMaxTimeoutMS {
timeoutMs = constants.DomainProbeMaxTimeoutMS
}
timeout := time.Duration(timeoutMs) * time.Millisecond
type probeResult struct {
Domain string `json:"domain"`
Target string `json:"target"`
Success bool `json:"success"`
LatencyMs int64 `json:"latency_ms,omitempty"`
Error string `json:"error,omitempty"`
Domain string `json:"domain"`
Target string `json:"target"`
Success bool `json:"success"`
LatencyMs int64 `json:"latency_ms,omitempty"`
Error string `json:"error,omitempty"`
NginxSSLPort int `json:"nginx_ssl_port,omitempty"`
}
// 读取本机 nginx 配置，构造 domain -> ssl 端口映射
nginxPortMap := readNginxSSLPorts(payload.Domains)
results := make([]probeResult, 0, len(payload.Domains))
resultCh := make(chan probeResult, len(payload.Domains))
sem := make(chan struct{}, 16)
sem := make(chan struct{}, constants.DomainProbeConcurrency)
var wg sync.WaitGroup
for _, domain := range payload.Domains {
@@ -1145,7 +1150,7 @@ func (c *Client) handleDomainLatencyProbe(conn *websocket.Conn, payload WSDomain
return
}
_ = tcpConn.Close()
resultCh <- probeResult{Domain: host, Target: target, Success: true, LatencyMs: time.Since(start).Milliseconds()}
resultCh <- probeResult{Domain: host, Target: target, Success: true, LatencyMs: time.Since(start).Milliseconds(), NginxSSLPort: nginxPortMap[host]}
}()
}
@@ -1155,7 +1160,7 @@ func (c *Client) handleDomainLatencyProbe(conn *websocket.Conn, payload WSDomain
results = append(results, r)
}
// Sort: success first, then by latency
// 排序：成功优先，再按延迟升序
sort.Slice(results, func(i, j int) bool {
if results[i].Success != results[j].Success {
return results[i].Success
@@ -1201,9 +1206,80 @@ func (c *Client) handleDomainLatencyProbe(conn *websocket.Conn, payload WSDomain
log.Printf("[Agent] Sent domain_latency_result: %d results", len(results))
}
// sendScanResult scans local xray status and sends results to master
// readNginxSSLPorts 读取 nginx 配置并返回 domain -> SSL 端口映射。
// 会在常见 nginx 配置目录下查找 servers/{domain}.conf。
func readNginxSSLPorts(domains []string) map[string]int {
result := make(map[string]int)
if len(domains) == 0 {
return result
}
confDirs := constants.NginxSSLServerDirPaths
for _, domain := range domains {
host := domain
if h, _, err := net.SplitHostPort(domain); err == nil && h != "" {
host = h
}
for _, dir := range confDirs {
confPath := filepath.Join(dir, host+".conf")
data, err := os.ReadFile(confPath)
if err != nil {
continue
}
if port := extractSSLListenPort(string(data)); port > 0 {
result[host] = port
break
}
}
}
return result
}
// 提取 nginx 配置块中第一个 "listen <port> ssl" 端口。
func extractSSLListenPort(conf string) int {
// 匹配示例: listen 58443 ssl
for _, line := range strings.Split(conf, "\n") {
line = strings.TrimSpace(line)
if !strings.HasPrefix(line, "listen") {
continue
}
// 去掉 "listen" 前缀和结尾分号
rest := strings.TrimPrefix(line, "listen")
rest = strings.TrimRight(rest, ";")
rest = strings.TrimSpace(rest)
fields := strings.Fields(rest)
if len(fields) < 2 {
continue
}
// 判断字段中是否包含 "ssl"
hasSSL := false
for _, f := range fields[1:] {
if f == "ssl" {
hasSSL = true
break
}
}
if !hasSSL {
continue
}
// 第一个字段是端口（或 [::]:port）
portStr := fields[0]
// 兼容 [::]:port 形式
if idx := strings.LastIndex(portStr, ":"); idx >= 0 {
portStr = portStr[idx+1:]
}
port, err := strconv.Atoi(portStr)
if err == nil && port > 0 {
return port
}
}
return 0
}
// 扫描本机 xray 状态并上报主控端。
func (c *Client) sendScanResult(conn *websocket.Conn) {
// Check xray running status
// 检查 xray 运行状态
xrayRunning := false
xrayVersion := ""
cmd := exec.Command("xray", "version")
@@ -1214,12 +1290,9 @@ func (c *Client) sendScanResult(conn *websocket.Conn) {
xrayRunning = true
}
// Read inbounds from config
// 从配置读取入站列表
var inbounds []map[string]interface{}
configPaths := []string{
"/usr/local/etc/xray/config.json",
"/etc/xray/config.json",
}
configPaths := constants.DefaultXrayConfigPaths
for _, cfgPath := range configPaths {
data, err := os.ReadFile(cfgPath)
if err != nil {
internal/collector/metrics.go
+22 -21
View File
@@ -8,28 +8,29 @@ import (
"os"
"strconv"
"strings"
"time"
"mmw-agent/internal/constants"
)
// XrayMetrics represents the metrics response from Xray's /debug/vars endpoint
// XrayMetrics 表示 Xray /debug/vars 的响应结构。
type XrayMetrics struct {
Stats *XrayStats `json:"stats,omitempty"`
}
// XrayStats contains inbound, outbound, and user traffic stats
// XrayStats 包含入站、出站和用户维度的流量统计。
type XrayStats struct {
Inbound map[string]TrafficData `json:"inbound,omitempty"`
Outbound map[string]TrafficData `json:"outbound,omitempty"`
User map[string]TrafficData `json:"user,omitempty"`
}
// TrafficData contains uplink and downlink traffic in bytes
// TrafficData 表示上下行流量（字节）。
type TrafficData struct {
Uplink int64 `json:"uplink"`
Downlink int64 `json:"downlink"`
}
// XrayConfig represents the structure of xray config.json for reading metrics port
// XrayConfig 用于读取 xray config.json 中的 metrics 监听配置。
type XrayConfig struct {
Log json.RawMessage `json:"log,omitempty"`
DNS json.RawMessage `json:"dns,omitempty"`
@@ -42,51 +43,51 @@ type XrayConfig struct {
Metrics *MetricsConfig `json:"metrics,omitempty"`
}
// MetricsConfig represents the metrics section in xray config
// MetricsConfig 对应 xray 配置中的 metrics 段。
type MetricsConfig struct {
Tag string `json:"tag,omitempty"`
Listen string `json:"listen,omitempty"` // Format: "127.0.0.1:38889"
Listen string `json:"listen,omitempty"` // 格式示例: "127.0.0.1:38889"
}
// Collector collects traffic metrics from Xray servers
// Collector 负责采集 Xray 流量指标。
type Collector struct {
httpClient *http.Client
defaultMetricsPort int
defaultMetricsHost string
}
// NewCollector creates a new metrics collector
// 创建指标采集器。
func NewCollector() *Collector {
return &Collector{
httpClient: &http.Client{Timeout: 10 * time.Second},
defaultMetricsPort: 38889,
defaultMetricsHost: "127.0.0.1",
httpClient: &http.Client{Timeout: constants.DefaultHTTPClientTimeout},
defaultMetricsPort: constants.DefaultMetricsPort,
defaultMetricsHost: constants.DefaultMetricsHost,
}
}
// GetMetricsPortFromConfig reads the metrics port from xray config file
// 从 xray 配置中读取 metrics 监听地址和端口。
func (c *Collector) GetMetricsPortFromConfig(configPath string) (string, int, error) {
if configPath == "" {
return "127.0.0.1", c.defaultMetricsPort, nil
return constants.DefaultMetricsHost, c.defaultMetricsPort, nil
}
data, err := os.ReadFile(configPath)
if err != nil {
return "127.0.0.1", c.defaultMetricsPort, fmt.Errorf("read config file: %w", err)
return constants.DefaultMetricsHost, c.defaultMetricsPort, fmt.Errorf("read config file: %w", err)
}
var config XrayConfig
if err := json.Unmarshal(data, &config); err != nil {
return "127.0.0.1", c.defaultMetricsPort, fmt.Errorf("parse config file: %w", err)
return constants.DefaultMetricsHost, c.defaultMetricsPort, fmt.Errorf("parse config file: %w", err)
}
if config.Metrics == nil || config.Metrics.Listen == "" {
return "", 0, fmt.Errorf("metrics not configured in xray config")
}
// Parse listen address (format: "127.0.0.1:38889" or ":38889")
// 解析监听地址，支持 "127.0.0.1:38889" 或 ":38889"
listen := config.Metrics.Listen
host := "127.0.0.1"
host := constants.DefaultMetricsHost
var port int
if strings.Contains(listen, ":") {
@@ -102,7 +103,7 @@ func (c *Collector) GetMetricsPortFromConfig(configPath string) (string, int, er
port = p
}
} else {
// Try to parse as port only
// 兼容仅填写端口的写法
p, err := strconv.Atoi(listen)
if err != nil {
return "", 0, fmt.Errorf("invalid metrics listen format: %s", listen)
@@ -117,7 +118,7 @@ func (c *Collector) GetMetricsPortFromConfig(configPath string) (string, int, er
return host, port, nil
}
// FetchMetrics fetches metrics from Xray's /debug/vars endpoint
// 从 Xray 的 /debug/vars 拉取指标。
func (c *Collector) FetchMetrics(host string, port int) (*XrayMetrics, error) {
url := fmt.Sprintf("http://%s:%d/debug/vars", host, port)
@@ -149,7 +150,7 @@ func (c *Collector) FetchMetrics(host string, port int) (*XrayMetrics, error) {
return &metrics, nil
}
// MergeStats merges source stats into dest stats
// 将 source 的统计合并到 dest。
func MergeStats(dest, source *XrayStats) {
if source == nil {
return
internal/config/config.go
+24 -26
View File
@@ -5,12 +5,14 @@ import (
"strconv"
"time"
"mmw-agent/internal/constants"
"gopkg.in/yaml.v3"
)
const AgentUserAgent = "miaomiaowux/0.1"
const AgentUserAgent = constants.AgentUserAgent
// Config holds the agent configuration
// Config 保存 agent 的运行配置。
type Config struct {
MasterURL string `yaml:"master_url"`
Token string `yaml:"token"`
@@ -21,20 +23,16 @@ type Config struct {
SpeedReportInterval time.Duration `yaml:"speed_report_interval"`
}
// XrayServer represents a local Xray server configuration
// XrayServer 表示本机 Xray 节点配置。
type XrayServer struct {
Name string `yaml:"name"`
ConfigPath string `yaml:"config_path"`
}
// DefaultXrayConfigPaths are the default paths to search for Xray config
var DefaultXrayConfigPaths = []string{
"/usr/local/etc/xray/config.json",
"/etc/xray/config.json",
"/opt/xray/config.json",
}
// DefaultXrayConfigPaths 是默认的 Xray 配置搜索路径。
var DefaultXrayConfigPaths = append([]string(nil), constants.DefaultXrayConfigPaths...)
// Load loads configuration from a YAML file
// 从 YAML 文件加载配置。
func Load(path string) (*Config, error) {
data, err := os.ReadFile(path)
if err != nil {
@@ -46,13 +44,13 @@ func Load(path string) (*Config, error) {
return nil, err
}
// Apply defaults
// 补齐默认值
config.applyDefaults()
return &config, nil
}
// FromEnv creates configuration from environment variables
// 从环境变量构造配置。
func FromEnv() *Config {
config := &Config{
MasterURL: os.Getenv("MMWX_MASTER_URL"),
@@ -61,14 +59,14 @@ func FromEnv() *Config {
ListenPort: os.Getenv("MMWX_LISTEN_PORT"),
}
// Parse Xray config path from env
// 读取 Xray 配置路径
if xrayConfig := os.Getenv("MMWX_XRAY_CONFIG"); xrayConfig != "" {
config.XrayServers = []XrayServer{
{Name: "primary", ConfigPath: xrayConfig},
}
}
// Parse intervals
// 读取上报间隔
if interval := os.Getenv("MMWX_TRAFFIC_INTERVAL"); interval != "" {
if d, err := time.ParseDuration(interval); err == nil {
config.TrafficReportInterval = d
@@ -84,7 +82,7 @@ func FromEnv() *Config {
return config
}
// Merge merges environment config into file config (env takes precedence)
// 合并环境变量配置到文件配置（环境变量优先）。
func (c *Config) Merge(env *Config) {
if env.MasterURL != "" {
c.MasterURL = env.MasterURL
@@ -109,28 +107,28 @@ func (c *Config) Merge(env *Config) {
}
}
// applyDefaults sets default values for unset fields
// 为空字段填充默认值。
func (c *Config) applyDefaults() {
if c.ConnectionMode == "" {
c.ConnectionMode = "auto"
c.ConnectionMode = constants.ConnectionModeAuto
}
if c.ListenPort == "" {
c.ListenPort = "23889"
c.ListenPort = constants.DefaultListenPort
}
if c.TrafficReportInterval == 0 {
c.TrafficReportInterval = 1 * time.Minute
c.TrafficReportInterval = constants.DefaultTrafficReportInterval
}
if c.SpeedReportInterval == 0 {
c.SpeedReportInterval = 3 * time.Second
c.SpeedReportInterval = constants.DefaultSpeedReportInterval
}
// Auto-discover Xray servers if not configured
// 未显式配置时自动探测 Xray 配置
if len(c.XrayServers) == 0 {
c.XrayServers = c.discoverXrayServers()
}
}
// discoverXrayServers scans default paths for Xray config files
// 扫描默认路径中的 Xray 配置文件。
func (c *Config) discoverXrayServers() []XrayServer {
var servers []XrayServer
for i, path := range DefaultXrayConfigPaths {
@@ -144,11 +142,11 @@ func (c *Config) discoverXrayServers() []XrayServer {
return servers
}
// Validate checks if the configuration is valid
// 校验配置是否合法。
func (c *Config) Validate() error {
// Token is required for non-pull modes
if c.ConnectionMode != "pull" && c.Token == "" {
// Allow empty token, will work in pull mode only
// 拉取模式之外通常需要 token
if c.ConnectionMode != constants.ConnectionModePull && c.Token == "" {
// 兼容空 token，实际仅拉取模式可正常工作
}
return nil
}
internal/constants/defaults.go
+109
View File
@@ -0,0 +1,109 @@
package constants
import "time"
const (
AgentUserAgent = "miaomiaowux/0.1"
)
const (
HeaderAuthorization = "Authorization"
HeaderContentType = "Content-Type"
HeaderMMRemoteToken = "MM-Remote-Token"
HeaderUserAgent = "User-Agent"
ContentTypeJSON = "application/json"
BearerPrefix = "Bearer "
)
const (
ConnectionModeAuto = "auto"
ConnectionModePull = "pull"
)
const (
DefaultListenPort = "23889"
)
const (
DefaultTrafficReportInterval = 1 * time.Minute
DefaultSpeedReportInterval = 3 * time.Second
DefaultHTTPClientTimeout = 10 * time.Second
DefaultReadTimeout = 30 * time.Second
DefaultShutdownTimeout = 10 * time.Second
DefaultRPCShortTimeout = 5 * time.Second
)
const (
DefaultMetricsHost = "127.0.0.1"
DefaultMetricsPort = 38889
DefaultMetricsListen = "127.0.0.1:38889"
LocalhostIP = "127.0.0.1"
)
const (
WebSocketMaxConsecutiveFailures = 5
WebSocketMaxAuthFailures = 10
AuthFailureSleepBackoff = 30 * time.Minute
PullModeTrafficReportThreshold = 30 * time.Second
)
const (
ReconnectBaseBackoff = 5 * time.Second
ReconnectMaxBackoff = 5 * time.Minute
AuthFailureBackoffStep = 30 * time.Second
AuthFailureMaxBackoff = 10 * time.Minute
AutoModePullFallbackBackoff = 30 * time.Second
WebSocketHandshakeTimeout = 10 * time.Second
WebSocketReadDeadline = 10 * time.Second
WebSocketHeartbeatInterval = 30 * time.Second
WebSocketIdleDeadline = 5 * time.Minute
)
const (
DomainProbeDefaultTimeoutMS = 2000
DomainProbeMinTimeoutMS = 200
DomainProbeMaxTimeoutMS = 10000
DomainProbeMaxCount = 200
DomainProbeConcurrency = 16
)
var (
NginxPrimaryPrefixDir = "/usr/local/nginx"
DefaultXrayConfigPaths = []string{
"/usr/local/etc/xray/config.json",
"/etc/xray/config.json",
"/opt/xray/config.json",
}
XrayConfigDirPaths = []string{
"/usr/local/etc/xray",
"/etc/xray",
"/opt/xray",
}
DefaultNginxConfigPaths = []string{
"/etc/nginx/nginx.conf",
"/usr/local/nginx/conf/nginx.conf",
}
NginxConfigDirPaths = []string{
"/etc/nginx",
"/etc/nginx/sites-available",
"/etc/nginx/sites-enabled",
"/etc/nginx/conf.d",
"/usr/local/nginx/conf",
}
NginxSSLServerDirPaths = []string{
"/usr/local/nginx/servers",
"/usr/local/nginx/conf/servers",
"/etc/nginx/servers",
"/etc/nginx/conf.d",
}
XrayBinarySearchPaths = []string{
"/usr/local/bin/xray",
"/usr/bin/xray",
"/opt/xray/xray",
}
NginxBinarySearchPaths = []string{
"/usr/local/nginx/sbin/nginx",
"nginx",
}
)
internal/constants/routes.go
+43
View File
@@ -0,0 +1,43 @@
package constants
const (
PathHealth = "/health"
)
const (
PathChildTraffic = "/api/child/traffic"
PathChildSpeed = "/api/child/speed"
PathChildServiceStats = "/api/child/services/status"
PathChildServiceCtl = "/api/child/services/control"
PathChildXrayInstall = "/api/child/xray/install"
PathChildXrayRemove = "/api/child/xray/remove"
PathChildXrayConfig = "/api/child/xray/config"
PathChildXraySysCfg = "/api/child/xray/system-config"
PathChildXrayCfgFiles = "/api/child/xray/config-files"
PathChildNginxInstall = "/api/child/nginx/install"
PathChildNginxRemove = "/api/child/nginx/remove"
PathChildNginxConfig = "/api/child/nginx/config"
PathChildNginxCfgFile = "/api/child/nginx/config-files"
PathChildSystemInfo = "/api/child/system/info"
PathChildInbounds = "/api/child/inbounds"
PathChildOutbounds = "/api/child/outbounds"
PathChildRouting = "/api/child/routing"
PathChildScan = "/api/child/scan"
PathChildCertDeploy = "/api/child/cert/deploy"
PathChildNginxSetup = "/api/child/nginx/setup-ssl"
PathChildDomainProbe = "/api/child/domains/latency"
)
const (
PathChildXrayInstallStream = "/api/child/xray/install-stream"
PathChildXrayRemoveStream = "/api/child/xray/remove-stream"
PathChildNginxInstallSSE = "/api/child/nginx/install-stream"
PathChildNginxRemoveSSE = "/api/child/nginx/remove-stream"
)
const (
PathRemoteWebSocket = "/api/remote/ws"
PathRemoteHeartbeat = "/api/remote/heartbeat"
PathRemoteTraffic = "/api/remote/traffic"
PathRemoteSpeed = "/api/remote/speed"
)
internal/handler/domain_latency.go → internal/handler/domain_latency_handler.go
+12 -10
View File
@@ -9,6 +9,8 @@ import (
"strings"
"sync"
"time"
"mmw-agent/internal/constants"
)
type DomainLatencyProbeRequest struct {
@@ -24,7 +26,7 @@ type DomainLatencyProbeResult struct {
Error string `json:"error,omitempty"`
}
// HandleDomainLatencyProbe handles POST /api/child/domains/latency
// 处理 POST /api/child/domains/latency 请求。
func (h *ManageHandler) HandleDomainLatencyProbe(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -49,13 +51,13 @@ func (h *ManageHandler) HandleDomainLatencyProbe(w http.ResponseWriter, r *http.
timeoutMs := req.TimeoutMs
if timeoutMs <= 0 {
timeoutMs = 2000
timeoutMs = constants.DomainProbeDefaultTimeoutMS
}
if timeoutMs < 200 {
timeoutMs = 200
if timeoutMs < constants.DomainProbeMinTimeoutMS {
timeoutMs = constants.DomainProbeMinTimeoutMS
}
if timeoutMs > 10000 {
timeoutMs = 10000
if timeoutMs > constants.DomainProbeMaxTimeoutMS {
timeoutMs = constants.DomainProbeMaxTimeoutMS
}
timeout := time.Duration(timeoutMs) * time.Millisecond
@@ -64,13 +66,13 @@ func (h *ManageHandler) HandleDomainLatencyProbe(w http.ResponseWriter, r *http.
writeError(w, http.StatusBadRequest, "no valid domain to probe")
return
}
if len(domains) > 200 {
domains = domains[:200]
if len(domains) > constants.DomainProbeMaxCount {
domains = domains[:constants.DomainProbeMaxCount]
}
results := make([]DomainLatencyProbeResult, 0, len(domains))
resultCh := make(chan DomainLatencyProbeResult, len(domains))
sem := make(chan struct{}, 16)
sem := make(chan struct{}, constants.DomainProbeConcurrency)
var wg sync.WaitGroup
for _, domain := range domains {
@@ -206,7 +208,7 @@ func splitHostPortLoose(input string) (host string, port string, ok bool) {
}
}
// Fallback for "domain:443" without brackets handling.
// 兼容不带方括号的 "domain:443" 写法。
idx := strings.LastIndex(s, ":")
if idx <= 0 || idx >= len(s)-1 {
return "", "", false
internal/handler/manage.go → internal/handler/management_handler.go
+138 -149
View File
@@ -17,7 +17,7 @@ import (
"sync/atomic"
"time"
"mmw-agent/internal/config"
"mmw-agent/internal/constants"
"mmw-agent/internal/xrpc"
"github.com/xtls/xray-core/app/proxyman/command"
@@ -26,21 +26,21 @@ import (
var nginxInstalling atomic.Bool
// ManageHandler handles management API requests for child servers
// ManageHandler 处理子端管理接口请求。
type ManageHandler struct {
configToken string
}
// NewManageHandler creates a new management handler
// 创建管理处理器。
func NewManageHandler(configToken string) *ManageHandler {
return &ManageHandler{
configToken: configToken,
}
}
// authenticate checks if the request is authorized (token + User-Agent)
// 校验请求身份（token + User-Agent）。
func (h *ManageHandler) authenticate(r *http.Request) bool {
if r.Header.Get("User-Agent") != config.AgentUserAgent {
if r.Header.Get(constants.HeaderUserAgent) != constants.AgentUserAgent {
return false
}
@@ -48,30 +48,30 @@ func (h *ManageHandler) authenticate(r *http.Request) bool {
return true
}
auth := r.Header.Get("Authorization")
auth := r.Header.Get(constants.HeaderAuthorization)
if auth == "" {
auth = r.Header.Get("MM-Remote-Token")
auth = r.Header.Get(constants.HeaderMMRemoteToken)
}
if auth == "" {
return false
}
if strings.HasPrefix(auth, "Bearer ") {
token := strings.TrimPrefix(auth, "Bearer ")
if strings.HasPrefix(auth, constants.BearerPrefix) {
token := strings.TrimPrefix(auth, constants.BearerPrefix)
return token == h.configToken
}
return auth == h.configToken
}
// writeJSON writes JSON response
// 输出 JSON 响应。
func writeJSON(w http.ResponseWriter, statusCode int, data interface{}) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set(constants.HeaderContentType, constants.ContentTypeJSON)
w.WriteHeader(statusCode)
json.NewEncoder(w).Encode(data)
}
// writeError writes error response
// 输出错误响应。
func writeError(w http.ResponseWriter, statusCode int, message string) {
writeJSON(w, statusCode, map[string]interface{}{
"success": false,
@@ -79,23 +79,23 @@ func writeError(w http.ResponseWriter, statusCode int, message string) {
})
}
// ================== System Services Status ==================
// ================== 系统服务状态 ==================
// ServicesStatusResponse represents the response for services status
// ServicesStatusResponse 表示服务状态查询响应。
type ServicesStatusResponse struct {
Success bool `json:"success"`
Xray *ServiceStatus `json:"xray,omitempty"`
Nginx *ServiceStatus `json:"nginx,omitempty"`
}
// ServiceStatus represents a service status
// ServiceStatus 表示单个服务状态。
type ServiceStatus struct {
Installed bool `json:"installed"`
Running bool `json:"running"`
Version string `json:"version,omitempty"`
}
// HandleServicesStatus handles GET /api/child/services/status
// 处理 GET /api/child/services/status。
func (h *ManageHandler) HandleServicesStatus(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -121,8 +121,7 @@ func (h *ManageHandler) getXrayStatus() *ServiceStatus {
xrayPath, err := exec.LookPath("xray")
if err != nil {
commonPaths := []string{"/usr/local/bin/xray", "/usr/bin/xray", "/opt/xray/xray"}
for _, p := range commonPaths {
for _, p := range constants.XrayBinarySearchPaths {
if _, err := os.Stat(p); err == nil {
xrayPath = p
break
@@ -142,7 +141,7 @@ func (h *ManageHandler) getXrayStatus() *ServiceStatus {
}
}
// Check systemctl first
// 优先使用 systemctl 检查
cmd := exec.Command("systemctl", "is-active", "xray")
output, _ := cmd.Output()
if strings.TrimSpace(string(output)) == "active" {
@@ -150,14 +149,14 @@ func (h *ManageHandler) getXrayStatus() *ServiceStatus {
return status
}
// Fallback: check if xray process is running via pgrep
// 兜底：用 pgrep 检查 xray 进程
pgrepCmd := exec.Command("pgrep", "-x", "xray")
if err := pgrepCmd.Run(); err == nil {
status.Running = true
return status
}
// Fallback: check via ps for processes containing "xray"
// 兜底：用 ps 检查包含 "xray" 的进程
psCmd := exec.Command("bash", "-c", "ps aux | grep -v grep | grep -E '[x]ray' | head -1")
if output, err := psCmd.Output(); err == nil && len(strings.TrimSpace(string(output))) > 0 {
status.Running = true
@@ -169,12 +168,17 @@ func (h *ManageHandler) getXrayStatus() *ServiceStatus {
func (h *ManageHandler) getNginxStatus() *ServiceStatus {
status := &ServiceStatus{}
// Check PATH first, then compiled install path
// 先查 PATH，再查编译安装路径
nginxPath, err := exec.LookPath("nginx")
if err != nil {
if _, statErr := os.Stat("/usr/local/nginx/sbin/nginx"); statErr == nil {
nginxPath = "/usr/local/nginx/sbin/nginx"
err = nil
for _, candidate := range constants.NginxBinarySearchPaths {
if strings.Contains(candidate, "/") {
if _, statErr := os.Stat(candidate); statErr == nil {
nginxPath = candidate
err = nil
break
}
}
}
}
if err == nil {
@@ -190,7 +194,7 @@ func (h *ManageHandler) getNginxStatus() *ServiceStatus {
status.Version = "安装中..."
}
// Check systemctl first
// 优先使用 systemctl 检查
cmd := exec.Command("systemctl", "is-active", "nginx")
output, _ := cmd.Output()
if strings.TrimSpace(string(output)) == "active" {
@@ -198,14 +202,14 @@ func (h *ManageHandler) getNginxStatus() *ServiceStatus {
return status
}
// Fallback: check if nginx process is running via pgrep
// 兜底：用 pgrep 检查 nginx 进程
pgrepCmd := exec.Command("pgrep", "-x", "nginx")
if err := pgrepCmd.Run(); err == nil {
status.Running = true
return status
}
// Fallback: check via ps for nginx master process
// 兜底：用 ps 检查 nginx master 进程
psCmd := exec.Command("bash", "-c", "ps aux | grep -v grep | grep -E 'nginx: master' | head -1")
if output, err := psCmd.Output(); err == nil && len(strings.TrimSpace(string(output))) > 0 {
status.Running = true
@@ -214,15 +218,15 @@ func (h *ManageHandler) getNginxStatus() *ServiceStatus {
return status
}
// ================== Service Control ==================
// ================== 服务控制 ==================
// ServiceControlRequest represents a service control request
// ServiceControlRequest 表示服务控制请求。
type ServiceControlRequest struct {
Service string `json:"service"`
Action string `json:"action"`
}
// HandleServiceControl handles POST /api/child/services/control
// 处理 POST /api/child/services/control。
func (h *ManageHandler) HandleServiceControl(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -265,9 +269,9 @@ func (h *ManageHandler) HandleServiceControl(w http.ResponseWriter, r *http.Requ
})
}
// ================== Xray Installation ==================
// ================== Xray 安装 ==================
// HandleXrayInstall handles POST /api/child/xray/install
// 处理 POST /api/child/xray/install。
func (h *ManageHandler) HandleXrayInstall(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -297,7 +301,7 @@ func (h *ManageHandler) HandleXrayInstall(w http.ResponseWriter, r *http.Request
log.Printf("[Manage] Xray installed successfully")
// Deploy default config if no config exists
// 若无配置则下发默认配置
h.deployDefaultXrayConfig()
writeJSON(w, http.StatusOK, map[string]interface{}{
@@ -307,7 +311,7 @@ func (h *ManageHandler) HandleXrayInstall(w http.ResponseWriter, r *http.Request
})
}
// HandleXrayRemove handles POST /api/child/xray/remove
// 处理 POST /api/child/xray/remove。
func (h *ManageHandler) HandleXrayRemove(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -344,9 +348,9 @@ func (h *ManageHandler) HandleXrayRemove(w http.ResponseWriter, r *http.Request)
})
}
// ================== Xray Configuration ==================
// ================== Xray 配置 ==================
// HandleXrayConfig handles GET/POST /api/child/xray/config
// 处理 GET/POST /api/child/xray/config。
func (h *ManageHandler) HandleXrayConfig(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -364,11 +368,7 @@ func (h *ManageHandler) HandleXrayConfig(w http.ResponseWriter, r *http.Request)
}
func (h *ManageHandler) getXrayConfig(w http.ResponseWriter, r *http.Request) {
configPaths := []string{
"/usr/local/etc/xray/config.json",
"/etc/xray/config.json",
"/opt/xray/config.json",
}
configPaths := constants.DefaultXrayConfigPaths
var configPath string
var content []byte
@@ -413,7 +413,7 @@ func (h *ManageHandler) setXrayConfig(w http.ResponseWriter, r *http.Request) {
configPath := req.Path
if configPath == "" {
configPath = "/usr/local/etc/xray/config.json"
configPath = constants.DefaultXrayConfigPaths[0]
}
dir := filepath.Dir(configPath)
@@ -436,9 +436,9 @@ func (h *ManageHandler) setXrayConfig(w http.ResponseWriter, r *http.Request) {
})
}
// ================== Xray System Configuration ==================
// ================== Xray 系统配置 ==================
// XraySystemConfig represents the system configuration state
// XraySystemConfig 表示 Xray 系统配置状态。
type XraySystemConfig struct {
MetricsEnabled bool `json:"metrics_enabled"`
MetricsListen string `json:"metrics_listen"`
@@ -447,7 +447,7 @@ type XraySystemConfig struct {
GrpcPort int `json:"grpc_port"`
}
// HandleXraySystemConfig handles GET/POST /api/child/xray/system-config
// 处理 GET/POST /api/child/xray/system-config。
func (h *ManageHandler) HandleXraySystemConfig(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -484,7 +484,7 @@ func (h *ManageHandler) getXraySystemConfig(w http.ResponseWriter, r *http.Reque
}
sysConfig := &XraySystemConfig{
MetricsListen: "127.0.0.1:38889",
MetricsListen: constants.DefaultMetricsListen,
GrpcPort: 46736,
}
@@ -605,10 +605,10 @@ func (h *ManageHandler) updateXraySystemConfig(w http.ResponseWriter, r *http.Re
apiInbound := map[string]interface{}{
"tag": "api",
"port": float64(req.GrpcPort),
"listen": "127.0.0.1",
"listen": constants.LocalhostIP,
"protocol": "dokodemo-door",
"settings": map[string]interface{}{
"address": "127.0.0.1",
"address": constants.LocalhostIP,
},
}
config["inbounds"] = append([]interface{}{apiInbound}, inbounds...)
@@ -618,10 +618,10 @@ func (h *ManageHandler) updateXraySystemConfig(w http.ResponseWriter, r *http.Re
map[string]interface{}{
"tag": "api",
"port": float64(req.GrpcPort),
"listen": "127.0.0.1",
"listen": constants.LocalhostIP,
"protocol": "dokodemo-door",
"settings": map[string]interface{}{
"address": "127.0.0.1",
"address": constants.LocalhostIP,
},
},
}
@@ -722,9 +722,9 @@ func (h *ManageHandler) removeAPIRoutingRule(config map[string]interface{}) {
routing["rules"] = newRules
}
// ================== Nginx Installation ==================
// ================== Nginx 安装 ==================
// HandleNginxInstall handles POST /api/child/nginx/install
// 处理 POST /api/child/nginx/install。
func (h *ManageHandler) HandleNginxInstall(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -777,7 +777,7 @@ func (h *ManageHandler) HandleNginxInstall(w http.ResponseWriter, r *http.Reques
})
}
// HandleNginxRemove handles POST /api/child/nginx/remove
// 处理 POST /api/child/nginx/remove。
func (h *ManageHandler) HandleNginxRemove(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -814,9 +814,9 @@ func (h *ManageHandler) HandleNginxRemove(w http.ResponseWriter, r *http.Request
})
}
// ================== Nginx Configuration ==================
// ================== Nginx 配置 ==================
// HandleNginxConfig handles GET/POST /api/child/nginx/config
// 处理 GET/POST /api/child/nginx/config。
func (h *ManageHandler) HandleNginxConfig(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -834,10 +834,7 @@ func (h *ManageHandler) HandleNginxConfig(w http.ResponseWriter, r *http.Request
}
func (h *ManageHandler) getNginxConfig(w http.ResponseWriter, r *http.Request) {
configPaths := []string{
"/etc/nginx/nginx.conf",
"/usr/local/nginx/conf/nginx.conf",
}
configPaths := constants.DefaultNginxConfigPaths
var configPath string
var content []byte
@@ -876,7 +873,7 @@ func (h *ManageHandler) setNginxConfig(w http.ResponseWriter, r *http.Request) {
configPath := req.Path
if configPath == "" {
configPath = "/etc/nginx/nginx.conf"
configPath = constants.DefaultNginxConfigPaths[0]
}
backupPath := configPath + ".bak." + time.Now().Format("20060102150405")
@@ -908,9 +905,9 @@ func (h *ManageHandler) setNginxConfig(w http.ResponseWriter, r *http.Request) {
})
}
// ================== System Info ==================
// ================== 系统信息 ==================
// HandleSystemInfo handles GET /api/child/system/info
// 处理 GET /api/child/system/info。
func (h *ManageHandler) HandleSystemInfo(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -960,9 +957,9 @@ func (h *ManageHandler) HandleSystemInfo(w http.ResponseWriter, r *http.Request)
writeJSON(w, http.StatusOK, info)
}
// ================== Config Files Management ==================
// ================== 配置文件管理 ==================
// ConfigFileInfo represents a config file entry
// ConfigFileInfo 表示配置文件条目。
type ConfigFileInfo struct {
Name string `json:"name"`
Path string `json:"path"`
@@ -970,7 +967,7 @@ type ConfigFileInfo struct {
ModTime string `json:"mod_time"`
}
// HandleXrayConfigFiles handles listing and managing xray config files
// 处理 xray 配置文件的列表与读写。
func (h *ManageHandler) HandleXrayConfigFiles(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -993,11 +990,7 @@ func (h *ManageHandler) HandleXrayConfigFiles(w http.ResponseWriter, r *http.Req
}
func (h *ManageHandler) listXrayConfigFiles(w http.ResponseWriter, r *http.Request) {
configDirs := []string{
"/usr/local/etc/xray",
"/etc/xray",
"/opt/xray",
}
configDirs := constants.XrayConfigDirPaths
var files []ConfigFileInfo
var baseDir string
@@ -1043,9 +1036,9 @@ func (h *ManageHandler) getXrayConfigFile(w http.ResponseWriter, r *http.Request
file = filepath.Clean(file)
configDirs := []string{
"/usr/local/etc/xray",
"/etc/xray",
"/opt/xray",
constants.XrayConfigDirPaths[0],
constants.XrayConfigDirPaths[1],
constants.XrayConfigDirPaths[2],
}
var filePath string
@@ -1100,9 +1093,9 @@ func (h *ManageHandler) saveXrayConfigFile(w http.ResponseWriter, r *http.Reques
}
configDirs := []string{
"/usr/local/etc/xray",
"/etc/xray",
"/opt/xray",
constants.XrayConfigDirPaths[0],
constants.XrayConfigDirPaths[1],
constants.XrayConfigDirPaths[2],
}
var configDir string
@@ -1114,7 +1107,7 @@ func (h *ManageHandler) saveXrayConfigFile(w http.ResponseWriter, r *http.Reques
}
if configDir == "" {
configDir = "/usr/local/etc/xray"
configDir = constants.XrayConfigDirPaths[0]
if err := os.MkdirAll(configDir, 0755); err != nil {
writeError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to create config directory: %v", err))
return
@@ -1145,7 +1138,7 @@ func (h *ManageHandler) saveXrayConfigFile(w http.ResponseWriter, r *http.Reques
})
}
// HandleNginxConfigFiles handles listing and managing nginx config files
// 处理 nginx 配置文件的列表与读写。
func (h *ManageHandler) HandleNginxConfigFiles(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -1172,10 +1165,10 @@ func (h *ManageHandler) listNginxConfigFiles(w http.ResponseWriter, r *http.Requ
dir string
description string
}{
{"/etc/nginx", "main"},
{"/etc/nginx/sites-available", "sites-available"},
{"/etc/nginx/sites-enabled", "sites-enabled"},
{"/etc/nginx/conf.d", "conf.d"},
{constants.NginxConfigDirPaths[0], "main"},
{constants.NginxConfigDirPaths[1], "sites-available"},
{constants.NginxConfigDirPaths[2], "sites-enabled"},
{constants.NginxConfigDirPaths[3], "conf.d"},
}
result := make(map[string][]ConfigFileInfo)
@@ -1218,11 +1211,11 @@ func (h *ManageHandler) getNginxConfigFile(w http.ResponseWriter, r *http.Reques
file = filepath.Clean(file)
allowedDirs := []string{
"/etc/nginx",
"/etc/nginx/sites-available",
"/etc/nginx/sites-enabled",
"/etc/nginx/conf.d",
"/usr/local/nginx/conf",
constants.NginxConfigDirPaths[0],
constants.NginxConfigDirPaths[1],
constants.NginxConfigDirPaths[2],
constants.NginxConfigDirPaths[3],
constants.NginxConfigDirPaths[4],
}
var filePath string
@@ -1283,8 +1276,8 @@ func (h *ManageHandler) saveNginxConfigFile(w http.ResponseWriter, r *http.Reque
req.Path = filepath.Clean(req.Path)
allowedDirs := []string{
"/etc/nginx",
"/usr/local/nginx/conf",
constants.NginxConfigDirPaths[0],
constants.NginxConfigDirPaths[4],
}
allowed := false
@@ -1338,16 +1331,16 @@ func (h *ManageHandler) saveNginxConfigFile(w http.ResponseWriter, r *http.Reque
})
}
// ================== Xray Inbounds Management ==================
// ================== Xray 入站管理 ==================
// InboundRequest represents inbound management request
// InboundRequest 表示入站管理请求。
type InboundRequest struct {
Action string `json:"action"`
Inbound map[string]interface{} `json:"inbound,omitempty"`
Tag string `json:"tag,omitempty"`
}
// HandleInbounds handles inbound management
// 处理入站管理请求。
func (h *ManageHandler) HandleInbounds(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -1412,7 +1405,7 @@ func (h *ManageHandler) getInboundTagsFromGRPC() []string {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
clients, err := xrpc.New(ctx, "127.0.0.1", uint16(apiPort))
clients, err := xrpc.New(ctx, constants.LocalhostIP, uint16(apiPort))
if err != nil {
log.Printf("[Manage] Failed to connect to Xray gRPC: %v", err)
return nil
@@ -1517,7 +1510,7 @@ func (h *ManageHandler) manageInbound(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
clients, err := xrpc.New(ctx, "127.0.0.1", uint16(apiPort))
clients, err := xrpc.New(ctx, constants.LocalhostIP, uint16(apiPort))
if err != nil {
writeError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to connect to Xray: %v", err))
return
@@ -1557,27 +1550,27 @@ func (h *ManageHandler) manageInbound(w http.ResponseWriter, r *http.Request) {
return
}
// Try to remove from runtime (ignore error if not running)
// 尝试从运行态移除（未运行时报错可忽略）
runtimeErr := h.removeInbound(ctx, clients.Handler, req.Tag)
if runtimeErr != nil {
log.Printf("[Manage] Warning: Failed to remove inbound from runtime: %v", runtimeErr)
}
// Remove from config file (this is the primary operation)
// 从配置文件移除（主流程）
configErr := h.removeInboundFromConfig(req.Tag)
if configErr != nil {
log.Printf("[Manage] Warning: Failed to remove inbound from config: %v", configErr)
}
// Success if config operation succeeded (runtime removal is optional)
// The inbound might not exist in runtime if Xray wasn't restarted after config change
// 配置文件操作成功即可视为成功（运行态移除可选）
// 配置改动后若未重启，运行态可能还没有该入站
if configErr != nil {
// Config file operation failed
// 配置文件操作失败
if runtimeErr != nil {
// Both failed - check if it's just "not found" errors
// 两边都失败时，判断是否只是“未找到”错误
if strings.Contains(runtimeErr.Error(), "not enough information") {
// Xray says the inbound doesn't exist in runtime, which is fine
// Just report config error
// Xray 返回运行态不存在该入站，这属于可接受情况
// 仅返回配置文件错误
writeError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to remove inbound from config: %v", configErr))
} else {
writeError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to remove inbound: runtime=%v, config=%v", runtimeErr, configErr))
@@ -1588,7 +1581,7 @@ func (h *ManageHandler) manageInbound(w http.ResponseWriter, r *http.Request) {
return
}
// Config succeeded, runtime error is acceptable (inbound might not be loaded)
// 配置成功时，运行态报错可接受（可能尚未加载）
writeJSON(w, http.StatusOK, map[string]interface{}{
"success": true,
"message": "Inbound removed successfully",
@@ -1599,16 +1592,16 @@ func (h *ManageHandler) manageInbound(w http.ResponseWriter, r *http.Request) {
}
}
// ================== Xray Outbounds Management ==================
// ================== Xray 出站管理 ==================
// OutboundRequest represents outbound management request
// OutboundRequest 表示出站管理请求。
type OutboundRequest struct {
Action string `json:"action"`
Outbound map[string]interface{} `json:"outbound,omitempty"`
Tag string `json:"tag,omitempty"`
}
// HandleOutbounds handles outbound management
// 处理出站管理请求。
func (h *ManageHandler) HandleOutbounds(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -1673,7 +1666,7 @@ func (h *ManageHandler) manageOutbound(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
clients, err := xrpc.New(ctx, "127.0.0.1", uint16(apiPort))
clients, err := xrpc.New(ctx, constants.LocalhostIP, uint16(apiPort))
if err != nil {
writeError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to connect to Xray: %v", err))
return
@@ -1726,9 +1719,9 @@ func (h *ManageHandler) manageOutbound(w http.ResponseWriter, r *http.Request) {
}
}
// ================== Xray Routing Management ==================
// ================== Xray 路由管理 ==================
// RoutingRequest represents routing management request
// RoutingRequest 表示路由管理请求。
type RoutingRequest struct {
Action string `json:"action"`
Routing map[string]interface{} `json:"routing,omitempty"`
@@ -1736,7 +1729,7 @@ type RoutingRequest struct {
Index int `json:"index,omitempty"`
}
// HandleRouting handles routing management
// 处理路由管理请求。
func (h *ManageHandler) HandleRouting(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(r) {
writeError(w, http.StatusUnauthorized, "Unauthorized")
@@ -1871,14 +1864,10 @@ func (h *ManageHandler) manageRouting(w http.ResponseWriter, r *http.Request) {
})
}
// ================== Helper Functions ==================
// ================== 辅助函数 ==================
func (h *ManageHandler) findXrayConfigPath() string {
configPaths := []string{
"/usr/local/etc/xray/config.json",
"/etc/xray/config.json",
"/opt/xray/config.json",
}
configPaths := constants.DefaultXrayConfigPaths
for _, p := range configPaths {
if _, err := os.Stat(p); err == nil {
@@ -2124,9 +2113,9 @@ func (h *ManageHandler) removeOutboundFromConfig(tag string) error {
return os.WriteFile(configPath, newContent, 0644)
}
// ================== Scan ==================
// ================== 扫描 ==================
// ScanResponse represents the response for scan operation
// ScanResponse 表示扫描接口响应。
type ScanResponse struct {
Success bool `json:"success"`
Message string `json:"message"`
@@ -2139,7 +2128,7 @@ type ScanResponse struct {
ConfigAddedSections []string `json:"config_added_sections,omitempty"`
}
// HandleScan handles POST /api/child/scan
// 处理 POST /api/child/scan。
func (h *ManageHandler) HandleScan(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -2220,9 +2209,9 @@ func (h *ManageHandler) HandleScan(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, response)
}
// ================== Xray Config Auto-Complete ==================
// ================== Xray 配置自动补全 ==================
// EnsureXrayConfigResult holds the result of config check
// EnsureXrayConfigResult 表示配置检查结果。
type EnsureXrayConfigResult struct {
ConfigPath string `json:"config_path"`
Modified bool `json:"modified"`
@@ -2230,7 +2219,7 @@ type EnsureXrayConfigResult struct {
Error string `json:"error,omitempty"`
}
// EnsureXrayConfig checks and completes Xray configuration
// 检查并补全 Xray 配置。
func (h *ManageHandler) EnsureXrayConfig() *EnsureXrayConfigResult {
result := &EnsureXrayConfigResult{}
@@ -2279,7 +2268,7 @@ func (h *ManageHandler) EnsureXrayConfig() *EnsureXrayConfigResult {
if _, ok := config["metrics"]; !ok {
config["metrics"] = map[string]interface{}{
"tag": "Metrics",
"listen": "127.0.0.1:38889",
"listen": constants.DefaultMetricsListen,
}
result.AddedSections = append(result.AddedSections, "metrics")
modified = true
@@ -2376,10 +2365,10 @@ func (h *ManageHandler) addAPIInbound(config map[string]interface{}) {
apiInbound := map[string]interface{}{
"tag": "api",
"port": float64(46736),
"listen": "127.0.0.1",
"listen": constants.LocalhostIP,
"protocol": "dokodemo-door",
"settings": map[string]interface{}{
"address": "127.0.0.1",
"address": constants.LocalhostIP,
},
}
@@ -2432,9 +2421,9 @@ func (h *ManageHandler) addAPIRoutingRule(config map[string]interface{}) {
routing["rules"] = append([]interface{}{apiRule}, rules...)
}
// ================== Certificate Deploy ==================
// ================== 证书部署 ==================
// CertDeployRequest represents a certificate deploy request from master
// CertDeployRequest 表示主控端下发的证书部署请求。
type CertDeployRequest struct {
Domain string `json:"domain"`
CertPEM string `json:"cert_pem"`
@@ -2444,7 +2433,7 @@ type CertDeployRequest struct {
Reload string `json:"reload"` // nginx, xray, both, none
}
// HandleCertDeploy handles POST /api/child/cert/deploy
// 处理 POST /api/child/cert/deploy。
func (h *ManageHandler) HandleCertDeploy(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "method not allowed")
@@ -2513,9 +2502,9 @@ func deployNginxSSLConfig(domain string) {
return
}
confDir := "/usr/local/nginx/conf"
confDir := constants.NginxConfigDirPaths[4]
if _, err := os.Stat(confDir); err != nil {
confDir = "/etc/nginx"
confDir = constants.NginxConfigDirPaths[0]
}
certDir := filepath.Join(confDir, "cert")
@@ -2539,7 +2528,7 @@ func deployNginxSSLConfig(domain string) {
}
`, domain, domain, domain)
// Write to conf.d or append via include
// 写入 conf.d，或通过 include 挂载
confDDir := filepath.Join(confDir, "conf.d")
os.MkdirAll(confDDir, 0755)
sslConfPath := filepath.Join(confDDir, "ssl.conf")
@@ -2549,7 +2538,7 @@ func deployNginxSSLConfig(domain string) {
return
}
// Ensure main nginx.conf includes conf.d/*.conf
// 确保主 nginx.conf 包含 conf.d/*.conf
mainConf := filepath.Join(confDir, "nginx.conf")
content, err := os.ReadFile(mainConf)
if err != nil {
@@ -2559,7 +2548,7 @@ func deployNginxSSLConfig(domain string) {
includeDirective := "include conf.d/*.conf;"
if !strings.Contains(string(content), includeDirective) {
// Insert include before the last closing brace of http block
// 在 http 块最后一个右括号前插入 include
text := string(content)
lastBrace := strings.LastIndex(text, "}")
if lastBrace > 0 {
@@ -2574,8 +2563,8 @@ func deployNginxSSLConfig(domain string) {
log.Printf("[Manage] Nginx SSL config deployed for domain %s at %s", domain, sslConfPath)
}
// HandleNginxSetupSSL handles POST /api/child/nginx/setup-ssl
// Deploys nginx.conf + domain server block to servers/{domain}.conf.
// HandleNginxSetupSSL 处理 POST /api/child/nginx/setup-ssl。
// 部署 nginx.conf 和 servers/{domain}.conf。
func (h *ManageHandler) HandleNginxSetupSSL(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeError(w, http.StatusMethodNotAllowed, "Method not allowed")
@@ -2598,17 +2587,17 @@ func (h *ManageHandler) HandleNginxSetupSSL(w http.ResponseWriter, r *http.Reque
domain := strings.ToLower(strings.TrimSpace(req.Domain))
confDir := "/usr/local/nginx"
confDir := constants.NginxPrimaryPrefixDir
if _, err := os.Stat(confDir); err != nil {
confDir = "/etc/nginx"
confDir = constants.NginxConfigDirPaths[0]
}
// Ensure cert and servers directories exist
// 确保证书和 servers 目录存在
os.MkdirAll(filepath.Join(confDir, "cert"), 0755)
os.MkdirAll(filepath.Join(confDir, "servers"), 0755)
if req.NginxConfig != "" {
// Deploy base nginx.conf
// 下发主 nginx.conf
mainConf := filepath.Join(confDir, "nginx.conf")
if content, err := os.ReadFile(mainConf); err == nil {
os.WriteFile(mainConf+".bak."+time.Now().Format("20060102150405"), content, 0644)
@@ -2621,7 +2610,7 @@ func (h *ManageHandler) HandleNginxSetupSSL(w http.ResponseWriter, r *http.Reque
}
if req.DomainConfig != "" {
// Deploy domain-specific server block to servers/{domain}.conf
// 下发域名 server 配置到 servers/{domain}.conf
domainConfPath := filepath.Join(confDir, "servers", domain+".conf")
if err := os.WriteFile(domainConfPath, []byte(req.DomainConfig), 0644); err != nil {
writeError(w, http.StatusInternalServerError, fmt.Sprintf("Failed to write domain config: %v", err))
@@ -2629,11 +2618,11 @@ func (h *ManageHandler) HandleNginxSetupSSL(w http.ResponseWriter, r *http.Reque
}
log.Printf("[Manage] Domain config deployed at %s", domainConfPath)
} else {
// Fallback: legacy behavior
// 兜底：沿用旧逻辑
deployNginxSSLConfig(domain)
}
// Reload nginx to apply
// 重载 nginx 使配置生效
if err := reloadNginx(); err != nil {
log.Printf("[Manage] Nginx reload after setup-ssl failed: %v", err)
}
@@ -2645,7 +2634,7 @@ func (h *ManageHandler) HandleNginxSetupSSL(w http.ResponseWriter, r *http.Reque
}
func reloadNginx() error {
for _, bin := range []string{"/usr/local/nginx/sbin/nginx", "nginx"} {
for _, bin := range constants.NginxBinarySearchPaths {
if path, err := exec.LookPath(bin); err == nil {
return runCommand(path, "-s", "reload")
}
@@ -2660,11 +2649,11 @@ func runCommand(name string, args ...string) error {
return nil
}
// deployDefaultXrayConfig deploys the embedded default xray config if no config exists.
// 在缺失配置时下发内置默认配置。
func (h *ManageHandler) deployDefaultXrayConfig() {
configPath := "/usr/local/etc/xray/config.json"
configPath := constants.DefaultXrayConfigPaths[0]
if _, err := os.Stat(configPath); err == nil {
// Config already exists — run EnsureXrayConfig to add missing sections
// 配置已存在，执行 EnsureXrayConfig 补齐缺失段
result := h.EnsureXrayConfig()
if result.Modified {
log.Printf("[Manage] Xray config updated after install: added %v", result.AddedSections)
@@ -2685,7 +2674,7 @@ func (h *ManageHandler) deployDefaultXrayConfig() {
exec.Command("systemctl", "restart", "xray").Run()
}
// ================== SSE Streaming Install/Remove ==================
// ================== SSE 流式安装/卸载 ==================
func sseStreamCmd(w http.ResponseWriter, r *http.Request, cmd *exec.Cmd, completeMsg string) {
w.Header().Set("Content-Type", "text/event-stream")
@@ -2775,7 +2764,7 @@ func (h *ManageHandler) HandleXrayInstallStream(w http.ResponseWriter, r *http.R
cmd.Env = os.Environ()
sseStreamCmd(w, r, cmd, "Xray installed successfully")
// Deploy default config after install
// 安装完成后下发默认配置
h.deployDefaultXrayConfig()
}
internal/handler/api.go → internal/handler/pull_api_handler.go
+15 -15
View File
@@ -7,16 +7,16 @@ import (
"strings"
"mmw-agent/internal/agent"
"mmw-agent/internal/config"
"mmw-agent/internal/constants"
)
// APIHandler handles API requests from the master server (for pull mode)
// APIHandler 处理来自主控端的请求（拉取模式）。
type APIHandler struct {
client *agent.Client
configToken string
}
// NewAPIHandler creates a new API handler
// 创建 API 处理器。
func NewAPIHandler(client *agent.Client, configToken string) *APIHandler {
return &APIHandler{
client: client,
@@ -24,7 +24,7 @@ func NewAPIHandler(client *agent.Client, configToken string) *APIHandler {
}
}
// ServeHTTP handles the HTTP request for traffic data
// 返回流量数据。
func (h *APIHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -32,7 +32,7 @@ func (h *APIHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
if !h.authenticate(r) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set(constants.HeaderContentType, constants.ContentTypeJSON)
w.WriteHeader(http.StatusUnauthorized)
json.NewEncoder(w).Encode(map[string]interface{}{
"success": false,
@@ -44,7 +44,7 @@ func (h *APIHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
stats, err := h.client.GetStats()
if err != nil {
log.Printf("[API] Failed to get stats: %v", err)
w.Header().Set("Content-Type", "application/json")
w.Header().Set(constants.HeaderContentType, constants.ContentTypeJSON)
w.WriteHeader(http.StatusInternalServerError)
json.NewEncoder(w).Encode(map[string]interface{}{
"success": false,
@@ -53,14 +53,14 @@ func (h *APIHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set(constants.HeaderContentType, constants.ContentTypeJSON)
json.NewEncoder(w).Encode(map[string]interface{}{
"success": true,
"stats": stats,
})
}
// ServeSpeedHTTP handles the HTTP request for speed data
// 返回速率数据。
func (h *APIHandler) ServeSpeedHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -68,7 +68,7 @@ func (h *APIHandler) ServeSpeedHTTP(w http.ResponseWriter, r *http.Request) {
}
if !h.authenticate(r) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set(constants.HeaderContentType, constants.ContentTypeJSON)
w.WriteHeader(http.StatusUnauthorized)
json.NewEncoder(w).Encode(map[string]interface{}{
"success": false,
@@ -79,7 +79,7 @@ func (h *APIHandler) ServeSpeedHTTP(w http.ResponseWriter, r *http.Request) {
uploadSpeed, downloadSpeed := h.client.GetSpeed()
w.Header().Set("Content-Type", "application/json")
w.Header().Set(constants.HeaderContentType, constants.ContentTypeJSON)
json.NewEncoder(w).Encode(map[string]interface{}{
"success": true,
"upload_speed": uploadSpeed,
@@ -87,9 +87,9 @@ func (h *APIHandler) ServeSpeedHTTP(w http.ResponseWriter, r *http.Request) {
})
}
// authenticate checks if the request is authorized (token + User-Agent)
// 校验请求身份（token + User-Agent）。
func (h *APIHandler) authenticate(r *http.Request) bool {
if r.Header.Get("User-Agent") != config.AgentUserAgent {
if r.Header.Get(constants.HeaderUserAgent) != constants.AgentUserAgent {
return false
}
@@ -97,13 +97,13 @@ func (h *APIHandler) authenticate(r *http.Request) bool {
return true
}
auth := r.Header.Get("Authorization")
auth := r.Header.Get(constants.HeaderAuthorization)
if auth == "" {
return false
}
if strings.HasPrefix(auth, "Bearer ") {
token := strings.TrimPrefix(auth, "Bearer ")
if strings.HasPrefix(auth, constants.BearerPrefix) {
token := strings.TrimPrefix(auth, constants.BearerPrefix)
return token == h.configToken
}
internal/handler/routes.go
+41
View File
@@ -0,0 +1,41 @@
package handler
import (
"net/http"
"mmw-agent/internal/constants"
)
// 注册子端 API 路由
func RegisterChildRoutes(mux *http.ServeMux, apiHandler *APIHandler, manageHandler *ManageHandler) {
// 拉取模式数据接口
mux.HandleFunc(constants.PathChildTraffic, apiHandler.ServeHTTP)
mux.HandleFunc(constants.PathChildSpeed, apiHandler.ServeSpeedHTTP)
// 管理接口
mux.HandleFunc(constants.PathChildServiceStats, manageHandler.HandleServicesStatus)
mux.HandleFunc(constants.PathChildServiceCtl, manageHandler.HandleServiceControl)
mux.HandleFunc(constants.PathChildXrayInstall, manageHandler.HandleXrayInstall)
mux.HandleFunc(constants.PathChildXrayRemove, manageHandler.HandleXrayRemove)
mux.HandleFunc(constants.PathChildXrayConfig, manageHandler.HandleXrayConfig)
mux.HandleFunc(constants.PathChildXraySysCfg, manageHandler.HandleXraySystemConfig)
mux.HandleFunc(constants.PathChildXrayCfgFiles, manageHandler.HandleXrayConfigFiles)
mux.HandleFunc(constants.PathChildNginxInstall, manageHandler.HandleNginxInstall)
mux.HandleFunc(constants.PathChildNginxRemove, manageHandler.HandleNginxRemove)
mux.HandleFunc(constants.PathChildNginxConfig, manageHandler.HandleNginxConfig)
mux.HandleFunc(constants.PathChildNginxCfgFile, manageHandler.HandleNginxConfigFiles)
mux.HandleFunc(constants.PathChildSystemInfo, manageHandler.HandleSystemInfo)
mux.HandleFunc(constants.PathChildInbounds, manageHandler.HandleInbounds)
mux.HandleFunc(constants.PathChildOutbounds, manageHandler.HandleOutbounds)
mux.HandleFunc(constants.PathChildRouting, manageHandler.HandleRouting)
mux.HandleFunc(constants.PathChildScan, manageHandler.HandleScan)
mux.HandleFunc(constants.PathChildCertDeploy, manageHandler.HandleCertDeploy)
mux.HandleFunc(constants.PathChildNginxSetup, manageHandler.HandleNginxSetupSSL)
mux.HandleFunc(constants.PathChildDomainProbe, manageHandler.HandleDomainLatencyProbe)
// SSE 流式安装和卸载接口
mux.HandleFunc(constants.PathChildXrayInstallStream, manageHandler.HandleXrayInstallStream)
mux.HandleFunc(constants.PathChildXrayRemoveStream, manageHandler.HandleXrayRemoveStream)
mux.HandleFunc(constants.PathChildNginxInstallSSE, manageHandler.HandleNginxInstallStream)
mux.HandleFunc(constants.PathChildNginxRemoveSSE, manageHandler.HandleNginxRemoveStream)
}
internal/handler/embed.go → internal/handler/xray_default_config_embed.go
View File
internal/xrpc/client.go
+2 -2
View File
@@ -11,7 +11,7 @@ import (
"google.golang.org/grpc/credentials/insecure"
)
// Clients groups the gRPC stubs that the samples rely on.
// Clients 汇总当前用到的 gRPC 客户端。
type Clients struct {
Connection *grpc.ClientConn
Handler handlerpb.HandlerServiceClient
@@ -19,7 +19,7 @@ type Clients struct {
Stats statspb.StatsServiceClient
}
// New establishes an insecure (plaintext) connection against a running Xray API endpoint.
// 连接到运行中的 Xray API，默认使用明文连接。
func New(ctx context.Context, addr string, port uint16, dialOpts ...grpc.DialOption) (*Clients, error) {
target := fmt.Sprintf("%s:%d", addr, port)
opts := []grpc.DialOption{
internal/xrpc/services/handler/inbound.go
+10 -11
View File
@@ -21,7 +21,7 @@ import (
vmessin "github.com/xtls/xray-core/proxy/vmess/inbound"
)
// AddVMessInbound demonstrates HandlerServiceClient.AddInbound for VMess inbound.
// 添加 VMess 入站示例。
func AddVMessInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
inbound := inboundConfig(
tag,
@@ -46,7 +46,7 @@ func AddVMessInbound(ctx context.Context, client command.HandlerServiceClient, t
return err
}
// AddVLESSInbound adds a VLESS inbound with Vision style fallbacks.
// 添加带 Vision 回落配置的 VLESS 入站。
func AddVLESSInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
inbound := inboundConfig(
tag,
@@ -80,7 +80,7 @@ func AddVLESSInbound(ctx context.Context, client command.HandlerServiceClient, t
return err
}
// AddTrojanInbound registers a Trojan inbound with two users and ALPN fallback.
// 添加带双用户和 ALPN 回落的 Trojan 入站。
func AddTrojanInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
inbound := inboundConfig(
tag,
@@ -115,7 +115,7 @@ func AddTrojanInbound(ctx context.Context, client command.HandlerServiceClient,
return err
}
// AddShadowsocksInbound adds an AEAD Shadowsocks inbound supporting both TCP and UDP.
// 添加支持 TCP/UDP 的 AEAD Shadowsocks 入站。
func AddShadowsocksInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
inbound := inboundConfig(
tag,
@@ -137,7 +137,7 @@ func AddShadowsocksInbound(ctx context.Context, client command.HandlerServiceCli
return err
}
// AddShadowsocks2022Inbound covers both single user and multi-user deployment.
// 添加 Shadowsocks 2022 入站，兼容单用户和多用户场景。
func AddShadowsocks2022Inbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
server := &ss2022.MultiUserServerConfig{
Method: "2022-blake3-aes-128-gcm",
@@ -168,7 +168,7 @@ func AddShadowsocks2022Inbound(ctx context.Context, client command.HandlerServic
return err
}
// AddSocksInbound exposes a SOCKS5 server with username/password authentication.
// 添加带账号密码认证的 SOCKS5 入站。
func AddSocksInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
inbound := inboundConfig(
tag,
@@ -184,7 +184,7 @@ func AddSocksInbound(ctx context.Context, client command.HandlerServiceClient, t
return err
}
// AddHTTPInbound adds an HTTP proxy inbound with basic auth.
// 添加带基础认证的 HTTP 入站。
func AddHTTPInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
inbound := inboundConfig(
tag,
@@ -199,7 +199,7 @@ func AddHTTPInbound(ctx context.Context, client command.HandlerServiceClient, ta
return err
}
// AddDokodemoInbound configures a dokodemo-door mirror port.
// 添加 dokodemo-door 入站并转发到目标端口。
func AddDokodemoInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32, targetPort uint32) error {
inbound := inboundConfig(
tag,
@@ -216,7 +216,7 @@ func AddDokodemoInbound(ctx context.Context, client command.HandlerServiceClient
return err
}
// AddDNSInbound exposes the built-in DNS server on an API port.
// 添加内置 DNS 入站。
func AddDNSInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32) error {
inbound := inboundConfig(
tag,
@@ -235,7 +235,7 @@ func AddDNSInbound(ctx context.Context, client command.HandlerServiceClient, tag
return err
}
// AddLoopbackInbound ties an inbound to an existing outbound chain.
// 添加 loopback 入站并绑定已有出站链路。
func AddLoopbackInbound(ctx context.Context, client command.HandlerServiceClient, tag string, port uint32, targetInbound string) error {
inbound := inboundConfig(
tag,
@@ -245,4 +245,3 @@ func AddLoopbackInbound(ctx context.Context, client command.HandlerServiceClient
_, err := client.AddInbound(ctx, &command.AddInboundRequest{Inbound: inbound})
return err
}
internal/xrpc/services/handler/outbound.go
-1
View File
@@ -195,4 +195,3 @@ func AddVMessOutbound(ctx context.Context, client command.HandlerServiceClient,
_, err := client.AddOutbound(ctx, &command.AddOutboundRequest{Outbound: cfg})
return err
}
internal/xrpc/services/handler/users.go
+6 -6
View File
@@ -13,7 +13,7 @@ import (
"github.com/xtls/xray-core/proxy/vmess"
)
// AddVMessUser demonstrates AlterInbound(AddUserOperation) for VMess.
// 通过 AlterInbound(AddUserOperation) 为 VMess 入站加用户。
func AddVMessUser(ctx context.Context, client command.HandlerServiceClient, inboundTag, email string) error {
req := &command.AlterInboundRequest{
Tag: inboundTag,
@@ -34,7 +34,7 @@ func AddVMessUser(ctx context.Context, client command.HandlerServiceClient, inbo
return err
}
// AddVLESSUser shows how to add VLESS users dynamically.
// 为 VLESS 入站动态加用户。
func AddVLESSUser(ctx context.Context, client command.HandlerServiceClient, inboundTag, email string) error {
req := &command.AlterInboundRequest{
Tag: inboundTag,
@@ -53,7 +53,7 @@ func AddVLESSUser(ctx context.Context, client command.HandlerServiceClient, inbo
return err
}
// AddTrojanUser adds a Trojan password to an inbound handler.
// 为 Trojan 入站加密码用户。
func AddTrojanUser(ctx context.Context, client command.HandlerServiceClient, inboundTag, email, password string) error {
req := &command.AlterInboundRequest{
Tag: inboundTag,
@@ -71,7 +71,7 @@ func AddTrojanUser(ctx context.Context, client command.HandlerServiceClient, inb
return err
}
// AddShadowsocksUser sets up a Shadowsocks AEAD credential.
// 为 Shadowsocks 入站加 AEAD 凭据。
func AddShadowsocksUser(ctx context.Context, client command.HandlerServiceClient, inboundTag, email, password string) error {
req := &command.AlterInboundRequest{
Tag: inboundTag,
@@ -90,7 +90,7 @@ func AddShadowsocksUser(ctx context.Context, client command.HandlerServiceClient
return err
}
// AddShadowsocks2022User covers key rotation for SS2022.
// 为 SS2022 入站新增用户密钥。
func AddShadowsocks2022User(ctx context.Context, client command.HandlerServiceClient, inboundTag, email string) error {
req := &command.AlterInboundRequest{
Tag: inboundTag,
@@ -107,7 +107,7 @@ func AddShadowsocks2022User(ctx context.Context, client command.HandlerServiceCl
return err
}
// RemoveUser removes any user (identified by email) from an inbound.
// 按邮箱从入站移除用户。
func RemoveUser(ctx context.Context, client command.HandlerServiceClient, inboundTag, email string) error {
req := &command.AlterInboundRequest{
Tag: inboundTag,
internal/xrpc/services/logger/logger.go
+4 -3
View File
@@ -2,14 +2,15 @@ package logger
import (
"context"
"time"
"mmw-agent/internal/constants"
loggerpb "github.com/xtls/xray-core/app/log/command"
)
// RestartLogger triggers the LoggerService restartLogger RPC and waits for completion.
// 调用 LoggerService 的 restartLogger 接口并等待完成。
func RestartLogger(ctx context.Context, client loggerpb.LoggerServiceClient) error {
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
ctx, cancel := context.WithTimeout(ctx, constants.DefaultRPCShortTimeout)
defer cancel()
_, err := client.RestartLogger(ctx, &loggerpb.RestartLoggerRequest{})
return err
internal/xrpc/services/stats/stats.go
+4 -3
View File
@@ -2,13 +2,14 @@ package stats
import (
"context"
"time"
"mmw-agent/internal/constants"
statspb "github.com/xtls/xray-core/app/stats/command"
)
func QueryTraffic(ctx context.Context, client statspb.StatsServiceClient, pattern string, reset bool) (int64, error) {
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
ctx, cancel := context.WithTimeout(ctx, constants.DefaultRPCShortTimeout)
defer cancel()
resp, err := client.QueryStats(ctx, &statspb.QueryStatsRequest{
Pattern: pattern,
@@ -24,7 +25,7 @@ func QueryTraffic(ctx context.Context, client statspb.StatsServiceClient, patter
}
func GetSystemStats(ctx context.Context, client statspb.StatsServiceClient) (*statspb.SysStatsResponse, error) {
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
ctx, cancel := context.WithTimeout(ctx, constants.DefaultRPCShortTimeout)
defer cancel()
return client.GetSysStats(ctx, &statspb.SysStatsRequest{})
}