使用 golang 客户端从远程 kubernetes 命令流输出

如何解决使用 golang 客户端从远程 kubernetes 命令流输出

我正在使用 golang kubernetes 客户端来创建 kubernetes pod 并在其中执行远程命令。但是，我发现在远程执行完成之前我无法获得有关远程执行状态的反馈，因为我无法弄清楚如何流式传输远程命令的日志。这是我当前执行远程命令的实现：

func (k *KubernetesClient) RunCommand(ctx context.Context,args *RunCommandArgs) (string,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 "",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 "",fmt.Errorf("could not exec command: %w",err)
    }
    var stdout,stderr bytes.Buffer
    var streamErr error
    streamErr = exec.Stream(remotecommand.StreamOptions{
        Stdin:  nil,Stdout: &stdout,Stderr: &stderr,Tty:    false,})

    if streamErr != nil {
        if strings.Contains(streamErr.Error(),"command terminated with exit code") {
            return stderr.String(),1,nil
        } else {
            return "",fmt.Errorf("could not stream results: %w",streamErr)
        }
    }
    return stdout.String(),nil
}

在这个实现中，我不知道远程命令的状态，直到它完成执行，此时我会立即获得所有输出日志。

有没有办法在调用 stdout 写入 stderr/exec.Stream 时读取它们？在理想的世界中，我希望能够逐行打印远程命令行的输出。我注意到 bytes.Buffer 有一个接受分隔符的 ReadString 方法。这看起来是一个有用的方法，但我一直无法弄清楚如何使用它。

解决方法

这只是部分答案，但如果我设置使用以下 PodExecOptions 和 StreamOptions 那么我会看到每个日志行都被实时打印（注意 Tty 是 {{1 }} 并且我使用的是标准输入和标准输出，而不是自定义缓冲区）：

true

和

v1.PodExecOptions{
        Stdin:     true,Stdout:    true,Stderr:    false,TTY:       true,Container: args.ContainerId,Command:   []string{"sh","-c",args.Command},}

但是，如果我尝试使用除 remotecommand.StreamOptions{ Stdin: os.Stdin,Stdout: os.Stdout,Stderr: nil,Tty: true,} 和 os.Stdin 以外的其他内容，那么我永远不会得到任何日志行。例如，以下用法不打印任何内容：

os.Stdout

我仍在尝试弄清楚如何使用自定义缓冲区，以便我可以管理写入日志的内容，而不是直接管道到标准输出（我想将一些自定义字段附加到记录的每一行）。

---

编辑：好的，我想出了一个有效的解决方案。这是完整的代码

    var stdout,stdin bytes.Buffer
    var streamErr error

    go func() {
        streamErr = exec.Stream(remotecommand.StreamOptions{
            Stdin:  &stdin,Stdout: &stdout,})
    }()
    time.Sleep(5*time.Second)
    log.Info("doing raw string calls on both buffers")
    log.Info(stdin.String())
    log.Info(stdout.String())
    log.Info("starting scan of stdin")
    scanner := bufio.NewScanner(&stdin)
    scanner.Split(bufio.ScanLines)
    for scanner.Scan() {
        m := scanner.Text()
        fmt.Println(m)
    }
    log.Info("starting scan of stdout")
    scanner = bufio.NewScanner(&stdout)
    scanner.Split(bufio.ScanLines)
    for scanner.Scan() {
        m := scanner.Text()
        fmt.Println(m)
    }
    log.Info("finished scanning of stdout")

我创建了一个结构体，它实现了 type LogStreamer struct{ b bytes.Buffer } func (l *LogStreamer) String() string { return l.b.String() } func (l *LogStreamer) Write(p []byte) (n int,err error) { a := strings.TrimSpace(string(p)) l.b.WriteString(a) log.Info(a) return len(p),nil } func (k *KubernetesClient) RunCommand(ctx context.Context,args *RunCommandArgs) (string,int,error) { req := k.clientset.CoreV1().RESTClient().Post().Resource("pods").Name(args.ContainerId).Namespace(k.namespace).SubResource("exec").Param("container","worker") scheme := runtime.NewScheme() if err := v1.AddToScheme(scheme); err != nil { return "",fmt.Errorf("could not add to scheme: %w",err) } parameterCodec := runtime.NewParameterCodec(scheme) req.VersionedParams(&v1.PodExecOptions{ Stdin: true,},parameterCodec) exec,err := remotecommand.NewSPDYExecutor(k.config,"POST",req.URL()) if err != nil { return "",fmt.Errorf("could not exec command: %w",err) } var streamErr error l := &LogStreamer{} streamErr = exec.Stream(remotecommand.StreamOptions{ Stdin: os.Stdin,Stdout: l,}) if streamErr != nil { if strings.Contains(streamErr.Error(),"command terminated with exit code") { return l.String(),1,nil } else { return "",fmt.Errorf("could not stream results: %w",streamErr) } } return l.String(),nil } 接口并在 io.Writer 结构体中使用它。另请注意，我不得不在 StreamOptions 结构中使用 os.Stdin 否则只有一行会被流回 StreamOptions。

另请注意，我必须修剪传递给 Stdout 的缓冲区，因为回车或换行似乎会导致 logrus 包出现问题。这个解决方案还有更多的改进要添加，但它肯定朝着正确的方向前进。