monitor ssh connection live-ness

communicator/ssh/communicator.go

@@ -27,17 +27,23 @@ import (
 const (
 	// DefaultShebang is added at the top of a SSH script file
 	DefaultShebang = "#!/bin/sh\n"
+)
+
+var (
+	// randShared is a global random generator object that is shared.  This must be
+	// shared since it is seeded by the current time and creating multiple can
+	// result in the same values. By using a shared RNG we assure different numbers
+	// per call.
+	randLock   sync.Mutex
+	randShared *rand.Rand
 
 	// enable ssh keeplive probes by default
 	keepAliveInterval = 2 * time.Second
-)
 
-// randShared is a global random generator object that is shared.
-// This must be shared since it is seeded by the current time and
-// creating multiple can result in the same values. By using a shared
-// RNG we assure different numbers per call.
-var randLock sync.Mutex
-var randShared *rand.Rand
+	// max time to wait for for a KeepAlive response before considering the
+	// connection to be dead.
+	maxKeepAliveDelay = 120 * time.Second
+)
 
 // Communicator represents the SSH communicator
 type Communicator struct {
@@ -225,20 +231,50 @@ func (c *Communicator) Connect(o terraform.UIOutput) (err error) {
 	// long-running commands.
 	log.Printf("[DEBUG] starting ssh KeepAlives")
 	go func() {
-		t := time.NewTicker(keepAliveInterval)
-		defer t.Stop()
+		defer cancelKeepAlive()
+		// Along with the KeepAlives generating packets to keep the tcp
+		// connection open, we will use the replies to verify liveness of the
+		// connection. This will prevent dead connections from blocking the
+		// provisioner indefinitely.
+		respCh := make(chan error, 1)
+
+		go func() {
+			t := time.NewTicker(keepAliveInterval)
+			defer t.Stop()
+			for {
+				select {
+				case <-t.C:
+					_, _, err := c.client.SendRequest("keepalive@terraform.io", true, nil)
+					respCh <- err
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+
+		after := time.NewTimer(maxKeepAliveDelay)
+		defer after.Stop()
+
 		for {
 			select {
-			case <-t.C:
-				// there's no useful response to these, just abort when there's
-				// an error.
-				_, _, err := c.client.SendRequest("keepalive@terraform.io", true, nil)
+			case err := <-respCh:
 				if err != nil {
+					log.Printf("[ERROR] ssh keepalive: %s", err)
+					sshConn.Close()
 					return
 				}
+			case <-after.C:
+				// abort after too many missed keepalives
+				log.Println("[ERROR] no reply from ssh server")
+				sshConn.Close()
+				return
 			case <-ctx.Done():
 				return
 			}
+			if !after.Stop() {
+				<-after.C
+			}
+			after.Reset(maxKeepAliveDelay)
 		}
 	}()
 

communicator/ssh/communicator_test.go

@@ -100,16 +100,19 @@ func newMockLineServer(t *testing.T, signer ssh.Signer, pubKey string) string {
 
 			go func(in <-chan *ssh.Request) {
 				for req := range in {
+					// since this channel's requests are serviced serially,
+					// this will block keepalive probes, and can simulate a
+					// hung connection.
+					if bytes.Contains(req.Payload, []byte("sleep")) {
+						time.Sleep(time.Second)
+					}
+
 					if req.WantReply {
 						req.Reply(true, nil)
 					}
 				}
 			}(requests)
 
-			go func(newChannel ssh.NewChannel) {
-				conn.OpenChannel(newChannel.ChannelType(), nil)
-			}(newChannel)
-
 			defer channel.Close()
 		}
 		conn.Close()
@@ -182,6 +185,10 @@ func TestStart(t *testing.T) {
 // TestKeepAlives verifies that the keepalive messages don't interfere with
 // normal operation of the client.
 func TestKeepAlives(t *testing.T) {
+	ivl := keepAliveInterval
+	keepAliveInterval = 250 * time.Millisecond
+	defer func() { keepAliveInterval = ivl }()
+
 	address := newMockLineServer(t, nil, testClientPublicKey)
 	parts := strings.Split(address, ":")
 
@@ -193,7 +200,6 @@ func TestKeepAlives(t *testing.T) {
 				"password": "pass",
 				"host":     parts[0],
 				"port":     parts[1],
-				"timeout":  "30s",
 			},
 		},
 	}
@@ -209,18 +215,64 @@ func TestKeepAlives(t *testing.T) {
 
 	var cmd remote.Cmd
 	stdout := new(bytes.Buffer)
-	cmd.Command = "echo foo"
+	cmd.Command = "sleep"
 	cmd.Stdout = stdout
 
 	// wait a bit before executing the command, so that at least 1 keepalive is sent
-	time.Sleep(3 * time.Second)
+	time.Sleep(500 * time.Millisecond)
 
 	err = c.Start(&cmd)
 	if err != nil {
 		t.Fatalf("error executing remote command: %s", err)
 	}
 }
 
+// TestDeadConnection verifies that failed keepalive messages will eventually
+// kill the connection.
+func TestFailedKeepAlives(t *testing.T) {
+	ivl := keepAliveInterval
+	del := maxKeepAliveDelay
+	maxKeepAliveDelay = 500 * time.Millisecond
+	keepAliveInterval = 250 * time.Millisecond
+	defer func() {
+		keepAliveInterval = ivl
+		maxKeepAliveDelay = del
+	}()
+
+	address := newMockLineServer(t, nil, testClientPublicKey)
+	parts := strings.Split(address, ":")
+
+	r := &terraform.InstanceState{
+		Ephemeral: terraform.EphemeralState{
+			ConnInfo: map[string]string{
+				"type":     "ssh",
+				"user":     "user",
+				"password": "pass",
+				"host":     parts[0],
+				"port":     parts[1],
+			},
+		},
+	}
+
+	c, err := New(r)
+	if err != nil {
+		t.Fatalf("error creating communicator: %s", err)
+	}
+
+	if err := c.Connect(nil); err != nil {
+		t.Fatal(err)
+	}
+	var cmd remote.Cmd
+	stdout := new(bytes.Buffer)
+	cmd.Command = "sleep"
+	cmd.Stdout = stdout
+
+	err = c.Start(&cmd)
+	if err == nil {
+		t.Fatal("expected connection error")
+	}
+}
+
 func TestLostConnection(t *testing.T) {
 	address := newMockLineServer(t, nil, testClientPublicKey)
 	parts := strings.Split(address, ":")

go.mod

@@ -120,7 +120,7 @@ require (
 	go.uber.org/atomic v1.3.2 // indirect
 	go.uber.org/multierr v1.1.0 // indirect
 	go.uber.org/zap v1.9.1 // indirect
-	golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734
+	golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4
 	golang.org/x/net v0.0.0-20190502183928-7f726cade0ab
 	golang.org/x/oauth2 v0.0.0-20190220154721-9b3c75971fc9
 	google.golang.org/api v0.1.0

go.sum

@@ -449,6 +449,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90Pveol
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734 h1:p/H982KKEjUnLJkM3tt/LemDnOc1GiZL5FCVlORJ5zo=
 golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc=
+golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=