WIP: Fix podRecreationTimeout + containerd cleanup

Author: geropl

components/ws-daemon/pkg/container/container.go

@@ -47,6 +47,9 @@ type Runtime interface {
 
 	// IsContainerdReady returns is the status of containerd.
 	IsContainerdReady(ctx context.Context) (bool, error)
+
+	// DisposeContainer removes a stopped container, and everything we know about it
+	DisposeContainer(ctx context.Context, workspaceInstanceID string)
 }
 
 var (

components/ws-daemon/pkg/container/containerd.go

@@ -427,6 +427,21 @@ func (s *Containerd) WaitForContainerStop(ctx context.Context, workspaceInstance
 	}
 }
 
+func (s *Containerd) DisposeContainer(ctx context.Context, workspaceInstanceID string) {
+	s.cond.L.Lock()
+	defer s.cond.L.Unlock()
+
+	info, ok := s.wsiIdx[workspaceInstanceID]
+	if !ok {
+		// seems we are already done here
+		return
+	}
+
+	delete(s.wsiIdx, info.InstanceID)
+	delete(s.podIdx, info.PodName)
+	delete(s.cntIdx, info.ID)
+}
+
 // ContainerExists finds out if a container with the given ID exists.
 func (s *Containerd) ContainerExists(ctx context.Context, id ID) (exists bool, err error) {
 	_, err = s.Client.ContainerService().Get(ctx, string(id))

components/ws-daemon/pkg/controller/workspace_operations.go

@@ -304,15 +304,15 @@ func (wso *DefaultWorkspaceOperations) WipeWorkspace(ctx context.Context, instan
 		return err
 	}
 
+	// dispose all running "dispatch handlers", e.g. all code running on the "pod informer"-triggered part of ws-daemon
+	wso.dispatch.DisposeWorkspace(ctx, instanceID)
+
 	// remove workspace daemon directory in the node
 	if err := os.RemoveAll(ws.ServiceLocDaemon); err != nil {
 		glog.WithError(err).WithFields(ws.OWI()).Error("cannot delete workspace daemon directory")
 		return err
 	}
 
-	// dispose all running "dispatch handlers", e.g. all code running on the "pod informer"-triggered part of ws-daemon
-	wso.dispatch.DisposeWorkspace(ctx, instanceID)
-
 	// remove the reference from the WorkspaceProvider, e.g. the "workspace controller" part of ws-daemon
 	wso.provider.Remove(ctx, instanceID)
 

components/ws-daemon/pkg/dispatch/dispatch.go

@@ -7,6 +7,7 @@ package dispatch
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -59,7 +60,8 @@ func NewDispatch(runtime container.Runtime, kubernetes kubernetes.Interface, k8s
 		Listener:            listener,
 		NodeName:            nodename,
 
-		ctxs: make(map[string]*workspaceState),
+		ctxs:         make(map[string]*workspaceState),
+		disposedCtxs: make(map[string]struct{}),
 	}
 
 	return d, nil
@@ -76,9 +78,10 @@ type Dispatch struct {
 
 	Listener []Listener
 
-	stopchan chan struct{}
-	ctxs     map[string]*workspaceState
-	mu       sync.Mutex
+	stopchan     chan struct{}
+	ctxs         map[string]*workspaceState
+	disposedCtxs map[string]struct{}
+	mu           sync.Mutex
 }
 
 type workspaceState struct {
@@ -187,7 +190,7 @@ func (d *Dispatch) WorkspaceExistsOnNode(instanceID string) (ok bool) {
 	return
 }
 
-// DisposeWorkspace makes sure
+// DisposeWorkspace disposes the workspace incl. all running handler code for that pod
 func (d *Dispatch) DisposeWorkspace(ctx context.Context, instanceID string) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
@@ -206,10 +209,26 @@ func (d *Dispatch) DisposeWorkspace(ctx context.Context, instanceID string) {
 	// ...and wait for all long-running/async processes/go-routines to finish
 	state.HandlerWaitGroup.Wait()
 
+	// Make sure the container is stopped
+	err := d.Runtime.WaitForContainerStop(ctx, instanceID)
+	if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) {
+		log.WithError(err).WithFields(log.WorkspaceInstanceID(instanceID)).Error("unexpected waiting for container to stop")
+	}
+
+	// Make the runtome drop all state it might still have about this workspace
+	d.Runtime.DisposeContainer(ctx, instanceID)
+
+	// Mark as disposed, so we do not handle any further updates for it (except deletion)
+	d.disposedCtxs[disposedKey(instanceID, state.Workspace.Pod)] = struct{}{}
+
 	delete(d.ctxs, instanceID)
 	log.WithField("instanceID", instanceID).Debugf("WS DISPOSE DONE: %s", instanceID)
 }
 
+func disposedKey(instanceID string, pod *corev1.Pod) string {
+	return fmt.Sprintf("%s-%s", instanceID, pod.CreationTimestamp.String())
+}
+
 func (d *Dispatch) handlePodUpdate(oldPod, newPod *corev1.Pod) {
 	workspaceID, ok := newPod.Labels[wsk8s.MetaIDLabel]
 	if !ok {
@@ -222,6 +241,11 @@ func (d *Dispatch) handlePodUpdate(oldPod, newPod *corev1.Pod) {
 	if d.NodeName != "" && newPod.Spec.NodeName != d.NodeName {
 		return
 	}
+	disposedKey := disposedKey(workspaceInstanceID, newPod)
+	if _, alreadyDisposed := d.disposedCtxs[disposedKey]; alreadyDisposed {
+		log.WithField("disposedKey", disposedKey).Debug("DROPPING POD UPDATE FOR DISPOSED POD")
+		return
+	}
 	log.WithField("instanceID", workspaceInstanceID).Debugf("POD UPDATE: %s", workspaceInstanceID)
 
 	d.mu.Lock()
@@ -340,6 +364,8 @@ func (d *Dispatch) handlePodDeleted(pod *corev1.Pod) {
 	if state.Cancel != nil {
 		state.Cancel()
 	}
+
 	delete(d.ctxs, instanceID)
+
 	log.WithField("instanceID", instanceID).Debugf("POD DELETED DONE: %s", instanceID)
 }

components/ws-manager-api/go/crd/v1/workspace_types.go

@@ -170,10 +170,11 @@ func (ps PortSpec) Equal(other PortSpec) bool {
 
 // WorkspaceStatus defines the observed state of Workspace
 type WorkspaceStatus struct {
-	PodStarts    int    `json:"podStarts"`
-	PodRecreated int    `json:"podRecreated"`
-	URL          string `json:"url,omitempty" scrub:"redact"`
-	OwnerToken   string `json:"ownerToken,omitempty" scrub:"redact"`
+	PodStarts       int          `json:"podStarts"`
+	PodRecreated    int          `json:"podRecreated"`
+	PodDeletionTime *metav1.Time `json:"podDeletionTime,omitempty"`
+	URL             string       `json:"url,omitempty" scrub:"redact"`
+	OwnerToken      string       `json:"ownerToken,omitempty" scrub:"redact"`
 
 	// +kubebuilder:default=Unknown
 	Phase WorkspacePhase `json:"phase,omitempty"`

components/ws-manager-mk2/config/crd/bases/workspace.gitpod.io_workspaces.yaml

@@ -545,6 +545,9 @@ spec:
                 type: integer
               podRecreated:
                 type: integer
+              podDeletionTime:
+                format: date-time
+                type: string
               runtime:
                 properties:
                   hostIP:

components/ws-manager-mk2/controllers/status.go

@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	wsk8s "github.com/gitpod-io/gitpod/common-go/kubernetes"
 	"github.com/gitpod-io/gitpod/common-go/tracing"
@@ -71,6 +72,14 @@ func (r *WorkspaceReconciler) updateWorkspaceStatus(ctx context.Context, workspa
 			workspace.Status.Phase = workspacev1.WorkspacePhaseStopped
 		}
 
+		log.WithValues("podDeletionTime", workspace.Status.PodDeletionTime).Info("PodDeletionTimeValue")
+		if workspace.Status.Phase == workspacev1.WorkspacePhaseStopped && workspace.Status.PodDeletionTime == nil {
+			// Set the timestamp when we first saw the pod as deleted.
+			// This is used for the delaying eventual pod restarts
+			podDeletionTime := metav1.NewTime(time.Now())
+			workspace.Status.PodDeletionTime = &podDeletionTime
+		}
+
 		workspace.UpsertConditionOnStatusChange(workspacev1.NewWorkspaceConditionContainerRunning(metav1.ConditionFalse))
 		return nil
 	case 1:

components/ws-manager-mk2/controllers/suite_test.go

@@ -159,7 +159,9 @@ func newTestConfig() config.Configuration {
 				Name: "default",
 			},
 		},
-		WorkspaceURLTemplate: "{{ .ID }}-{{ .Prefix }}-{{ .Host }}",
+		WorkspaceURLTemplate:    "{{ .ID }}-{{ .Prefix }}-{{ .Host }}",
+		PodRecreationMaxRetries: 3,
+		PodRecreationBackoff:    util.Duration(500 * time.Millisecond),
 	}
 }
 

components/ws-manager-mk2/controllers/workspace_controller.go

@@ -181,24 +181,6 @@ func (r *WorkspaceReconciler) actOnStatus(ctx context.Context, workspace *worksp
 		// if there isn't a workspace pod and we're not currently deleting this workspace,// create one.
 		switch {
 		case workspace.Status.PodStarts == 0 || workspace.Status.PodStarts-workspace.Status.PodRecreated < 1:
-			if workspace.Status.PodRecreated > 0 {
-				// This is a re-creation: Make sure to wait at least for
-				c := wsk8s.GetCondition(workspace.Status.Conditions, string(workspacev1.WorkspaceConditionPodRejected))
-				if c == nil {
-					err = fmt.Errorf("failed to retrieve PodRejected condition")
-					log.Error(err, "failed to trigger pod recreation")
-					return ctrl.Result{}, err
-				}
-
-				recreationTimeout := r.podRecreationTimeout()
-				waitTime := time.Until(c.LastTransitionTime.Add(recreationTimeout))
-				if waitTime > 0 {
-					log.WithValues("waitTime", waitTime).Info("waiting for pod recreation timeout")
-					return ctrl.Result{Requeue: true, RequeueAfter: waitTime}, nil
-				}
-				log.WithValues("waitedTime", waitTime.Abs().String()).Info("waited for pod recreation timeout")
-			}
-
 			sctx, err := newStartWorkspaceContext(ctx, r.Config, workspace)
 			if err != nil {
 				log.Error(err, "unable to create startWorkspace context")
@@ -244,6 +226,21 @@ func (r *WorkspaceReconciler) actOnStatus(ctx context.Context, workspace *worksp
 			}
 			log.WithValues("PodStarts", workspace.Status.PodStarts, "PodRecreated", workspace.Status.PodRecreated, "Phase", workspace.Status.Phase).Info("trigger pod recreation")
 
+			// Make sure to wait for "recreationTimeout" before creating the pod again
+			if workspace.Status.PodDeletionTime == nil {
+				log.Info("want to wait for pod recreation timeout, but podDeletionTime not set (yet)")
+				return ctrl.Result{Requeue: true, RequeueAfter: 5 * time.Second}, nil
+			}
+
+			recreationTimeout := r.podRecreationTimeout()
+			podDeletionTime := workspace.Status.PodDeletionTime.Time
+			waitTime := time.Until(podDeletionTime.Add(recreationTimeout))
+			if waitTime > 0 {
+				log.WithValues("waitTime", waitTime).Info("waiting for pod recreation timeout")
+				return ctrl.Result{Requeue: true, RequeueAfter: waitTime}, nil
+			}
+			log.WithValues("waitedTime", waitTime.Abs().String()).Info("waited for pod recreation timeout")
+
 			// Must persist the modification pod starts, and ensure we retry on conflict.
 			// If we fail to persist this value, it's possible that the Pod gets recreated endlessly
 			// when the workspace stops, due to PodStarts still being 0 when the original Pod
@@ -269,8 +266,7 @@ func (r *WorkspaceReconciler) actOnStatus(ctx context.Context, workspace *worksp
 			r.metrics.forgetWorkspace(workspace)
 
 			r.Recorder.Event(workspace, corev1.EventTypeNormal, "Recreating", "")
-			requeueAfter := r.podRecreationTimeout()
-			return ctrl.Result{Requeue: true, RequeueAfter: requeueAfter}, nil
+			return ctrl.Result{Requeue: true}, nil
 
 		case workspace.Status.Phase == workspacev1.WorkspacePhaseStopped:
 			if err := r.deleteWorkspaceSecrets(ctx, workspace); err != nil {
@@ -377,7 +373,7 @@ func (r *WorkspaceReconciler) actOnStatus(ctx context.Context, workspace *worksp
 }
 
 func (r *WorkspaceReconciler) podRecreationTimeout() time.Duration {
-	recreationTimeout := 5 * time.Second
+	recreationTimeout := 15 * time.Second // waiting less time creates issues with ws-daemon's pod-centric control loop ("Dispatch") if the workspace ends up on the same node again
 	if r.Config.PodRecreationBackoff != 0 {
 		recreationTimeout = time.Duration(r.Config.PodRecreationBackoff)
 	}