使用Golang Kubernetes客户端执行Pod远程命令如何实现流式输出读取

你当前的实现使用bytes.Buffer作为exec输出的接收载体，exec.Stream方法会阻塞到命令完全执行结束后才会把所有输出写入Buffer，因此只能一次性读取全量内容，无法实现流式逐行获取。

核心是使用io.Pipe替换bytes.Buffer，利用管道的同步读写特性，实时接收远程命令的输出，再通过bufio.Scanner实现按行拆分处理。你可以通过回调函数或者输出通道的方式把实时行数据暴露给调用方。

这里提供支持自定义行处理回调的实现版本，你可以直接在回调里实现逐行打印、存储等逻辑：

import (
    "bufio"
    "context"
    "errors"
    "fmt"
    "io"
    "sync"

    v1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/client-go/tools/remotecommand"
)

// 行处理回调函数定义：参数分别为行内容、是否为stderr输出
type LineHandler func(line string, isStderr bool)

func (k *KubernetesClient) RunCommandStream(ctx context.Context, args *RunCommandArgs, onLine LineHandler) (int, error) {
    req := k.clientset.CoreV1().RESTClient().Post().Resource("pods").Name(args.ContainerId).Namespace(k.namespace).SubResource("exec").Param("container", CONTAINER_NAME)
    scheme := runtime.NewScheme()
    if err := v1.AddToScheme(scheme); err != nil {
        return 0, fmt.Errorf("could not add to scheme: %w", err)
    }
    parameterCodec := runtime.NewParameterCodec(scheme)
    req.VersionedParams(&v1.PodExecOptions{
        Stdin:     false,
        Stdout:    true,
        Stderr:    true,
        TTY:       false,
        Container: args.ContainerId,
        Command:   []string{"sh", "-c", args.Command},
    }, parameterCodec)
    exec, err := remotecommand.NewSPDYExecutor(k.config, "POST", req.URL())
    if err != nil {
        return 0, fmt.Errorf("could not exec command: %w", err)
    }

    // 创建stdout、stderr管道
    stdoutReader, stdoutWriter := io.Pipe()
    stderrReader, stderrWriter := io.Pipe()
    defer func() {
        _ = stdoutReader.Close()
        _ = stderrReader.Close()
    }()

    // 启动协程异步处理stdout输出
    var wg sync.WaitGroup
    wg.Add(2)
    go func() {
        defer wg.Done()
        scanner := bufio.NewScanner(stdoutReader)
        // 若有超长行需求，可自定义缓冲区大小：scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
        for scanner.Scan() {
            if onLine != nil {
                onLine(scanner.Text(), false)
            }
        }
    }()

    // 启动协程异步处理stderr输出
    go func() {
        defer wg.Done()
        scanner := bufio.NewScanner(stderrReader)
        for scanner.Scan() {
            if onLine != nil {
                onLine(scanner.Text(), true)
            }
        }
    }()

    // 执行命令流，高版本client-go可替换为StreamWithContext绑定ctx生命周期
    streamErr := exec.Stream(remotecommand.StreamOptions{
        Stdin:  nil,
        Stdout: stdoutWriter,
        Stderr: stderrWriter,
        Tty:    false,
    })

    // 命令执行完关闭写入端，让读取协程正常退出
    _ = stdoutWriter.Close()
    _ = stderrWriter.Close()
    // 等待所有行处理完成再返回
    wg.Wait()

    if streamErr != nil {
        var exitErr *remotecommand.ExitError
        if errors.As(streamErr, &exitErr) {
            return exitErr.ExitStatus(), nil
        }
        return 0, fmt.Errorf("could not stream results: %w", streamErr)
    }
    return 0, nil
}

你可以在调用时传入回调实现逐行打印：

exitCode, err := client.RunCommandStream(ctx, args, func(line string, isStderr bool) {
    if isStderr {
        fmt.Printf("[STDERR] %s\n", line)
    } else {
        fmt.Printf("[STDOUT] %s\n", line)
    }
})

• 如果需要同时收集全量输出，可以在回调里把行内容追加到全局的bytes.Buffer中，不需要单独再做收集逻辑
• 如果你不想用回调，也可以把行内容写入带缓冲的channel返回给调用方，注意要在所有内容处理完后关闭channel避免调用方阻塞
• v0.24.0及以上版本的client-go，可以替换exec.Stream为exec.StreamWithContext(ctx, opts)，实现上下文取消时自动终止远程命令执行

内容的提问来源于stack exchange，提问作者Paymahn Moghadasian