[supervisor] Don't emit DeadlineExceeded if client closes connection (feature flag: supervisor_terminal_no_deadline_exceeded)

Author: geropl

components/supervisor/pkg/supervisor/supervisor.go

@@ -29,6 +29,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -336,9 +337,10 @@ func Run(options ...RunOption) {
 		}
 	}
 
+	terminalNoDeadlineExceeded := watchTerminalNoDeadlineExceeded(ctx, exps, host)
 	willShutdownCtx, fireWillShutdown := context.WithCancel(ctx)
 	termMux := terminal.NewMux()
-	termMuxSrv := terminal.NewMuxTerminalService(termMux)
+	termMuxSrv := terminal.NewMuxTerminalService(termMux, terminalNoDeadlineExceeded)
 	termMuxSrv.DefaultWorkdir = cfg.RepoRoot
 	if cfg.WorkspaceRoot != "" {
 		termMuxSrv.DefaultWorkdirProvider = func() string {
@@ -582,6 +584,36 @@ func getIDENotReadyShutdownDuration(ctx context.Context, exps experiments.Client
 	}
 }
 
+func watchTerminalNoDeadlineExceeded(ctx context.Context, exps experiments.Client, gitpodHost string) *atomic.Bool {
+	newBool := func(v bool) *atomic.Bool {
+		r := atomic.Bool{}
+		r.Store(v)
+		return &r
+	}
+	if exps == nil {
+		return newBool(false)
+	}
+
+	value := exps.GetBoolValue(ctx, "supervisor_terminal_no_deadline_exceeded", false, experiments.Attributes{GitpodHost: gitpodHost})
+	result := newBool(value)
+
+	go (func() {
+		t := time.NewTicker(30 * time.Second)
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				value := exps.GetBoolValue(ctx, "supervisor_terminal_no_deadline_exceeded", false, experiments.Attributes{GitpodHost: gitpodHost})
+				result.Store(value)
+			}
+		}
+	})()
+
+	return result
+}
+
 func isShallowRepository(rootDir string) bool {
 	cmd := runAsGitpodUser(exec.Command("git", "rev-parse", "--is-shallow-repository"))
 	cmd.Dir = rootDir

components/supervisor/pkg/supervisor/tasks_test.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -21,6 +22,12 @@ import (
 	"github.com/gitpod-io/gitpod/supervisor/pkg/terminal"
 )
 
+func newBool(b bool) *atomic.Bool {
+	result := atomic.Bool{}
+	result.Store(b)
+	return &result
+}
+
 var (
 	skipCommand = "echo \"skip\""
 	failCommand = "exit 1"
@@ -216,7 +223,7 @@ func TestTaskManager(t *testing.T) {
 			}
 
 			var (
-				terminalService = terminal.NewMuxTerminalService(terminal.NewMux())
+				terminalService = terminal.NewMuxTerminalService(terminal.NewMux(), newBool(true))
 				contentState    = NewInMemoryContentState("")
 				reporter        = testHeadlessTaskProgressReporter{}
 				taskManager     = newTasksManager(&Config{

components/supervisor/pkg/terminal/service.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -26,7 +27,7 @@ import (
 )
 
 // NewMuxTerminalService creates a new terminal service.
-func NewMuxTerminalService(m *Mux) *MuxTerminalService {
+func NewMuxTerminalService(m *Mux, terminalNoDeadlineExceeded *atomic.Bool) *MuxTerminalService {
 	shell := os.Getenv("SHELL")
 	if shell == "" {
 		shell = "/bin/bash"
@@ -36,6 +37,8 @@ func NewMuxTerminalService(m *Mux) *MuxTerminalService {
 		DefaultWorkdir: "/workspace",
 		DefaultShell:   shell,
 		Env:            os.Environ(),
+
+		terminalNoDeadlineExceeded: terminalNoDeadlineExceeded,
 	}
 }
 
@@ -53,6 +56,8 @@ type MuxTerminalService struct {
 	DefaultCreds       *syscall.Credential
 	DefaultAmbientCaps []uintptr
 
+	terminalNoDeadlineExceeded *atomic.Bool
+
 	api.UnimplementedTerminalServiceServer
 }
 
@@ -286,6 +291,9 @@ func (srv *MuxTerminalService) Listen(req *api.ListenTerminalRequest, resp api.T
 			err = resp.Send(message)
 		case err = <-errchan:
 		case <-resp.Context().Done():
+			if srv.terminalNoDeadlineExceeded.Load() {
+				return nil
+			}
 			return status.Error(codes.DeadlineExceeded, resp.Context().Err().Error())
 		}
 		if err == io.EOF {

components/supervisor/pkg/terminal/terminal_test.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -21,6 +22,12 @@ import (
 	"github.com/gitpod-io/gitpod/supervisor/api"
 )
 
+func newBool(b bool) *atomic.Bool {
+	result := atomic.Bool{}
+	result.Store(b)
+	return &result
+}
+
 func TestTitle(t *testing.T) {
 	t.Skip("skipping flakey tests")
 
@@ -59,7 +66,7 @@ func TestTitle(t *testing.T) {
 			}
 			defer os.RemoveAll(tmpWorkdir)
 
-			terminalService := NewMuxTerminalService(mux)
+			terminalService := NewMuxTerminalService(mux, newBool(true))
 			terminalService.DefaultWorkdir = tmpWorkdir
 
 			term, err := terminalService.OpenWithOptions(ctx, &api.OpenTerminalRequest{}, TermOptions{
@@ -197,7 +204,7 @@ func TestAnnotations(t *testing.T) {
 			mux := NewMux()
 			defer mux.Close(ctx)
 
-			terminalService := NewMuxTerminalService(mux)
+			terminalService := NewMuxTerminalService(mux, newBool(true))
 			var err error
 			if test.Opts == nil {
 				_, err = terminalService.Open(ctx, test.Req)
@@ -248,7 +255,7 @@ func TestTerminals(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.Desc, func(t *testing.T) {
-			terminalService := NewMuxTerminalService(NewMux())
+			terminalService := NewMuxTerminalService(NewMux(), newBool(true))
 			resp, err := terminalService.Open(context.Background(), &api.OpenTerminalRequest{})
 			if err != nil {
 				t.Fatal(err)
@@ -329,7 +336,7 @@ func TestWorkDirProvider(t *testing.T) {
 	mux := NewMux()
 	defer mux.Close(ctx)
 
-	terminalService := NewMuxTerminalService(mux)
+	terminalService := NewMuxTerminalService(mux, newBool(true))
 
 	type AssertWorkDirTest struct {
 		expectedWorkDir string