[loadgen] Resubscribe and log ids (#17080)

Author: WVerlaek

dev/loadgen/configs/prod-benchmark.yaml

@@ -78,17 +78,29 @@ repos:
     score: 20
     workspaceImage: registry.hub.docker.com/gitpod/workspace-full:latest
     workspaceClass: "g1-large"
+    environment:
+    - name: "GITPOD_TASKS"
+      value: "[{\"name\":\"Open port 1\",\"command\":\"gp ports visibility 6879:private\"},{\"name\":\"Open port 2\",\"command\":\"gp ports visibility 7869:private\"}]"
   - cloneURL: https://github.com/gitpod-io/template-typescript-react
     cloneTarget: main
     score: 20
     workspaceImage: registry.hub.docker.com/gitpod/workspace-full:latest
     workspaceClass: "default"
+    environment:
+    - name: "GITPOD_TASKS"
+      value: "[{\"name\":\"Open port 1\",\"command\":\"gp ports visibility 6879:private\"},{\"name\":\"Open port 2\",\"command\":\"gp ports visibility 7869:private\"}]"
   - cloneURL: https://github.com/gitpod-io/template-python-django
     cloneTarget: main
     score: 20
     workspaceImage: registry.hub.docker.com/gitpod/workspace-full:latest
     workspaceClass: "gitpodio-internal-xl"
+    environment:
+    - name: "GITPOD_TASKS"
+      value: "[{\"name\":\"Open port 1\",\"command\":\"gp ports visibility 6879:private\"},{\"name\":\"Open port 2\",\"command\":\"gp ports visibility 7869:private\"}]"
   - cloneURL: https://github.com/gitpod-io/non-gitpodified-repo
     score: 20
     cloneTarget: main
     workspaceImage: registry.hub.docker.com/gitpod/workspace-full:latest
+    environment:
+    - name: "GITPOD_TASKS"
+      value: "[{\"name\":\"Open port 1\",\"command\":\"gp ports visibility 6879:private\"},{\"name\":\"Open port 2\",\"command\":\"gp ports visibility 7869:private\"}]"

dev/loadgen/pkg/loadgen/executor.go

@@ -8,7 +8,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"math/rand"
 	"os"
 	"sync"
@@ -158,36 +157,45 @@ func (w *WsmanExecutor) Observe() (<-chan WorkspaceUpdate, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	w.Sub = append(w.Sub, cancel)
 
-	sub, err := w.C.Subscribe(ctx, &api.SubscribeRequest{
-		MustMatch: w.loadgenSessionFilter(),
-	})
-	if err != nil {
-		return nil, err
-	}
 	go func() {
 		defer close(res)
 		for {
-			resp, err := sub.Recv()
+			sub, err := w.C.Subscribe(ctx, &api.SubscribeRequest{
+				MustMatch: w.loadgenSessionFilter(),
+			})
 			if err != nil {
-				if err != io.EOF && status.Code(err) != codes.Canceled {
-					log.WithError(err).Warn("subscription failure")
-				}
-				return
-			}
-			status := resp.GetStatus()
-			if status == nil {
+				log.WithError(err).Warn("failed to subscribe to ws-manager, retrying...")
+				time.Sleep(5 * time.Second)
 				continue
 			}
 
-			res <- WorkspaceUpdate{
-				InstanceID:  status.Id,
-				WorkspaceID: status.Metadata.MetaId,
-				OwnerID:     status.Metadata.Owner,
-				Failed:      status.Conditions.Failed != "",
-				Phase:       status.Phase,
+			for {
+				resp, err := sub.Recv()
+				if err != nil {
+					if status.Code(err) != codes.Canceled {
+						log.WithError(err).Warn("lost connection to ws-manager, retrying...")
+						time.Sleep(5 * time.Second)
+						// Break and resubscribe.
+						break
+					}
+					return
+				}
+				status := resp.GetStatus()
+				if status == nil {
+					continue
+				}
+
+				res <- WorkspaceUpdate{
+					InstanceID:  status.Id,
+					WorkspaceID: status.Metadata.MetaId,
+					OwnerID:     status.Metadata.Owner,
+					Failed:      status.Conditions.Failed != "",
+					Phase:       status.Phase,
+				}
 			}
 		}
 	}()
+
 	return res, nil
 }
 

dev/loadgen/pkg/observer/success.go

@@ -7,6 +7,7 @@ package observer
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -73,13 +74,13 @@ func (o *SuccessObserver) Wait(ctx context.Context, expected int) error {
 		case <-ticker.C:
 			o.m.Lock()
 			running := 0
-			stopped := 0
-			for _, ws := range o.workspaces {
+			var stopped []string
+			for id, ws := range o.workspaces {
 				switch ws.Phase {
 				case api.WorkspacePhase_RUNNING:
 					running += 1
 				case api.WorkspacePhase_STOPPED:
-					stopped += 1
+					stopped = append(stopped, id)
 				}
 			}
 
@@ -88,9 +89,9 @@ func (o *SuccessObserver) Wait(ctx context.Context, expected int) error {
 			}
 
 			// Quit early if too many workspaces have stopped already. They'll never become ready.
-			maxRunning := len(o.workspaces) - stopped
+			maxRunning := len(o.workspaces) - len(stopped)
 			if float32(maxRunning) < float32(len(o.workspaces))*o.successRate {
-				return fmt.Errorf("too many workspaces in stopped state (%d), will never get enough ready workspaces", stopped)
+				return fmt.Errorf("too many workspaces in stopped state (%d), will never get enough ready workspaces. Stopped workspaces: %v", len(stopped), strings.Join(stopped, ", "))
 			}
 
 			o.m.Unlock()