fix: e2e test

Author: rosstimothy

integration/kube_integration_test.go

@@ -34,7 +34,6 @@ import (
 	"os/user"
 	"strconv"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -2217,13 +2216,17 @@ func testKubeJoin(t *testing.T, suite *KubeSuite) {
 	require.NoError(t, err)
 	participantStdoutR, participantStdoutW, err := os.Pipe()
 	require.NoError(t, err)
-	streamsMu := &sync.Mutex{}
-	streams := make([]*client.KubeSession, 0, 3)
-	observerCaptures := make([]*bytes.Buffer, 0, 2)
+
+	observerCaptures := make([]*bytes.Buffer, 2)
 	albProxy := helpers.MustStartMockALBProxy(t, teleport.Config.Proxy.WebAddr.Addr)
 
 	// join peer by KubeProxyAddr
 	group.Go(func() error {
+		defer func() {
+			// close participant stdout so that we can read it after till EOF
+			participantStdoutW.Close()
+		}()
+
 		tc, err := teleport.NewClient(helpers.ClientConfig{
 			Login:   hostUsername,
 			Cluster: helpers.Site,
@@ -2245,43 +2248,52 @@ func testKubeJoin(t *testing.T, suite *KubeSuite) {
 		if err != nil {
 			return trace.Wrap(err)
 		}
-		streamsMu.Lock()
-		streams = append(streams, stream)
-		streamsMu.Unlock()
+
 		stream.Wait()
-		// close participant stdout so that we can read it after till EOF
-		participantStdoutW.Close()
+
+		if err := stream.Detach(); err != nil && !terminal.IsOKWebsocketCloseError(err) {
+			return trace.Wrap(err)
+		}
+
 		return nil
 	})
 
 	// join observer by WebProxyAddr
 	group.Go(func() error {
 		stream, capture := kubeJoinByWebAddr(t, teleport, participantUsername, kubeUsers, kubeGroups)
-		streamsMu.Lock()
-		streams = append(streams, stream)
-		observerCaptures = append(observerCaptures, capture)
-		streamsMu.Unlock()
+		observerCaptures[0] = capture
 		stream.Wait()
+
+		if err := stream.Detach(); err != nil && !terminal.IsOKWebsocketCloseError(err) {
+			return trace.Wrap(err)
+		}
+
 		return nil
 	})
 
 	// join observer with ALPN conn upgrade
 	group.Go(func() error {
 		stream, capture := kubeJoinByALBAddr(t, teleport, participantUsername, kubeUsers, kubeGroups, albProxy.Addr().String())
-		streamsMu.Lock()
-		streams = append(streams, stream)
-		observerCaptures = append(observerCaptures, capture)
-		streamsMu.Unlock()
+		observerCaptures[1] = capture
 		stream.Wait()
+
+		if err := stream.Detach(); err != nil && !terminal.IsOKWebsocketCloseError(err) {
+			return trace.Wrap(err)
+		}
 		return nil
 	})
 
 	// We wait again for the second user to finish joining the session.
-	// We allow a bit of time to pass here to give the session manager time to recognize the
-	// new IO streams of the second client.
-	time.Sleep(time.Second * 5)
+	require.EventuallyWithT(t, func(t *assert.CollectT) {
+		session, err := teleport.Process.GetAuthServer().GetSessionTracker(context.Background(), session.GetName())
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		assert.Len(t, session.GetParticipants(), 2)
+	}, 15*time.Second, 500*time.Millisecond)
 
-	// sent a test message from the participant
+	// send a test message from the participant
 	participantStdinW.Write([]byte("\ahi from peer\n\r"))
 
 	// lets type "echo hi" followed by "enter" and then "exit" + "enter":