aiops/backend/pkg/remoteexec/persistent_client.go

// persistent_client.go - 任务级 SSH 长连接，多次 Exec 复用同一 client // Logan.Li
package remoteexec

import (
	"bytes"
	"context"
	"fmt"
	"sync"
	"time"

	"golang.org/x/crypto/ssh"
)

// PersistentClient 可复用的 SSH 长连接（任务生命周期内持有，Close 释放） // Logan.Li
type PersistentClient struct {
	cfg    SSHConfig
	client *ssh.Client
	mu     sync.Mutex
}

// NewPersistentClient 建立 SSH 长连接 // Logan.Li
func NewPersistentClient(cfg SSHConfig) (*PersistentClient, error) {
	client, err := connect(cfg)
	if err != nil {
		return nil, err
	}
	return &PersistentClient{cfg: cfg, client: client}, nil
}

// Exec 在长连接上执行命令；连接断开时自动重连一次 // Logan.Li
func (p *PersistentClient) Exec(command string, timeout time.Duration) (*ExecResult, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	result, err := p.execLocked(command, timeout)
	if err == nil {
		return result, nil
	}
	// 连接级失败：重连后重试一次 // Logan.Li
	if p.client != nil {
		_ = p.client.Close()
		p.client = nil
	}
	client, connErr := connect(p.cfg)
	if connErr != nil {
		return result, err
	}
	p.client = client
	return p.execLocked(command, timeout)
}

// execLocked 假定已持锁，在当前 client 上执行 // Logan.Li
func (p *PersistentClient) execLocked(command string, timeout time.Duration) (*ExecResult, error) {
	if p.client == nil {
		client, err := connect(p.cfg)
		if err != nil {
			return nil, err
		}
		p.client = client
	}
	return execWithClient(p.client, p.cfg, command, timeout)
}

// Close 关闭长连接 // Logan.Li
func (p *PersistentClient) Close() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.client == nil {
		return nil
	}
	err := p.client.Close()
	p.client = nil
	return err
}

// execWithClient 在已有 SSH client 上执行命令（供 Exec / PersistentClient 共用） // Logan.Li
func execWithClient(client *ssh.Client, cfg SSHConfig, command string, timeout time.Duration) (*ExecResult, error) {
	start := time.Now()
	result := &ExecResult{Host: cfg.Host, Port: cfg.Port}

	session, err := client.NewSession()
	if err != nil {
		result.Duration = time.Since(start).Milliseconds()
		result.Error = fmt.Sprintf("创建SSH会话失败: %v", err)
		return result, err
	}
	defer session.Close()

	if timeout <= 0 {
		timeout = 60 * time.Second
	}

	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	type execOut struct {
		output []byte
		err    error
	}
	ch := make(chan execOut, 1)

	go func() {
		var stdout, stderr bytes.Buffer
		session.Stdout = &stdout
		session.Stderr = &stderr
		runErr := session.Run(command)

		output := stdout.Bytes()
		if stderr.Len() > 0 {
			if len(output) > 0 {
				output = append(output, '\n')
			}
			output = append(output, stderr.Bytes()...)
		}
		ch <- execOut{output: output, err: runErr}
	}()

	select {
	case <-ctx.Done():
		result.Duration = time.Since(start).Milliseconds()
		result.Error = fmt.Sprintf("执行超时（%v）", timeout)
		result.ExitCode = -1
		return result, fmt.Errorf("exec timeout after %v", timeout)
	case out := <-ch:
		result.Duration = time.Since(start).Milliseconds()
		result.Output = string(out.output)

		if out.err != nil {
			if exitErr, ok := out.err.(*ssh.ExitError); ok {
				result.ExitCode = exitErr.ExitStatus()
				result.Success = false
			} else {
				result.ExitCode = -1
				result.Error = out.err.Error()
				result.Success = false
				return result, out.err
			}
		} else {
			result.ExitCode = 0
			result.Success = true
		}
	}
	return result, nil
}