[supervisor] merge run tasks

Author: akosyakov

components/supervisor/.gitignore

@@ -0,0 +1 @@
+supervisor

components/supervisor/pkg/ports/ports.go

@@ -872,3 +872,13 @@ func defaultRoutableIP() string {
 
 	return addresses[0].(*net.IPNet).IP.String()
 }
+
+func (pm *Manager) IsServed(port uint32) bool {
+	served := pm.served
+	for _, served := range served {
+		if served.Port == port {
+			return true
+		}
+	}
+	return false
+}

components/supervisor/pkg/supervisor/config.go

@@ -85,6 +85,9 @@ type StaticConfig struct {
 
 	// SSHPort is the port we run the SSH server on
 	SSHPort int `json:"sshPort"`
+
+	// RunEndpointPort is the port where to serve the run API endpoint on
+	RunEndpointPort *int `json:"runEndpointPort,omitempty"`
 }
 
 // Validate validates this configuration.

components/supervisor/pkg/supervisor/services.go

@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
@@ -244,10 +245,44 @@ func (s *statusService) TasksStatus(req *api.TasksStatusRequest, srv api.StatusS
 	case <-s.Tasks.ready:
 	}
 
+	var runStream api.StatusService_TasksStatusClient
+	if !s.Tasks.runGP && s.Tasks.config.RunEndpointPort != nil {
+		runPort := *s.Tasks.config.RunEndpointPort
+		if s.Ports.IsServed(uint32(runPort)) {
+			url := fmt.Sprintf("localhost:%d", *s.Tasks.config.RunEndpointPort)
+			conn, err := grpc.DialContext(srv.Context(), url, grpc.WithTransportCredentials(insecure.NewCredentials()))
+			if err != nil {
+				log.WithError(err).Debug("cannot connect to run supervisor")
+			} else {
+				defer conn.Close()
+				runStream, err = api.NewStatusServiceClient(conn).TasksStatus(srv.Context(), req)
+				if err != nil {
+					log.WithError(err).Debug("failed to stream run tasks state")
+				}
+			}
+		}
+	}
+
+	var runTasks []*api.TaskStatus
+	var subTasks []*api.TaskStatus
+	doSend := func() error {
+		var tasks []*api.TaskStatus
+		tasks = append(tasks, runTasks...)
+		tasks = append(tasks, subTasks...)
+		return srv.Send(&api.TasksStatusResponse{Tasks: tasks})
+	}
+
 	if !req.Observe {
-		return srv.Send(&api.TasksStatusResponse{
-			Tasks: s.Tasks.Status(),
-		})
+		if runStream != nil {
+			resp, err := runStream.Recv()
+			if err != nil {
+				log.WithError(err).Error("failed to receive run tasks state")
+			} else {
+				runTasks = resp.Tasks
+			}
+		}
+		subTasks = s.Tasks.Status()
+		return doSend()
 	}
 
 	sub := s.Tasks.Subscribe()
@@ -256,15 +291,37 @@ func (s *statusService) TasksStatus(req *api.TasksStatusRequest, srv api.StatusS
 	}
 	defer sub.Close()
 
+	respChan := make(chan []*api.TaskStatus, 5)
+	if runStream != nil {
+		go func() {
+			for {
+				resp, err := runStream.Recv()
+				if err != nil {
+					return
+				}
+				respChan <- resp.GetTasks()
+			}
+		}()
+	}
 	for {
 		select {
 		case <-srv.Context().Done():
 			return nil
+		case update := <-respChan:
+			if update == nil {
+				return nil
+			}
+			runTasks = update
+			err := doSend()
+			if err != nil {
+				return err
+			}
 		case update := <-sub.Updates():
 			if update == nil {
 				return nil
 			}
-			err := srv.Send(&api.TasksStatusResponse{Tasks: update})
+			subTasks = update
+			err := doSend()
 			if err != nil {
 				return err
 			}
@@ -936,5 +993,8 @@ func (s *portService) RetryAutoExpose(ctx context.Context, req *api.RetryAutoExp
 
 // ResourcesStatus provides workspace resources status information.
 func (s *statusService) ResourcesStatus(ctx context.Context, in *api.ResourcesStatuRequest) (*api.ResourcesStatusResponse, error) {
+	if s.topService == nil {
+		return nil, status.Errorf(codes.Unavailable, "top service is disabled in run mode")
+	}
 	return s.topService.data, nil
 }

components/supervisor/pkg/supervisor/supervisor.go

@@ -228,6 +228,9 @@ func Run(options ...RunOption) {
 	if cfg.DesktopIDE != nil {
 		internalPorts = append(internalPorts, desktopIDEPort)
 	}
+	if cfg.RunEndpointPort != nil {
+		internalPorts = append(internalPorts, uint32(*cfg.RunEndpointPort))
+	}
 
 	endpoint, host, err := cfg.GitpodAPIEndpoint()
 	if err != nil {
@@ -269,14 +272,15 @@ func Run(options ...RunOption) {
 		internalPorts...,
 	)
 
-	topService := NewTopService()
-	topService.Observe(ctx)
-
+	var topService *TopService
 	supervisorMetrics := metrics.NewMetrics()
 	var metricsReporter *metrics.GrpcMetricsReporter
 	if opts.RunGP {
 		cstate.MarkContentReady(csapi.WorkspaceInitFromOther)
 	} else {
+		topService := NewTopService()
+		topService.Observe(ctx)
+
 		if !cfg.isHeadless() {
 			go startAnalyze(ctx, cfg, gitpodConfigService, topService, gitpodService)
 		}
@@ -315,7 +319,7 @@ func Run(options ...RunOption) {
 		Gid: gitpodGID,
 	}
 
-	taskManager := newTasksManager(cfg, termMuxSrv, cstate, nil, ideReady, desktopIdeReady)
+	taskManager := newTasksManager(cfg, termMuxSrv, cstate, nil, ideReady, desktopIdeReady, opts.RunGP)
 
 	apiServices := []RegisterableService{
 		&statusService{

components/supervisor/pkg/supervisor/tasks.go

@@ -109,9 +109,10 @@ type tasksManager struct {
 	reporter        headlessTaskProgressReporter
 	ideReady        *ideReadyState
 	desktopIdeReady *ideReadyState
+	runGP           bool
 }
 
-func newTasksManager(config *Config, terminalService *terminal.MuxTerminalService, contentState ContentState, reporter headlessTaskProgressReporter, ideReady *ideReadyState, desktopIdeReady *ideReadyState) *tasksManager {
+func newTasksManager(config *Config, terminalService *terminal.MuxTerminalService, contentState ContentState, reporter headlessTaskProgressReporter, ideReady *ideReadyState, desktopIdeReady *ideReadyState, runGP bool) *tasksManager {
 	return &tasksManager{
 		config:          config,
 		terminalService: terminalService,
@@ -122,6 +123,7 @@ func newTasksManager(config *Config, terminalService *terminal.MuxTerminalServic
 		storeLocation:   logs.TerminalStoreLocation,
 		ideReady:        ideReady,
 		desktopIdeReady: desktopIdeReady,
+		runGP:           runGP,
 	}
 }
 
@@ -203,11 +205,19 @@ func (tm *tasksManager) init(ctx context.Context) {
 	for i, config := range *tasks {
 		id := strconv.Itoa(i)
 		presentation := &api.TaskPresentation{}
+
+		var name string
 		if config.Name != nil {
-			presentation.Name = *config.Name
-		} else {
-			presentation.Name = "Gitpod Task " + strconv.Itoa(i+1)
+			name = *config.Name
+		}
+		if name == "" {
+			name = "Gitpod Task " + strconv.Itoa(i+1)
 		}
+		if tm.runGP {
+			name = "Run: " + name
+		}
+		presentation.Name = name
+
 		if config.OpenIn != nil {
 			presentation.OpenIn = *config.OpenIn
 		}

components/supervisor/pkg/supervisor/tasks_test.go

@@ -224,7 +224,7 @@ func TestTaskManager(t *testing.T) {
 						GitpodTasks:    gitpodTasks,
 						GitpodHeadless: strconv.FormatBool(test.Headless),
 					},
-				}, terminalService, contentState, &reporter, nil, nil)
+				}, terminalService, contentState, &reporter, nil, nil, false)
 			)
 			taskManager.storeLocation = storeLocation
 			contentState.MarkContentReady(test.Source)

components/supervisor/supervisor-config.json

@@ -3,5 +3,6 @@
   "desktopIdeConfigLocation": "/ide-desktop/supervisor-ide-config.json",
   "frontendLocation": "/.supervisor/frontend/",
   "apiEndpointPort": 22999,
-  "sshPort": 23001
-}
+  "sshPort": 23001,
+  "runEndpointPort": 25000
+}