[ws-daemon] Review comments

- Reuse workspace
- Update metric name
- Check if backup timed out
- Ensure temp directory is deleted

Author: Furisto

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

@@ -32,8 +32,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
-// DefaultRetry is the recommended retry for a conflict where multiple clients
-// are making changes to the same resource.
 var retryParams = wait.Backoff{
 	Steps:    10,
 	Duration: 10 * time.Millisecond,
@@ -176,29 +174,28 @@ func (wsc *WorkspaceController) handleWorkspaceInit(ctx context.Context, ws *wor
 		}
 
 		err = retry.RetryOnConflict(retryParams, func() error {
-			var workspace workspacev1.Workspace
-			if err := wsc.Get(ctx, req.NamespacedName, &workspace); err != nil {
+			if err := wsc.Get(ctx, req.NamespacedName, ws); err != nil {
 				return err
 			}
 
 			if failure != "" {
-				workspace.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
+				ws.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
 					Type:               string(workspacev1.WorkspaceConditionContentReady),
 					Status:             metav1.ConditionFalse,
 					Message:            failure,
 					Reason:             "InitializationFailure",
 					LastTransitionTime: metav1.Now(),
 				})
 			} else {
-				workspace.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
+				ws.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
 					Type:               string(workspacev1.WorkspaceConditionContentReady),
 					Status:             metav1.ConditionTrue,
 					Reason:             "InitializationSuccess",
 					LastTransitionTime: metav1.Now(),
 				})
 			}
 
-			return wsc.Status().Update(ctx, &workspace)
+			return wsc.Status().Update(ctx, ws)
 		})
 
 		return ctrl.Result{}, err
@@ -243,31 +240,30 @@ func (wsc *WorkspaceController) handleWorkspaceStop(ctx context.Context, ws *wor
 	}
 
 	err = retry.RetryOnConflict(retryParams, func() error {
-		var workspace workspacev1.Workspace
-		if err := wsc.Get(ctx, req.NamespacedName, &workspace); err != nil {
+		if err := wsc.Get(ctx, req.NamespacedName, ws); err != nil {
 			return err
 		}
 
-		workspace.Status.GitStatus = toWorkspaceGitStatus(gitStatus)
+		ws.Status.GitStatus = toWorkspaceGitStatus(gitStatus)
 
 		if disposeErr != nil {
-			workspace.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
+			ws.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
 				Type:               string(workspacev1.WorkspaceConditionBackupFailure),
 				Status:             metav1.ConditionTrue,
 				Reason:             "BackupFailed",
 				Message:            disposeErr.Error(),
 				LastTransitionTime: metav1.Now(),
 			})
 		} else {
-			workspace.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
+			ws.Status.Conditions = wsk8s.AddUniqueCondition(ws.Status.Conditions, metav1.Condition{
 				Type:               string(workspacev1.WorkspaceConditionBackupComplete),
 				Status:             metav1.ConditionTrue,
 				Reason:             "BackupComplete",
 				LastTransitionTime: metav1.Now(),
 			})
 		}
 
-		return wsc.Status().Update(ctx, &workspace)
+		return wsc.Status().Update(ctx, ws)
 	})
 
 	return ctrl.Result{}, err

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

@@ -66,7 +66,7 @@ type DisposeOptions struct {
 }
 
 func NewWorkspaceOperations(config content.Config, store *session.Store, reg prometheus.Registerer) (*WorkspaceOperations, error) {
-	waitingTimeHist, waitingTimeoutCounter, err := content.RegisterConcurrentBackupMetrics(reg, "mk2")
+	waitingTimeHist, waitingTimeoutCounter, err := content.RegisterConcurrentBackupMetrics(reg, "_mk2")
 	if err != nil {
 		return nil, err
 	}
@@ -274,19 +274,27 @@ func (wso *WorkspaceOperations) uploadWorkspaceLogs(ctx context.Context, opts Di
 func (wso *WorkspaceOperations) uploadWorkspaceContent(ctx context.Context, sess *session.Workspace) error {
 	var backupName = storage.DefaultBackup
 	// Avoid too many simultaneous backups in order to avoid excessive memory utilization.
+	var timedOut bool
 	waitStart := time.Now()
 	select {
 	case wso.backupWorkspaceLimiter <- struct{}{}:
 	case <-time.After(15 * time.Minute):
 		// we timed out on the rate limit - let's upload anyways, because we don't want to actually block
 		// an upload. If we reach this point, chances are other things are broken. No upload should ever
 		// take this long.
+		timedOut = true
 		wso.metrics.BackupWaitingTimeoutCounter.Inc()
 	}
 
-	wso.metrics.BackupWaitingTimeHist.Observe(time.Since(waitStart).Seconds())
+	waitTime := time.Since(waitStart)
+	wso.metrics.BackupWaitingTimeHist.Observe(waitTime.Seconds())
 
 	defer func() {
+		// timeout -> we did not add to the limiter
+		if timedOut {
+			return
+		}
+
 		<-wso.backupWorkspaceLimiter
 	}()
 
@@ -312,12 +320,24 @@ func (wso *WorkspaceOperations) uploadWorkspaceContent(ctx context.Context, sess
 		tmpfSize int64
 	)
 
+	defer func() {
+		if tmpf != nil {
+			os.Remove(tmpf.Name())
+		}
+	}()
+
 	err = retryIfErr(ctx, wso.config.Backup.Attempts, glog.WithFields(sess.OWI()).WithField("op", "create archive"), func(ctx context.Context) (err error) {
 		tmpf, err = os.CreateTemp(wso.config.TmpDir, fmt.Sprintf("wsbkp-%s-*.tar", sess.InstanceID))
 		if err != nil {
 			return
 		}
-		defer tmpf.Close()
+
+		defer func() {
+			tmpf.Close()
+			if err != nil {
+				os.Remove(tmpf.Name())
+			}
+		}()
 
 		var opts []archive.TarOption
 		opts = append(opts)
@@ -357,12 +377,6 @@ func (wso *WorkspaceOperations) uploadWorkspaceContent(ctx context.Context, sess
 	if err != nil {
 		return xerrors.Errorf("cannot create archive: %w", err)
 	}
-	defer func() {
-		if tmpf != nil {
-			// always remove the archive file to not fill up the node needlessly
-			os.Remove(tmpf.Name())
-		}
-	}()
 
 	err = retryIfErr(ctx, wso.config.Backup.Attempts, glog.WithFields(sess.OWI()).WithField("op", "upload layer"), func(ctx context.Context) (err error) {
 		_, _, err = rs.Upload(ctx, tmpf.Name(), backupName, opts...)