How to read stdout and stderr from a process launched via cmd.Command?

Question

go

Hello, I need to read stdout and stderr from processes started by Go itself via cmd.Command. As an example, I run regular go programs that write to stdout at 2 second intervals:

The code is long, but it's simple (process.go):

spoiler

package main

import (
  "flag"
  "fmt"
  "os"
  "os/signal"
  "syscall"
  "time"
)

func main() {
  processId := getProcessIdent()

  welcomeMsg(processId)

  toStdout := makeStdoutWriter(processId)
  sig := makeNotifier()
loop:
  for {
    select {
    case <-sig:
      break loop
    case <-time.After(time.Second * 2):
      toStdout()
    }
  }

  fmt.Println("Good Luck")
}

func welcomeMsg(processId int) {
  fmt.Println(fmt.Sprintf("Welcome to Process %d", processId))
}

func getProcessIdent() int {
  var processId int
  flag.IntVar(&processId, "id", 999, "number of lines to read from the file")
  flag.Parse()

  return processId
}

func makeNotifier() <-chan os.Signal {
  sig := make(chan os.Signal)
  signal.Notify(sig, os.Interrupt, syscall.SIGTERM)

  return sig
}

func makeStdoutWriter(processId int) func() {
  write := func(processId int) {
    fmt.Fprintln(os.Stdout, fmt.Sprintf("Process work - %d", processId))
  }

  return func() {
    write(processId)
    write(processId)
    write(processId)
  }
}

Next, I start the child process and install io.Pipe for outputs: (part of main.go)

type Executor struct {
  cmd *exec.Cmd
  r   *io.PipeReader
  w   *io.PipeWriter
}

func executeAnotherGo(id int) (*Executor, error) {
  args := []string{
    "run", "./process.go", "-id", strconv.Itoa(id),
  }

  r, w := io.Pipe()

  cmd := exec.Command("go", args...)
  cmd.Stdout = w
  cmd.Stderr = w

  if err := cmd.Start(); err != nil {
    return nil, err
  }

  return &Executor{cmd: cmd, r: r, w: w}, nil
}

Well, the main method that listens to the stdout of the process and writes to the appropriate file (for example) (part of main.go)

func perform(ctx context.Context, id int) {
  executor, err := executeAnotherGo(id)

  if err != nil {
    log.Fatal("Error execute process ", id, "cause: ", err)
  }

  writer, err := os.OpenFile(fmt.Sprintf("./process_%d.log", id), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)

  if err != nil {
    fmt.Println(err)
    return
  }

  localContext, cancel := context.WithCancel(ctx)

  defer func() {
    // close pipe writer
    if err := executor.w.Close(); err != nil {
      fmt.Println(err)
    }

    // try kill process
    if err := executor.cmd.Process.Kill(); err != nil {
      fmt.Println(id, executor.cmd.Process.Pid, err)
    }

    cancel()
  }()

  buf := make([]byte, 1024)
  ffmpegOutput := make(chan []string)
  processKilled := make(chan error, 1)

  // Runs a separate sub-thread, because when running in a single thread,
  // there is a lock while waiting for the buffer to be read.
  // In turn blocking by the reader will not allow the background task to finish gracefully
  go func() {
    for {
      count, err := executor.r.Read(buf)
      if err != nil {
        fmt.Println("Close reader, cause: ", err)
        return
      }

      buf = buf[:count]
      str := string(buf)

      parts := strings.Split(strings.TrimSpace(str), "\n")

      if len(parts) <= 1 {
        continue
      }

      ffmpegOutput <- parts

      fmt.Println(str)
    }
  }()

  // We listen to the process termination signal,
  // this will provide an opportunity to remove the task from the pool and restart it if necessary
  //
  // Note: We listen to the context so as not to leave active goroutines when the task is completed
  go func() {
    select {
    case processKilled <- executor.cmd.Wait():
      return
    case <-localContext.Done():
      return
    }
  }()

loop:
  for {
    select {
    case <-localContext.Done():
      fmt.Println("Cancel process: ", id)
      break loop
    case err := <-processKilled:
      fmt.Println("Killed", id, executor.cmd.Process.Pid, err)
      break loop
    case outPartials := <-ffmpegOutput:
      if _, err := writer.WriteString(strings.Join(outPartials, "\n")); err != nil {
        fmt.Println(err)
      }
    }
  }
}

I call three child processes:

func main() {
  fmt.Println("Run program, wait processes")

  sig := make(chan os.Signal)
  signal.Notify(sig, os.Interrupt, syscall.SIGTERM)

  ctx, cancel := context.WithCancel(context.Background())

  go perform(ctx, 1)
  go perform(ctx, 2)
  go perform(ctx, 3)

  <-sig

  // cancel all sub process
  cancel()

  // wait all canceled
  <-time.After(time.Second * 2)

  fmt.Println("graceful exit")
}

The original process writes to stdout (if run separately by hand, it will be:

Welcome to Process 3
Process work - 3
Process work - 3
Process work - 3
Process work - 3
Process work - 3
Process work - 3

BUT if run via cmd.Command () and listening through io.Pipe is not a mess at all (the output is in the spoiler):

spoiler

Process work - 1
Proc
ess work - 1
Process
Process work - 3
Proc
ess work - 3
Process
Process work - 2
Proc
ess work - 2
Process
ork - 2
P
rk - 2
Pr
ork - 3
P
rk - 3
Pr
ork - 1
P
rk - 1
Pr
- 1
P
- 1
Pr
- 2
P
- 2
Pr
- 3
P
- 3
Pr
3
P
2
P
1
P
1
P
3
P
2
P

At some point, it may stop outputting something at all, and sometimes it may “fall asleep” at all, that stdout does not close even after the original (its) process ends. if the process is completed, you need to "destroy" the goroutine and all its descendants.

If you use the usual os.Stdout instead of a pipe, everything is displayed correctly, but it is clear that it is already in the standard stdout

r, w := io.Pipe()

cmd := exec.Command("go", args...)

// заменяем
// cmd.Stdout = w
// cmd.Stderr = w

// на
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr

Please help me, I've already broken my brain, what's wrong?

UPD: if you read through the buffer

bufioReader := bufio.NewReader(executor.r)

for {
  line, _, err := bufioReader.ReadLine()

then the conclusion is correct, and I realized that I messed up

strings.Join(outPartials, "\n") - убрать новую строку

now the output is:

Welcome to Process 2Process work - 2Process work - 2Processwork - 2Process work - 2Process work - 2Process work - 2Process work- 2Processwork- 2Processwork -2

i.e. The message is coming in..