WIP: refactor to use instance id

Author: rl-gitpod

components/ws-proxy/pkg/proxy/infoprovider.go

@@ -78,9 +78,9 @@ type WorkspaceInfo struct {
 	// (parsed from URL)
 	IDEPublicPort string
 
-	Ports     []PortInfo
-	Auth      *wsapi.WorkspaceAuthentication
-	StartedAt time.Time
+	Ports []PortInfo
+	Auth  *wsapi.WorkspaceAuthentication
+	Phase wsapi.WorkspacePhase
 }
 
 // PortInfo contains all information ws-proxy needs to know about a workspace port
@@ -266,7 +266,7 @@ func (p *RemoteWorkspaceInfoProvider) listen(client wsapi.WorkspaceManagerClient
 		}
 
 		if status.Phase == wsapi.WorkspacePhase_STOPPED {
-			p.cache.Delete(status.Metadata.MetaId, status.Metadata.StartedAt.AsTime())
+			p.cache.Delete(status.Metadata.MetaId, status.Id)
 		} else {
 			info := mapWorkspaceStatusToInfo(status)
 			p.cache.Insert(info)
@@ -310,11 +310,11 @@ func mapWorkspaceStatusToInfo(status *wsapi.WorkspaceStatus) *WorkspaceInfo {
 		IDEPublicPort: getPortStr(status.Spec.Url),
 		Ports:         portInfos,
 		Auth:          status.Auth,
-		StartedAt:     status.Metadata.StartedAt.AsTime(),
+		Phase:         status.Phase,
 	}
 }
 
-// WorkspaceInfo return the WorkspaceInfo avaiable for the given workspaceID.
+// WorkspaceInfo return the WorkspaceInfo available for the given workspaceID.
 // Callers should make sure their context gets canceled properly. For good measure
 // this function will timeout by itself as well.
 func (p *RemoteWorkspaceInfoProvider) WorkspaceInfo(ctx context.Context, workspaceID string) *WorkspaceInfo {
@@ -360,7 +360,7 @@ func getPortStr(urlStr string) string {
 // workspaceInfoCache stores WorkspaceInfo in a manner which is easy to query for WorkspaceInfoProvider
 type workspaceInfoCache struct {
 	// WorkspaceInfos indexed by workspaceID
-	infos map[string]*WorkspaceInfo
+	infos map[string][]*WorkspaceInfo
 	// WorkspaceCoords indexed by public (proxy) port (string)
 	coordsByPublicPort map[string]*WorkspaceCoords
 
@@ -373,7 +373,7 @@ type workspaceInfoCache struct {
 func newWorkspaceInfoCache() *workspaceInfoCache {
 	var mu sync.RWMutex
 	return &workspaceInfoCache{
-		infos:              make(map[string]*WorkspaceInfo),
+		infos:              make(map[string][]*WorkspaceInfo),
 		coordsByPublicPort: make(map[string]*WorkspaceCoords),
 		mu:                 &mu,
 		cond:               sync.NewCond(&mu),
@@ -384,7 +384,7 @@ func (c *workspaceInfoCache) Reinit(infos []*WorkspaceInfo) {
 	c.cond.L.Lock()
 	defer c.cond.L.Unlock()
 
-	c.infos = make(map[string]*WorkspaceInfo, len(infos))
+	c.infos = make(map[string][]*WorkspaceInfo, len(infos))
 	c.coordsByPublicPort = make(map[string]*WorkspaceCoords, len(c.coordsByPublicPort))
 
 	for _, info := range infos {
@@ -402,19 +402,17 @@ func (c *workspaceInfoCache) Insert(info *WorkspaceInfo) {
 }
 
 func (c *workspaceInfoCache) doInsert(info *WorkspaceInfo) {
-	existing := c.infos[info.WorkspaceID]
-	if existing != nil {
-		// Do not insert if the current workspace is newer
-		// This works around issues when a workspace is deleted then restarted in quick succession
-		// and the stopping event occurs after the replocement pod is started
-		if !existing.StartedAt.IsZero() && !info.StartedAt.IsZero() {
-			if existing.StartedAt.After(info.StartedAt) {
-				log.WithField("workspaceID", existing.WorkspaceID).Debug("ignoring insert of older workspace")
-				return
-			}
+	existingInfos := c.infos[info.WorkspaceID]
+	// Only insert if this is a new instance
+	for i, existingInfo := range existingInfos {
+		if existingInfo.InstanceID == info.InstanceID {
+			c.infos[info.WorkspaceID][i] = info
+			log.WithField("info", info).Debug("ignoring insert of existing instance for workspace")
+			return
 		}
 	}
-	c.infos[info.WorkspaceID] = info
+	log.WithField("workspaceID", info.WorkspaceID).WithField("instanceID", info.InstanceID).Debug("adding instance for workspace")
+	c.infos[info.WorkspaceID] = append(c.infos[info.WorkspaceID], info)
 	c.coordsByPublicPort[info.IDEPublicPort] = &WorkspaceCoords{
 		ID: info.WorkspaceID,
 	}
@@ -427,34 +425,57 @@ func (c *workspaceInfoCache) doInsert(info *WorkspaceInfo) {
 	}
 }
 
-func (c *workspaceInfoCache) Delete(workspaceID string, startedAt time.Time) {
+func (c *workspaceInfoCache) Delete(workspaceID string, instanceID string) {
 	c.cond.L.Lock()
 	defer c.cond.L.Unlock()
 
-	info, present := c.infos[workspaceID]
-	if !present || info == nil {
+	infos, present := c.infos[workspaceID]
+	if !present || infos == nil {
 		return
 	}
-	// Do not delete if the current workspace info is newer
-	// This works around issues when a workspace is deleted then restarted in quick succession
-	// and the delete event occurs after the replocement pod is started
-	if !startedAt.IsZero() && !info.StartedAt.IsZero() {
-		if info.StartedAt.After(startedAt) {
-			log.WithField("workspaceID", workspaceID).WithField("startedAt", startedAt).WithField("info", info).Debug("ignoring delete of older workspace")
+	// Delete only the matching instance ID
+	for _, info := range infos {
+		if info.InstanceID == instanceID {
+			log.WithField("workspaceID", workspaceID).WithField("instanceID", instanceID).Debug("deleting instance for workspace")
+			delete(c.infos, instanceID)
+			delete(c.coordsByPublicPort, info.IDEPublicPort)
 			return
 		}
 	}
-	delete(c.coordsByPublicPort, info.IDEPublicPort)
-	delete(c.infos, workspaceID)
+	log.WithField("workspaceID", workspaceID).WithField("instanceID", instanceID).Debug("ignoring delete of missing instance for workspace")
 }
 
 // WaitFor waits for workspace info until that info is available or the context is canceled.
 func (c *workspaceInfoCache) WaitFor(ctx context.Context, workspaceID string) (w *WorkspaceInfo, ok bool) {
 	c.mu.RLock()
-	w, ok = c.infos[workspaceID]
+	existing, ok := c.infos[workspaceID]
 	c.mu.RUnlock()
+
+	getCandidate := func(infos []*WorkspaceInfo) *WorkspaceInfo {
+		// Bail early...
+		if len(infos) == 1 {
+			return infos[0]
+		}
+		// Find the *best* candidate i.e. prefer any instance ramping up over stopping/stopped
+		// TODO(rl): make this a smarter container? i.e. priorised by (smart) status
+		candidate := infos[0]
+		for _, info := range infos[1:] {
+			if info.Phase < wsapi.WorkspacePhase_INTERRUPTED {
+				if candidate.Phase > info.Phase {
+					candidate = info
+				}
+			} else {
+				// Prefer interrupted
+				if candidate.Phase < info.Phase {
+					candidate = info
+				}
+			}
+		}
+		return candidate
+	}
+
 	if ok {
-		return
+		return getCandidate(existing), true
 	}
 
 	inc := make(chan *WorkspaceInfo)
@@ -469,12 +490,12 @@ func (c *workspaceInfoCache) WaitFor(ctx context.Context, workspaceID string) (w
 				return
 			}
 
-			info, ok := c.infos[workspaceID]
+			infos, ok := c.infos[workspaceID]
 			if !ok {
 				continue
 			}
 
-			inc <- info
+			inc <- getCandidate(infos)
 			return
 		}
 	}()

components/ws-proxy/pkg/proxy/infoprovider_test.go

@@ -15,7 +15,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/grpc"
-	"google.golang.org/protobuf/types/known/timestamppb"
 
 	wsapi "github.com/gitpod-io/gitpod/ws-manager/api"
 	wsmock "github.com/gitpod-io/gitpod/ws-manager/api/mock"
@@ -168,7 +167,7 @@ func TestRemoteInfoProvider(t *testing.T) {
 					if step.Action != nil {
 						act := step.Action(t, prov)
 						if diff := cmp.Diff(step.Expectation, act, cmpopts.IgnoreUnexported(wsapi.PortSpec{}, wsapi.WorkspaceAuthentication{})); diff != "" {
-							t.Errorf("Expectation mismatch (-want +got):\n%s", diff)
+							t.Errorf("%s Expectation mismatch (-want +got):\n%s", t.Name(), diff)
 						}
 					}
 				})
@@ -179,12 +178,10 @@ func TestRemoteInfoProvider(t *testing.T) {
 }
 
 var (
-	startedAt           = time.Date(2021, 1, 1, 00, 00, 00, 00, time.UTC)
 	testWorkspaceStatus = &wsapi.WorkspaceStatus{
 		Id: "e63cb5ff-f4e4-4065-8554-b431a32c0000",
 		Metadata: &wsapi.WorkspaceMetadata{
-			MetaId:    "e63cb5ff-f4e4-4065-8554-b431a32c2714",
-			StartedAt: timestamppb.New(startedAt),
+			MetaId: "e63cb5ff-f4e4-4065-8554-b431a32c2714",
 		},
 		Auth: &wsapi.WorkspaceAuthentication{
 			Admission:  wsapi.AdmissionLevel_ADMIT_OWNER_ONLY,
@@ -219,6 +216,6 @@ var (
 		},
 		URL:         testWorkspaceStatus.Spec.Url,
 		WorkspaceID: testWorkspaceStatus.Metadata.MetaId,
-		StartedAt:   startedAt,
+		Phase:       wsapi.WorkspacePhase_RUNNING,
 	}
 )