[supervisor] make it compatible with run-gp

Author: akosyakov

components/supervisor/.gitignore

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

components/supervisor/cmd/run.go

@@ -23,7 +23,7 @@ var runCmd = &cobra.Command{
 	Short: "starts the supervisor",
 
 	Run: func(cmd *cobra.Command, args []string) {
-		log.Init(ServiceName, Version, true, os.Getenv("SUPERVISOR_DEBUG_ENABLE") == "true")
+		log.Init(ServiceName, Version, !runOpts.RunGP, os.Getenv("SUPERVISOR_DEBUG_ENABLE") == "true")
 		common_grpc.SetupLogging()
 		supervisor.Version = Version
 		supervisor.Run(supervisor.WithRunGP(runOpts.RunGP))

components/supervisor/hot-swap.sh

@@ -0,0 +1,42 @@
+#!/bin/bash
+# Copyright (c) 2022 Gitpod GmbH. All rights reserved.
+# Licensed under the GNU Affero General Public License (AGPL).
+# See License.AGPL.txt in the project root for license information.
+
+set -Eeuo pipefail
+
+# This script swaps the backend startup endpoint with a built one
+# in a workspace and restarts the JB backend.
+
+component=${PWD##*/}
+workspaceUrl=$(echo "${1}" |sed -e "s/\/$//")
+echo "URL: $workspaceUrl"
+
+workspaceDesc=$(gpctl workspaces describe "$workspaceUrl" -o=json)
+
+podName=$(echo "$workspaceDesc" | jq .runtime.pod_name -r)
+echo "Pod: $podName"
+
+workspaceId=$(echo "$workspaceDesc" | jq .metadata.meta_id -r)
+echo "ID: $workspaceId"
+
+clusterHost=$(kubectl exec -it "$podName" -- printenv GITPOD_WORKSPACE_CLUSTER_HOST |sed -e "s/\s//g")
+echo "Cluster Host: $clusterHost"
+
+# prepare ssh
+ownerToken=$(kubectl get pod "$podName" -o=json | jq ".metadata.annotations.\"gitpod\/ownerToken\"" -r)
+sshConfig="./ssh-config"
+echo "Host $workspaceId" > "$sshConfig"
+echo "    Hostname \"$workspaceId.ssh.$clusterHost\"" >> "$sshConfig"
+echo "    User \"$workspaceId#$ownerToken\"" >> "$sshConfig"
+
+# build
+go build .
+echo "$component built"
+
+# upload
+uploadDest="/.supervisor/$component"
+echo "Upload Dest: $uploadDest"
+ssh -F "$sshConfig" "$workspaceId" "sudo chmod 777 $uploadDest && sudo rm $uploadDest"
+scp -F "$sshConfig" -r "./supervisor" "$workspaceId":"$uploadDest"
+echo "Swap complete"

components/supervisor/local-hot-swap.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+# Copyright (c) 2022 Gitpod GmbH. All rights reserved.
+# Licensed under the GNU Affero General Public License (AGPL).
+# See License.AGPL.txt in the project root for license information.
+
+set -Eeuo pipefail
+
+# This script swaps the backend startup endpoint with a built one
+# in a workspace and restarts the JB backend.
+
+component=${PWD##*/}
+
+# build
+go build .
+echo "$component built"
+
+sudo rm /.supervisor/supervisor
+sudo mv ./"$component" /.supervisor
+echo "Local Swap complete"

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/run/run.go

@@ -0,0 +1,58 @@
+// Copyright (c) 2022 Gitpod GmbH. All rights reserved.
+// Licensed under the GNU Affero General Public License (AGPL).
+// See License.AGPL.txt in the project root for license information.
+
+package run
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/gitpod-io/gitpod/supervisor/api"
+	"github.com/gitpod-io/gitpod/supervisor/pkg/ports"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type Option struct {
+	Port  uint32
+	Ports *ports.Manager
+}
+
+type Client struct {
+	option *Option
+
+	conn      *grpc.ClientConn
+	closeOnce sync.Once
+
+	Status   api.StatusServiceClient
+	Terminal api.TerminalServiceClient
+}
+
+func New(ctx context.Context, option *Option) (*Client, error) {
+	url := fmt.Sprintf("localhost:%d", option.Port)
+	conn, err := grpc.DialContext(ctx, url, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, err
+	}
+	return &Client{
+		option:   option,
+		conn:     conn,
+		Status:   api.NewStatusServiceClient(conn),
+		Terminal: api.NewTerminalServiceClient(conn),
+	}, nil
+}
+
+func (client *Client) Close() {
+	client.closeOnce.Do(func() {
+		client.conn.Close()
+	})
+}
+
+func (client *Client) Available() bool {
+	if client == nil {
+		return false
+	}
+	return client.option.Ports.IsServed(client.option.Port)
+}

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

@@ -26,6 +26,7 @@ import (
 	csapi "github.com/gitpod-io/gitpod/content-service/api"
 	"github.com/gitpod-io/gitpod/supervisor/api"
 	"github.com/gitpod-io/gitpod/supervisor/pkg/ports"
+	"github.com/gitpod-io/gitpod/supervisor/pkg/run"
 )
 
 // RegisterableService can register a service.
@@ -98,6 +99,7 @@ type statusService struct {
 	ideReady        *ideReadyState
 	desktopIdeReady *ideReadyState
 	topService      *TopService
+	runClient       *run.Client
 
 	api.UnimplementedStatusServiceServer
 }
@@ -244,10 +246,37 @@ func (s *statusService) TasksStatus(req *api.TasksStatusRequest, srv api.StatusS
 	case <-s.Tasks.ready:
 	}
 
+	var (
+		runStream api.StatusService_TasksStatusClient
+		err       error
+	)
+	if s.runClient.Available() {
+		runStream, err = s.runClient.Status.TasksStatus(srv.Context(), req)
+		if err != nil {
+			log.WithError(err).Error("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 +285,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 +987,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
 }