From 828c0b54cabb9764bdb68297806349cce07018ad Mon Sep 17 00:00:00 2001 From: dushixiang Date: Wed, 8 Jul 2026 22:33:43 +0800 Subject: [PATCH] Remove agent metrics buffer --- pkg/agent/service/agent.go | 141 +++++++--------- pkg/agent/service/metrics_buffer.go | 247 ---------------------------- pkg/agent/service/migration.go | 19 +++ pkg/agent/service/service.go | 16 +- 4 files changed, 84 insertions(+), 339 deletions(-) delete mode 100644 pkg/agent/service/metrics_buffer.go create mode 100644 pkg/agent/service/migration.go diff --git a/pkg/agent/service/agent.go b/pkg/agent/service/agent.go index 7c645d2..cd085b8 100644 --- a/pkg/agent/service/agent.go +++ b/pkg/agent/service/agent.go @@ -81,7 +81,6 @@ type Agent struct { activeConn *safeConn collectorMu sync.RWMutex collectorManager *collector.Manager - outboundBuffer *outboundBuffer tamperProtector *tamper.Protector sshMonitor *sshmonitor.Monitor } @@ -92,7 +91,6 @@ func New(cfg *config.Config) *Agent { cfg: cfg, idMgr: id.NewManager(), collectorManager: collector.NewManager(cfg), - outboundBuffer: newOutboundBuffer(), tamperProtector: tamper.NewProtector(), sshMonitor: sshmonitor.NewMonitor(), } @@ -177,7 +175,7 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error { // 连接到服务器 rawConn, _, err := dialer.Dial(wsURL, nil) if err != nil { - return fmt.Errorf("连接失败: %w", err) + return fmt.Errorf("connect failed: %w", err) } defer rawConn.Close() @@ -185,7 +183,7 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error { conn := &safeConn{conn: rawConn} if err := rawConn.SetReadDeadline(time.Now().Add(agentPongWait)); err != nil { - return fmt.Errorf("设置读取超时失败: %w", err) + return fmt.Errorf("set read deadline failed: %w", err) } rawConn.SetPongHandler(func(string) error { return rawConn.SetReadDeadline(time.Now().Add(agentPongWait)) @@ -201,7 +199,7 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error { // 发送注册消息 if err := a.registerAgent(conn); err != nil { - return fmt.Errorf("注册失败: %w", err) + return fmt.Errorf("register failed: %w", err) } onRegistered() @@ -223,7 +221,7 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error { wg.Go(func() { if err := a.readLoop(rawConn, done); err != nil { select { - case errChan <- fmt.Errorf("读取失败: %w", err): + case errChan <- fmt.Errorf("read failed: %w", err): default: } } @@ -233,7 +231,7 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error { wg.Go(func() { if err := a.pingLoop(ctx, conn, done); err != nil { select { - case errChan <- fmt.Errorf("ping失败: %w", err): + case errChan <- fmt.Errorf("ping failed: %w", err): default: } } @@ -346,7 +344,7 @@ func (a *Agent) registerAgent(conn *safeConn) error { // 加载或生成探针 ID agentID, err := a.idMgr.Load() if err != nil { - return fmt.Errorf("加载 agent ID 失败: %w", err) + return fmt.Errorf("load agent ID failed: %w", err) } slog.Info("Agent ID", "id", agentID, "path", a.idMgr.GetPath()) @@ -379,31 +377,31 @@ func (a *Agent) registerAgent(conn *safeConn) error { Type: protocol.MessageTypeRegister, Data: registerReq, }); err != nil { - return fmt.Errorf("发送注册消息失败: %w", err) + return fmt.Errorf("send register message failed: %w", err) } // 读取注册响应 var response protocol.InputMessage if err := conn.ReadJSON(&response); err != nil { - return fmt.Errorf("读取注册响应失败: %w", err) + return fmt.Errorf("read register response failed: %w", err) } // 检查响应类型 if response.Type == protocol.MessageTypeRegisterErr { var errResp protocol.RegisterResponse if err := json.Unmarshal(response.Data, &errResp); err == nil { - return fmt.Errorf("注册失败: %s", errResp.Message) + return fmt.Errorf("register failed: %s", errResp.Message) } - return fmt.Errorf("注册失败: 未知错误") + return fmt.Errorf("register failed: unknown error") } if response.Type != protocol.MessageTypeRegisterAck { - return fmt.Errorf("注册失败: 收到未知响应类型 %s", response.Type) + return fmt.Errorf("register failed: unexpected response type %s", response.Type) } var registerResp protocol.RegisterResponse if err := json.Unmarshal(response.Data, ®isterResp); err != nil { - return fmt.Errorf("解析注册响应失败: %w", err) + return fmt.Errorf("parse register response failed: %w", err) } slog.Info("注册成功", "agentId", registerResp.AgentID, "status", registerResp.Status) @@ -432,14 +430,19 @@ func (a *Agent) handleMonitorConfig(data json.RawMessage) { // 立即执行一次监控检测，结果以单元素 batch 上报 sample := manager.CollectMonitor(payload.Items) - writer := newOutboundWriter(a.getActiveConn(), a.outboundBuffer) - if err := writer.WriteJSON(protocol.OutboundMessage{ + conn := a.getActiveConn() + if conn == nil { + slog.Warn("当前连接不可用，无法上报监控结果") + return + } + + if err := conn.WriteJSON(protocol.OutboundMessage{ Type: protocol.MessageTypeMetrics, Data: protocol.MetricsBatch{Samples: []protocol.MetricSample{sample}}, }); err != nil { slog.Warn("发送监控结果失败", "error", err) } else { - slog.Info("服务监控检测完成，已上报或缓存监控项结果", "count", len(payload.Items)) + slog.Info("服务监控检测完成，已上报监控项结果", "count", len(payload.Items)) } } @@ -455,23 +458,17 @@ func (a *Agent) getActiveConn() *safeConn { return a.activeConn } -func (a *Agent) sendOutboundMessage(msg protocol.OutboundMessage) (bool, error) { +func (a *Agent) sendOutboundMessage(msg protocol.OutboundMessage) error { conn := a.getActiveConn() if conn == nil { - if err := a.outboundBuffer.Append(msg); err != nil { - return false, err - } - return true, nil + return fmt.Errorf("connection is unavailable") } if err := conn.WriteJSON(msg); err != nil { - if bufferErr := a.outboundBuffer.Append(msg); bufferErr != nil { - return false, fmt.Errorf("发送失败且写入缓存失败: %w", bufferErr) - } - return true, nil + return err } - return false, nil + return nil } func (a *Agent) setCollectorManager(manager *collector.Manager) { @@ -497,22 +494,21 @@ func (a *Agent) metricsLoop(ctx context.Context) { a.setCollectorManager(manager) } - // 立即采集一次（tick=0 时所有采集器都到期，等于全量采集） var tickCount uint64 - if err := a.collectAndSendAllMetrics(manager, tickCount); err != nil { - slog.Warn("初始数据采集失败", "error", err) - } - ticker := time.NewTicker(collectorBaseInterval) defer ticker.Stop() for { select { case <-ticker.C: - tickCount++ + if a.getActiveConn() == nil { + continue + } + if err := a.collectAndSendAllMetrics(manager, tickCount); err != nil { slog.Warn("数据采集失败", "error", err) } + tickCount++ case <-ctx.Done(): return } @@ -522,7 +518,7 @@ func (a *Agent) metricsLoop(ctx context.Context) { // collectAndSendAllMetrics 采集并发送所有动态指标 func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager, tickCount uint64) error { if manager == nil { - return fmt.Errorf("采集器未初始化") + return fmt.Errorf("collector is not initialized") } type result struct { @@ -538,7 +534,7 @@ func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager, tickCount u case r := <-done: return r.err case <-time.After(agentCollectTimeout): - return fmt.Errorf("指标采集超时 (超过 %v)", agentCollectTimeout) + return fmt.Errorf("metrics collection timed out after %v", agentCollectTimeout) } } @@ -566,13 +562,6 @@ type collectFn func() (protocol.MetricSample, error) // 各采集器按自身 interval 节流；本 tick 到期的采集器并行执行，避免串行累计耗时压垮 1s 节奏 func (a *Agent) doCollectAndSend(manager *collector.Manager, tickCount uint64) error { conn := a.getActiveConn() - if conn != nil { - if sent, err := a.outboundBuffer.Flush(conn); err != nil { - slog.Warn("发送缓存消息失败", "error", err) - } else if sent > 0 { - slog.Info("已发送缓存消息", "count", sent) - } - } // required=true 时采集失败计入错误；required=false 适用于 GPU/温度等可选项 // interval 必须为 collectorBaseInterval 的整数倍 @@ -650,30 +639,23 @@ func (a *Agent) doCollectAndSend(manager *collector.Manager, tickCount uint64) e samples = append(samples, r.sample) } - writer := newOutboundWriter(conn, a.outboundBuffer) - if len(samples) > 0 { - if err := writer.WriteJSON(protocol.OutboundMessage{ + if conn == nil { + return fmt.Errorf("connection is unavailable") + } + + if err := conn.WriteJSON(protocol.OutboundMessage{ Type: protocol.MessageTypeMetrics, Data: protocol.MetricsBatch{Samples: samples}, }); err != nil { slog.Warn("发送指标 batch 失败", "error", err) - hasError = true - } - } - - if writer.buffered { - if conn == nil { - slog.Info("当前连接不可用，消息已写入缓存") - } else if writer.sendErr != nil { - slog.Warn("发送消息失败，已写入缓存", "error", writer.sendErr) // 连接写失败视为连接异常，返回错误触发重连 - return fmt.Errorf("连接发送失败: %w", writer.sendErr) + return fmt.Errorf("connection send failed: %w", err) } } if hasError { - return fmt.Errorf("部分指标采集失败") + return fmt.Errorf("some metrics collection failed") } return nil @@ -738,7 +720,7 @@ func (a *Agent) sendCommandResponse(cmdID, cmdType, status, errMsg, result strin Result: result, } - if _, err := a.sendOutboundMessage(protocol.OutboundMessage{ + if err := a.sendOutboundMessage(protocol.OutboundMessage{ Type: protocol.MessageTypeCommandResp, Data: resp, }); err != nil { @@ -807,7 +789,7 @@ func (a *Agent) sendTamperProtectResponse(success bool, message string, paths [] Removed: removed, } - if _, err := a.sendOutboundMessage(protocol.OutboundMessage{ + if err := a.sendOutboundMessage(protocol.OutboundMessage{ Type: protocol.MessageTypeTamperProtect, Data: resp, }); err != nil { @@ -835,14 +817,11 @@ func (a *Agent) tamperEventLoop(ctx context.Context, done chan struct{}) { Details: event.Details, } - buffered, err := a.sendOutboundMessage(protocol.OutboundMessage{ + if err := a.sendOutboundMessage(protocol.OutboundMessage{ Type: protocol.MessageTypeTamperEvent, Data: eventData, - }) - if err != nil { + }); err != nil { slog.Warn("发送防篡改事件失败", "error", err) - } else if buffered { - slog.Info("防篡改事件已缓存", "path", event.Path, "operation", event.Operation) } else { slog.Info("已上报防篡改事件", "path", event.Path, "operation", event.Operation) } @@ -856,18 +835,11 @@ func (a *Agent) tamperEventLoop(ctx context.Context, done chan struct{}) { Restored: alert.Restored, } - buffered, err := a.sendOutboundMessage(protocol.OutboundMessage{ + if err := a.sendOutboundMessage(protocol.OutboundMessage{ Type: protocol.MessageTypeTamperEvent, Data: eventData, - }) - if err != nil { + }); err != nil { slog.Warn("发送防篡改告警失败", "error", err) - } else if buffered { - status := "未恢复" - if alert.Restored { - status = "已恢复" - } - slog.Info("防篡改告警已缓存", "path", alert.Path, "status", status) } else { status := "未恢复" if alert.Restored { @@ -904,7 +876,7 @@ func (a *Agent) handleDDNSConfig(data json.RawMessage) { if err := a.collectAndSendDDNSIP(manager, &ddnsConfig); err != nil { slog.Warn("DDNS IP 采集失败", "error", err) } else { - slog.Info("DDNS IP 地址已上报或缓存") + slog.Info("DDNS IP 地址已上报") } } @@ -953,7 +925,7 @@ func (a *Agent) handlePublicIPConfig(data json.RawMessage) { return } - if _, err := a.sendOutboundMessage(protocol.OutboundMessage{ + if err := a.sendOutboundMessage(protocol.OutboundMessage{ Type: protocol.MessageTypePublicIPReport, Data: report, }); err != nil { @@ -973,7 +945,7 @@ func (a *Agent) getPublicIPFromAPIs(manager *collector.Manager, apis []string, i continue } if !strings.HasPrefix(api, "http://") && !strings.HasPrefix(api, "https://") { - lastErr = fmt.Errorf("非法 API 地址: %s", api) + lastErr = fmt.Errorf("invalid API URL: %s", api) continue } ip, err := manager.GetPublicIP(api, isIPv6) @@ -986,7 +958,7 @@ func (a *Agent) getPublicIPFromAPIs(manager *collector.Manager, apis []string, i if lastErr != nil { return "", lastErr } - return "", fmt.Errorf("未能获取公网 IP 地址") + return "", fmt.Errorf("failed to get public IP address") } // collectAndSendDDNSIP 采集并发送 DDNS IP 地址 @@ -1017,14 +989,14 @@ func (a *Agent) collectAndSendDDNSIP(manager *collector.Manager, config *protoco // 如果没有获取到任何 IP，返回错误 if ipReport.IPv4 == "" && ipReport.IPv6 == "" { - return fmt.Errorf("未获取到任何 IP 地址") + return fmt.Errorf("no IP address was collected") } - if _, err := a.sendOutboundMessage(protocol.OutboundMessage{ + if err := a.sendOutboundMessage(protocol.OutboundMessage{ Type: protocol.MessageTypeDDNSIPReport, Data: ipReport, }); err != nil { - return fmt.Errorf("发送 IP 报告失败: %w", err) + return fmt.Errorf("send IP report failed: %w", err) } return nil @@ -1040,7 +1012,7 @@ func (a *Agent) getIPAddress(manager *collector.Manager, method, value string, i // 从网络接口获取 IP return manager.GetInterfaceIP(value, isIPv6) default: - return "", fmt.Errorf("不支持的 IP 获取方式: %s", method) + return "", fmt.Errorf("unsupported IP get method: %s", method) } } @@ -1112,7 +1084,7 @@ func (a *Agent) sendSSHLoginConfigResult(success bool, enabled bool, message str Data: result, } - if _, err := a.sendOutboundMessage(msg); err != nil { + if err := a.sendOutboundMessage(msg); err != nil { slog.Warn("发送SSH登录监控配置应用结果失败", "error", err) } } @@ -1129,14 +1101,11 @@ func (a *Agent) sshLoginEventLoop(ctx context.Context, done chan struct{}) { return case event := <-eventCh: // 上报到服务端 - buffered, err := a.sendOutboundMessage(protocol.OutboundMessage{ + if err := a.sendOutboundMessage(protocol.OutboundMessage{ Type: protocol.MessageTypeSSHLoginEvent, Data: event, - }) - if err != nil { + }); err != nil { slog.Warn("发送SSH登录事件失败", "error", err) - } else if buffered { - slog.Info("SSH登录事件已缓存", "user", event.Username, "ip", event.IP, "status", event.Status) } else { slog.Info("已上报SSH登录事件", "user", event.Username, "ip", event.IP, "status", event.Status) } diff --git a/pkg/agent/service/metrics_buffer.go b/pkg/agent/service/metrics_buffer.go deleted file mode 100644 index 631a12d..0000000 --- a/pkg/agent/service/metrics_buffer.go +++ /dev/null @@ -1,247 +0,0 @@ -package service - -import ( - "encoding/binary" - "encoding/json" - "fmt" - "log/slog" - "os" - "path/filepath" - "sync" - "time" - - "github.com/dushixiang/pika/internal/protocol" - "github.com/dushixiang/pika/pkg/agent/utils" - bolt "go.etcd.io/bbolt" -) - -const ( - outboundBufferDBName = "metrics_buffer.db" - outboundBufferBucket = "metrics_buffer" - outboundBufferTimeout = 2 * time.Second - outboundBufferRetention = 24 * time.Hour -) - -// WebSocketWriter 抽象 JSON 写入操作，由 outboundBuffer 和 outboundWriter 共用 -type WebSocketWriter interface { - WriteJSON(v interface{}) error -} - -type outboundBuffer struct { - path string - mu sync.Mutex -} - -type bufferedMessage struct { - Timestamp int64 `json:"ts"` - Payload json.RawMessage `json:"payload"` -} - -func newOutboundBuffer() *outboundBuffer { - path := filepath.Join(utils.GetSafeHomeDir(), ".pika", outboundBufferDBName) - return &outboundBuffer{path: path} -} - -func (b *outboundBuffer) Append(v interface{}) error { - payload, err := json.Marshal(v) - if err != nil { - return fmt.Errorf("序列化缓存消息失败: %w", err) - } - - wrapped, err := json.Marshal(bufferedMessage{ - Timestamp: time.Now().UnixMilli(), - Payload: payload, - }) - if err != nil { - return fmt.Errorf("序列化缓存封装失败: %w", err) - } - - b.mu.Lock() - defer b.mu.Unlock() - - db, err := b.openDB() - if err != nil { - return err - } - defer db.Close() - - if err := db.Update(func(tx *bolt.Tx) error { - bucket, err := tx.CreateBucketIfNotExists([]byte(outboundBufferBucket)) - if err != nil { - return fmt.Errorf("创建缓存桶失败: %w", err) - } - - seq, err := bucket.NextSequence() - if err != nil { - return fmt.Errorf("获取缓存序列失败: %w", err) - } - - key := make([]byte, 8) - binary.BigEndian.PutUint64(key, seq) - if err := bucket.Put(key, wrapped); err != nil { - return fmt.Errorf("写入缓存失败: %w", err) - } - - cutoff := time.Now().Add(-outboundBufferRetention).UnixMilli() - cursor := bucket.Cursor() - for k, v := cursor.First(); k != nil; k, v = cursor.Next() { - var item bufferedMessage - if err := json.Unmarshal(v, &item); err != nil || item.Timestamp == 0 { - break - } - if item.Timestamp >= cutoff { - break - } - if err := cursor.Delete(); err != nil { - return fmt.Errorf("删除过期缓存失败: %w", err) - } - } - - return nil - }); err != nil { - return err - } - - return nil -} - -func (b *outboundBuffer) Flush(writer WebSocketWriter) (int, error) { - b.mu.Lock() - defer b.mu.Unlock() - - db, err := b.openDB() - if err != nil { - if os.IsNotExist(err) { - return 0, nil - } - return 0, err - } - defer db.Close() - - var ( - sent int - sendErr error - ) - cutoff := time.Now().Add(-outboundBufferRetention).UnixMilli() - - if err := db.Update(func(tx *bolt.Tx) error { - bucket := tx.Bucket([]byte(outboundBufferBucket)) - if bucket == nil { - return nil - } - - cursor := bucket.Cursor() - for k, v := cursor.First(); k != nil; k, v = cursor.Next() { - if sendErr != nil { - break - } - - var wrapped bufferedMessage - if err := json.Unmarshal(v, &wrapped); err == nil && len(wrapped.Payload) > 0 { - if wrapped.Timestamp > 0 && wrapped.Timestamp < cutoff { - if err := cursor.Delete(); err != nil { - return fmt.Errorf("删除过期缓存失败: %w", err) - } - continue - } - - var msg protocol.OutboundMessage - if err := json.Unmarshal(wrapped.Payload, &msg); err != nil { - slog.Warn("缓存消息解析失败，已跳过", "error", err) - if err := cursor.Delete(); err != nil { - return fmt.Errorf("删除损坏缓存失败: %w", err) - } - continue - } - - if err := writer.WriteJSON(msg); err != nil { - sendErr = err - break - } - - if err := cursor.Delete(); err != nil { - return fmt.Errorf("删除已发送缓存失败: %w", err) - } - sent++ - continue - } - - var legacy protocol.OutboundMessage - if err := json.Unmarshal(v, &legacy); err != nil { - slog.Warn("缓存消息解析失败，已跳过", "error", err) - if err := cursor.Delete(); err != nil { - return fmt.Errorf("删除损坏缓存失败: %w", err) - } - continue - } - - if err := writer.WriteJSON(legacy); err != nil { - sendErr = err - break - } - - if err := cursor.Delete(); err != nil { - return fmt.Errorf("删除已发送缓存失败: %w", err) - } - sent++ - } - - return nil - }); err != nil { - return sent, err - } - - if sendErr != nil { - return sent, sendErr - } - - return sent, nil -} - -func (b *outboundBuffer) openDB() (*bolt.DB, error) { - if err := os.MkdirAll(filepath.Dir(b.path), 0755); err != nil { - return nil, fmt.Errorf("创建缓存目录失败: %w", err) - } - - db, err := bolt.Open(b.path, 0600, &bolt.Options{Timeout: outboundBufferTimeout}) - if err != nil { - return nil, fmt.Errorf("打开缓存数据库失败: %w", err) - } - - return db, nil -} - -type outboundWriter struct { - conn *safeConn - buffer *outboundBuffer - buffered bool - sendErr error -} - -func newOutboundWriter(conn *safeConn, buffer *outboundBuffer) *outboundWriter { - return &outboundWriter{ - conn: conn, - buffer: buffer, - } -} - -func (w *outboundWriter) WriteJSON(v interface{}) error { - if w.conn == nil { - if err := w.buffer.Append(v); err != nil { - return err - } - w.buffered = true - return nil - } - - if err := w.conn.WriteJSON(v); err != nil { - if bufferErr := w.buffer.Append(v); bufferErr != nil { - return fmt.Errorf("写入缓存失败: %w", bufferErr) - } - w.buffered = true - w.sendErr = err - return err - } - - return nil -} diff --git a/pkg/agent/service/migration.go b/pkg/agent/service/migration.go new file mode 100644 index 0000000..0d34cf1 --- /dev/null +++ b/pkg/agent/service/migration.go @@ -0,0 +1,19 @@ +package service + +import ( + "fmt" + "os" + "path/filepath" + + "github.com/dushixiang/pika/pkg/agent/utils" +) + +const legacyMetricsBufferDBName = "metrics_buffer.db" + +func cleanupLegacyMetricsBuffer() error { + path := filepath.Join(utils.GetSafeHomeDir(), ".pika", legacyMetricsBufferDBName) + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("remove legacy metrics buffer failed: %w", err) + } + return nil +} diff --git a/pkg/agent/service/service.go b/pkg/agent/service/service.go index 4fd9cad..96bfc07 100644 --- a/pkg/agent/service/service.go +++ b/pkg/agent/service/service.go @@ -37,6 +37,10 @@ func configureICMP() { // startAgent 启动 Agent 和自动更新（抽取通用逻辑） func startAgent(ctx context.Context, cfg *config.Config) *Agent { + if err := cleanupLegacyMetricsBuffer(); err != nil { + slog.Warn("failed to clean legacy metrics buffer", "error", err) + } + // 创建 Agent 实例 a := New(cfg) @@ -147,7 +151,7 @@ func NewServiceManager(cfg *config.Config) (*ServiceManager, error) { // 获取可执行文件路径 execPath, err := os.Executable() if err != nil { - return nil, fmt.Errorf("获取可执行文件路径失败: %w", err) + return nil, fmt.Errorf("get executable path failed: %w", err) } var options = service.KeyValue{ @@ -193,7 +197,7 @@ func NewServiceManager(cfg *config.Config) (*ServiceManager, error) { // 创建服务 s, err := service.New(prg, svcConfig) if err != nil { - return nil, fmt.Errorf("创建服务失败: %w", err) + return nil, fmt.Errorf("create service failed: %w", err) } return &ServiceManager{ @@ -323,13 +327,13 @@ func UninstallAgent(cfgPath string) error { // 加载配置 cfg, err := config.Load(cfgPath) if err != nil { - return fmt.Errorf("加载配置失败: %w", err) + return fmt.Errorf("load config failed: %w", err) } // 创建服务管理器 mgr, err := NewServiceManager(cfg) if err != nil { - return fmt.Errorf("创建服务管理器失败: %w", err) + return fmt.Errorf("create service manager failed: %w", err) } // 检查服务状态，如果在运行则停止 @@ -338,13 +342,13 @@ func UninstallAgent(cfgPath string) error { slog.Warn("获取服务状态失败", "error", err) } else if status != "已停止 (Stopped)" { if err := mgr.Stop(); err != nil { - return fmt.Errorf("停止服务失败: %w", err) + return fmt.Errorf("stop service failed: %w", err) } } // 卸载服务 if err := mgr.Uninstall(); err != nil { - return fmt.Errorf("卸载服务失败: %w", err) + return fmt.Errorf("uninstall service failed: %w", err) } // 清理 SSH 监控配置