send "EOF" event for stdin through exec session when stdin is closed

go.mod

@@ -60,7 +60,7 @@ require (
  github.com/mitchellh/cli v1.1.2
  github.com/mitchellh/copystructure v1.0.0
  github.com/mitchellh/go-glint v0.0.0-20201015034436-f80573c636de
- github.com/mitchellh/go-grpc-net-conn v0.0.0-20200407005438-c00174eff6c8
+ github.com/mitchellh/go-grpc-net-conn v0.0.0-20200427190222-eb030e4876f0
  github.com/mitchellh/go-homedir v1.1.0
  github.com/mitchellh/go-testing-interface v1.14.1
  github.com/mitchellh/go-wordwrap v1.0.1

go.sum

@@ -928,8 +928,8 @@ github.com/mitchellh/go-glint v0.0.0-20200930000256-df5e721f3258 h1:x7Z92EayV/P8
 github.com/mitchellh/go-glint v0.0.0-20200930000256-df5e721f3258/go.mod h1:NrJbv11op7A0gjSLtfvzm74YkVKRS24KxucwneYUX4M=
 github.com/mitchellh/go-glint v0.0.0-20201015034436-f80573c636de h1:zEtM2uLDYhUgehFO/lhsepZLz5TZpysDuwrezt+/w4k=
 github.com/mitchellh/go-glint v0.0.0-20201015034436-f80573c636de/go.mod h1:9X3rpO+I3yuihb6p8ktF8qWxROGwij9DBW/czUsMlhk=
-github.com/mitchellh/go-grpc-net-conn v0.0.0-20200407005438-c00174eff6c8 h1:vYarZ5R8DcSQ4MayciodsBAPc9pVGnFucqOfd+Rfaxw=
-github.com/mitchellh/go-grpc-net-conn v0.0.0-20200407005438-c00174eff6c8/go.mod h1:ZCzL0JMR6qfm7VrDC8HGwVtPA8D2Ijc/edUSBw58x94=
+github.com/mitchellh/go-grpc-net-conn v0.0.0-20200427190222-eb030e4876f0 h1:oZuel4h7224ILBLg2SlTxdaMYXDyqcVfL4Cg1PJQHZs=
+github.com/mitchellh/go-grpc-net-conn v0.0.0-20200427190222-eb030e4876f0/go.mod h1:ZCzL0JMR6qfm7VrDC8HGwVtPA8D2Ijc/edUSBw58x94=
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=

internal/ceb/exec.go

@@ -229,6 +229,10 @@ func (ceb *CEB) startExec(execConfig *pb.EntrypointConfig_Exec) {
  log.Trace("input received", "data", event.Input)
  io.Copy(stdinW, bytes.NewReader(event.Input))
 
+ case *pb.EntrypointExecResponse_InputEof:
+ log.Trace("input EOF, closing stdin")
+ stdinW.Close()
+
  case *pb.EntrypointExecResponse_Winch:
  log.Debug("window size change event, changing")
 

internal/server/execclient/client.go

@@ -5,11 +5,14 @@ import (
  "context"
  "fmt"
  "io"
+ "io/ioutil"
  "os"
  "os/signal"
+ "sync"
 
  "github.com/containerd/console"
  "github.com/golang/protobuf/proto"
+ "github.com/golang/protobuf/ptypes/empty"
  "github.com/hashicorp/go-hclog"
  grpc_net_conn "github.com/mitchellh/go-grpc-net-conn"
  sshterm "golang.org/x/crypto/ssh/terminal"
@@ -133,24 +136,71 @@ func (c *Client) Run() (int, error) {
  ctx, cancel := context.WithCancel(c.Context)
  defer cancel()
 
- input := &EscapeWatcher{Cancel: cancel, Input: c.Stdin}
+ // If we have interactive stdin WITHOUT a tty output, we treat stdin
+ // as if its closed. This allows commands such as "ls" to work without
+ // a terminal otherwise they'll hang open forever on stdin. Conversely,
+ // we allow interactive stdin with a TTY, and we also allow non-interactive
+ // stdin always since that'll end in an EOF at some point.
+ copyStdin := true
+ if stdinF, ok := c.Stdin.(*os.File); ok {
+ fi, err := stdinF.Stat()
+ if err != nil {
+ return 0, err
+ }
+
+ if fi.Mode()&os.ModeCharDevice != 0 && ptyF == nil {
+ // Stdin is from a terminal but we don't have a pty output.
+ // In this case, we treat stdin as if its closed.
+ c.Logger.Info("terminal stdin without a pty, not using input for command")
+ copyStdin = false
+ }
+ }
+
+ // We need to lock access to our stream writer since it is unsafe in
+ // gRPC to concurrently send data.
+ var streamLock sync.Mutex
 
  // Build our connection. We only build the stdin sending side because
  // we can receive other message types from our recv.
- go io.Copy(&grpc_net_conn.Conn{
- Stream: client,
- Request: &pb.ExecStreamRequest{},
- Encode: grpc_net_conn.SimpleEncoder(func(msg proto.Message) *[]byte {
- req := msg.(*pb.ExecStreamRequest)
- if req.Event == nil {
- req.Event = &pb.ExecStreamRequest_Input_{
- Input: &pb.ExecStreamRequest_Input{},
- }
- }
+ go func() {
+ input := &EscapeWatcher{Cancel: cancel, Input: c.Stdin}
+
+ // If we're copying stdin then start that copy process.
+ if copyStdin {
+ io.Copy(&grpc_net_conn.Conn{
+ Stream: client,
+ Request: &pb.ExecStreamRequest{},
+ ResponseLock: &streamLock,
+ Encode: grpc_net_conn.SimpleEncoder(func(msg proto.Message) *[]byte {
+ req := msg.(*pb.ExecStreamRequest)
+ if req.Event == nil {
+ req.Event = &pb.ExecStreamRequest_Input_{
+ Input: &pb.ExecStreamRequest_Input{},
+ }
+ }
+
+ return &req.Event.(*pb.ExecStreamRequest_Input_).Input.Data
+ }),
+ }, input)
+ } else {
+ // If we're NOT copying, we still start a copy to discard
+ // in the background so that we still handle escape sequences.
+ go io.Copy(ioutil.Discard, input)
+ }
 
- return &req.Event.(*pb.ExecStreamRequest_Input_).Input.Data
- }),
- }, input)
+ // After the copy ends, no matter what we send an EOF to the
+ // remote end because there will be no more input.
+ c.Logger.Debug("stdin closed, sending input EOF event")
+ streamLock.Lock()
+ defer streamLock.Unlock()
+ if err := client.Send(&pb.ExecStreamRequest{
+ Event: &pb.ExecStreamRequest_InputEof{
+ InputEof: &empty.Empty{},
+ },
+ }); err != nil {
+ c.Logger.Warn("error sending InputEOF event", "err", err)
+ }
+ }()
 
  // Add our recv blocker that sends data
  recvCh := make(chan *pb.ExecStreamResponse)
@@ -203,7 +253,8 @@ func (c *Client) Run() (int, error) {
  }
 
  // Send the new window size
- if err := client.Send(&pb.ExecStreamRequest{
+ streamLock.Lock()
+ err = client.Send(&pb.ExecStreamRequest{
  Event: &pb.ExecStreamRequest_Winch{
  Winch: &pb.ExecStreamRequest_WindowSize{
  Rows: int32(sz.Height),
@@ -212,7 +263,9 @@ func (c *Client) Run() (int, error) {
  Width: int32(sz.Width),
  },
  },
- }); err != nil {
+ })
+ streamLock.Unlock()
+ if err != nil {
  // Ignore this error
  continue
  }