newBuffered starts the buffered pty. If the context ends the process will be killed.
(ctx context.Context, logger slog.Logger, execer agentexec.Execer, cmd *pty.Cmd, options *Options)
| 40 | // newBuffered starts the buffered pty. If the context ends the process will be |
| 41 | // killed. |
| 42 | func newBuffered(ctx context.Context, logger slog.Logger, execer agentexec.Execer, cmd *pty.Cmd, options *Options) *bufferedReconnectingPTY { |
| 43 | rpty := &bufferedReconnectingPTY{ |
| 44 | activeConns: map[string]net.Conn{}, |
| 45 | command: cmd, |
| 46 | metrics: options.Metrics, |
| 47 | state: newState(), |
| 48 | timeout: options.Timeout, |
| 49 | } |
| 50 | |
| 51 | // Default to buffer 64KiB. |
| 52 | circularBuffer, err := circbuf.NewBuffer(64 << 10) |
| 53 | if err != nil { |
| 54 | rpty.state.setState(StateDone, xerrors.Errorf("create circular buffer: %w", err)) |
| 55 | return rpty |
| 56 | } |
| 57 | rpty.circularBuffer = circularBuffer |
| 58 | |
| 59 | // Add terminal environment then start the command with a pty. pty.Cmd |
| 60 | // duplicates Path as the first argument so remove it. |
| 61 | cmdWithEnv := execer.PTYCommandContext(ctx, cmd.Path, cmd.Args[1:]...) |
| 62 | cmdWithEnv.Env = withTerminalEnv(rpty.command.Env) |
| 63 | cmdWithEnv.Dir = rpty.command.Dir |
| 64 | ptty, process, err := pty.Start(cmdWithEnv) |
| 65 | if err != nil { |
| 66 | rpty.state.setState(StateDone, xerrors.Errorf("start pty: %w", err)) |
| 67 | return rpty |
| 68 | } |
| 69 | rpty.ptty = ptty |
| 70 | rpty.process = process |
| 71 | |
| 72 | go rpty.lifecycle(ctx, logger) |
| 73 | |
| 74 | // Multiplex the output onto the circular buffer and each active connection. |
| 75 | // We do not need to separately monitor for the process exiting. When it |
| 76 | // exits, our ptty.OutputReader() will return EOF after reading all process |
| 77 | // output. |
| 78 | go func() { |
| 79 | buffer := make([]byte, 1024) |
| 80 | for { |
| 81 | read, err := ptty.OutputReader().Read(buffer) |
| 82 | if err != nil { |
| 83 | // When the PTY is closed, this is triggered. |
| 84 | // Error is typically a benign EOF, so only log for debugging. |
| 85 | if errors.Is(err, io.EOF) { |
| 86 | logger.Debug(ctx, "unable to read pty output, command might have exited", slog.Error(err)) |
| 87 | } else { |
| 88 | logger.Warn(ctx, "unable to read pty output, command might have exited", slog.Error(err)) |
| 89 | rpty.metrics.WithLabelValues("output_reader").Add(1) |
| 90 | } |
| 91 | // Could have been killed externally or failed to start at all (command |
| 92 | // not found for example). |
| 93 | // TODO: Should we check the process's exit code in case the command was |
| 94 | // invalid? |
| 95 | rpty.Close(nil) |
| 96 | break |
| 97 | } |
| 98 | part := buffer[:read] |
| 99 | rpty.state.cond.L.Lock() |
no test coverage detected