[wsman-mk2] Replace in memory state handling

Author: Furisto

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

@@ -85,7 +85,7 @@ func (ssc *SnapshotReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, nil
 	}
 
-	snapshotURL, snapshotName, snapshotErr := ssc.operations.SnapshotIDs(snapshot.Spec.WorkspaceID)
+	snapshotURL, snapshotName, snapshotErr := ssc.operations.SnapshotIDs(ctx, snapshot.Spec.WorkspaceID)
 	if snapshotErr != nil {
 		return ctrl.Result{}, snapshotErr
 	}

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

@@ -148,7 +148,7 @@ func (wsc *WorkspaceController) handleWorkspaceInit(ctx context.Context, ws *wor
 		}
 
 		initStart := time.Now()
-		alreadyInit, failure, initErr := wsc.operations.InitWorkspaceContent(ctx, InitContentOptions{
+		failure, initErr := wsc.operations.InitWorkspaceContent(ctx, InitContentOptions{
 			Meta: WorkspaceMeta{
 				Owner:       ws.Spec.Ownership.Owner,
 				WorkspaceId: ws.Spec.Ownership.WorkspaceID,
@@ -158,10 +158,6 @@ func (wsc *WorkspaceController) handleWorkspaceInit(ctx context.Context, ws *wor
 			Headless:    ws.IsHeadless(),
 		})
 
-		if alreadyInit {
-			return ctrl.Result{}, nil
-		}
-
 		err = retry.RetryOnConflict(retryParams, func() error {
 			if err := wsc.Get(ctx, req.NamespacedName, ws); err != nil {
 				return err
@@ -226,7 +222,7 @@ func (wsc *WorkspaceController) handleWorkspaceStop(ctx context.Context, ws *wor
 	if ws.Spec.Type == workspacev1.WorkspaceTypeRegular {
 		snapshotName = storage.DefaultBackup
 	} else {
-		snapshotUrl, snapshotName, err = wsc.operations.SnapshotIDs(ws.Name)
+		snapshotUrl, snapshotName, err = wsc.operations.SnapshotIDs(ctx, ws.Name)
 		if err != nil {
 			return ctrl.Result{}, err
 		}
@@ -250,7 +246,7 @@ func (wsc *WorkspaceController) handleWorkspaceStop(ctx context.Context, ws *wor
 		}
 	}
 
-	alreadyDisposing, gitStatus, disposeErr := wsc.operations.DisposeWorkspace(ctx, DisposeOptions{
+	gitStatus, disposeErr := wsc.operations.DisposeWorkspace(ctx, DisposeOptions{
 		Meta: WorkspaceMeta{
 			Owner:       ws.Spec.Ownership.Owner,
 			WorkspaceId: ws.Spec.Ownership.WorkspaceID,
@@ -262,10 +258,6 @@ func (wsc *WorkspaceController) handleWorkspaceStop(ctx context.Context, ws *wor
 		UpdateGitStatus:   ws.Spec.Type == workspacev1.WorkspaceTypeRegular,
 	})
 
-	if alreadyDisposing {
-		return ctrl.Result{}, nil
-	}
-
 	err = retry.RetryOnConflict(retryParams, func() error {
 		if err := wsc.Get(ctx, req.NamespacedName, ws); err != nil {
 			return err

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

@@ -41,8 +41,9 @@ const (
 )
 
 type WorkspaceOperations struct {
-	config                 content.Config
-	store                  *session.Store
+	config content.Config
+	//store                  *session.Store
+	provider               *WorkspaceProvider
 	backupWorkspaceLimiter chan struct{}
 	metrics                *content.Metrics
 }
@@ -67,15 +68,15 @@ type DisposeOptions struct {
 	SnapshotName      string
 }
 
-func NewWorkspaceOperations(config content.Config, store *session.Store, reg prometheus.Registerer) (*WorkspaceOperations, error) {
+func NewWorkspaceOperations(config content.Config, store *WorkspaceProvider, reg prometheus.Registerer) (*WorkspaceOperations, error) {
 	waitingTimeHist, waitingTimeoutCounter, err := content.RegisterConcurrentBackupMetrics(reg, "_mk2")
 	if err != nil {
 		return nil, err
 	}
 
 	return &WorkspaceOperations{
 		config: config,
-		store:  store,
+		//store:  store,
 		metrics: &content.Metrics{
 			BackupWaitingTimeHist:       waitingTimeHist,
 			BackupWaitingTimeoutCounter: waitingTimeoutCounter,
@@ -85,34 +86,26 @@ func NewWorkspaceOperations(config content.Config, store *session.Store, reg pro
 	}, nil
 }
 
-func (wso *WorkspaceOperations) InitWorkspaceContent(ctx context.Context, options InitContentOptions) (bool, string, error) {
-	res, err := wso.store.NewWorkspace(
-		ctx, options.Meta.InstanceId, filepath.Join(wso.store.Location, options.Meta.InstanceId),
+func (wso *WorkspaceOperations) InitWorkspaceContent(ctx context.Context, options InitContentOptions) (string, error) {
+	ws, err := wso.provider.Create(ctx, options.Meta.InstanceId, filepath.Join(wso.provider.Location, options.Meta.InstanceId),
 		wso.creator(options.Meta.Owner, options.Meta.WorkspaceId, options.Meta.InstanceId, options.Initializer, false))
-	if errors.Is(err, storage.ErrNotFound) {
-		return false, "", nil
-	}
-
-	if errors.Is(err, session.ErrAlreadyExists) {
-		return true, "", nil
-	}
 
 	if err != nil {
-		return false, "bug: cannot add workspace to store", xerrors.Errorf("cannot add workspace to store: %w", err)
+		return "bug: cannot add workspace to store", xerrors.Errorf("cannot add workspace to store: %w", err)
 	}
 
-	rs, ok := res.NonPersistentAttrs[session.AttrRemoteStorage].(storage.DirectAccess)
+	rs, ok := ws.NonPersistentAttrs[session.AttrRemoteStorage].(storage.DirectAccess)
 	if rs == nil || !ok {
-		return false, "bug: workspace has no remote storage", xerrors.Errorf("workspace has no remote storage")
+		return "bug: workspace has no remote storage", xerrors.Errorf("workspace has no remote storage")
 	}
 	ps, err := storage.NewPresignedAccess(&wso.config.Storage)
 	if err != nil {
-		return false, "bug: no presigned storage available", xerrors.Errorf("no presigned storage available: %w", err)
+		return "bug: no presigned storage available", xerrors.Errorf("no presigned storage available: %w", err)
 	}
 
 	remoteContent, err := content.CollectRemoteContent(ctx, rs, ps, options.Meta.Owner, options.Initializer)
 	if err != nil {
-		return false, "remote content error", xerrors.Errorf("remote content error: %w", err)
+		return "remote content error", xerrors.Errorf("remote content error: %w", err)
 	}
 
 	// Initialize workspace.
@@ -138,13 +131,18 @@ func (wso *WorkspaceOperations) InitWorkspaceContent(ctx context.Context, option
 		},
 	}
 
-	err = content.RunInitializer(ctx, res.Location, options.Initializer, remoteContent, opts)
+	err = content.RunInitializer(ctx, ws.Location, options.Initializer, remoteContent, opts)
 	if err != nil {
 		glog.Infof("error running initializer %v", err)
-		return false, err.Error(), err
+		return err.Error(), err
+	}
+
+	err = ws.Persist()
+	if err != nil {
+		return "cannot persist workspace", err
 	}
 
-	return false, "", nil
+	return "", nil
 }
 
 func (wso *WorkspaceOperations) creator(owner, workspaceId, instanceId string, init *csapi.WorkspaceInitializer, storageDisabled bool) session.WorkspaceFactory {
@@ -173,25 +171,14 @@ func (wso *WorkspaceOperations) creator(owner, workspaceId, instanceId string, i
 	}
 }
 
-func (wso *WorkspaceOperations) DisposeWorkspace(ctx context.Context, opts DisposeOptions) (bool, *csapi.GitStatus, error) {
-	sess := wso.store.Get(opts.Meta.InstanceId)
-	if sess == nil {
-		return false, nil, fmt.Errorf("cannot find workspace %s during DisposeWorkspace", opts.Meta.InstanceId)
-	}
-
-	// Maybe there's someone else already trying to dispose the workspace.
-	// In that case we'll simply wait for that to happen.
-	done, repo, err := sess.WaitOrMarkForDisposal(ctx)
+func (wso *WorkspaceOperations) DisposeWorkspace(ctx context.Context, opts DisposeOptions) (*csapi.GitStatus, error) {
+	ws, err := wso.provider.Get(ctx, opts.Meta.InstanceId)
 	if err != nil {
-		return false, nil, fmt.Errorf("failed to wait for workspace disposal of %s", opts.Meta.InstanceId)
-	}
-
-	if done {
-		return true, repo, nil
+		return nil, fmt.Errorf("cannot find workspace %s during DisposeWorkspace", opts.Meta.InstanceId)
 	}
 
-	if sess.RemoteStorageDisabled {
-		return false, nil, xerrors.Errorf("workspace has no remote storage")
+	if ws.RemoteStorageDisabled {
+		return nil, fmt.Errorf("workspace has no remote storage")
 	}
 
 	if opts.BackupLogs {
@@ -202,14 +189,15 @@ func (wso *WorkspaceOperations) DisposeWorkspace(ctx context.Context, opts Dispo
 		}
 	}
 
-	err = wso.uploadWorkspaceContent(ctx, sess, opts.SnapshotName)
+	err = wso.uploadWorkspaceContent(ctx, ws, opts.SnapshotName)
 	if err != nil {
-		return false, nil, xerrors.Errorf("final backup failed for workspace %s", opts.Meta.InstanceId)
+		return nil, fmt.Errorf("final backup failed for workspace %s", opts.Meta.InstanceId)
 	}
 
+	var repo *csapi.GitStatus
 	if opts.UpdateGitStatus {
 		// Update the git status prior to deleting the workspace
-		repo, err = sess.UpdateGitStatus(ctx, false)
+		repo, err = ws.UpdateGitStatus(ctx, false)
 		if err != nil {
 			// do not fail workspace because we were unable to get git status
 			// which can happen for various reasons, including user corrupting his .git folder somehow
@@ -219,22 +207,22 @@ func (wso *WorkspaceOperations) DisposeWorkspace(ctx context.Context, opts Dispo
 		}
 	}
 
-	if err = sess.Dispose(ctx); err != nil {
+	if err = ws.Dispose(ctx); err != nil {
 		glog.WithError(err).Error("cannot dispose session")
 	}
 
 	// remove workspace daemon directory in the node
-	if err := os.RemoveAll(sess.ServiceLocDaemon); err != nil {
+	if err := os.RemoveAll(ws.ServiceLocDaemon); err != nil {
 		glog.WithError(err).Error("cannot delete workspace daemon directory")
 	}
 
-	return false, repo, nil
+	return repo, nil
 }
 
-func (wso *WorkspaceOperations) SnapshotIDs(workspaceID string) (snapshotUrl, snapshotName string, err error) {
-	sess := wso.store.Get(workspaceID)
-	if sess == nil {
-		return "", "", fmt.Errorf("cannot find workspace %s during SnapshotName", workspaceID)
+func (wso *WorkspaceOperations) SnapshotIDs(ctx context.Context, workspaceID string) (snapshotUrl, snapshotName string, err error) {
+	sess, err := wso.provider.Get(ctx, workspaceID)
+	if err != nil {
+		return "", "", fmt.Errorf("cannot find workspace %s during SnapshotName: %w", workspaceID, err)
 	}
 
 	baseName := fmt.Sprintf("snapshot-%d", time.Now().UnixNano())
@@ -258,16 +246,16 @@ func (wso *WorkspaceOperations) TakeSnapshot(ctx context.Context, workspaceID, s
 		return fmt.Errorf("workspaceID is required")
 	}
 
-	sess := wso.store.Get(workspaceID)
-	if sess == nil {
+	ws, err := wso.provider.Get(ctx, workspaceID)
+	if err != nil {
 		return fmt.Errorf("cannot find workspace %s during DisposeWorkspace", workspaceID)
 	}
 
-	if sess.RemoteStorageDisabled {
+	if ws.RemoteStorageDisabled {
 		return fmt.Errorf("workspace has no remote storage")
 	}
 
-	err = wso.uploadWorkspaceContent(ctx, sess, snapshotName)
+	err = wso.uploadWorkspaceContent(ctx, ws, snapshotName)
 	if err != nil {
 		return fmt.Errorf("snapshot failed for workspace %s", workspaceID)
 	}

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

@@ -0,0 +1,84 @@
+// Copyright (c) 2023 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 controller
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/gitpod-io/gitpod/common-go/log"
+	"github.com/gitpod-io/gitpod/common-go/tracing"
+	"github.com/opentracing/opentracing-go"
+
+	"github.com/gitpod-io/gitpod/ws-daemon/pkg/internal/session"
+)
+
+type WorkspaceProvider struct {
+	hooks    map[session.WorkspaceState][]session.WorkspaceLivecycleHook
+	Location string
+}
+
+func NewWorkspaceProvider(hooks map[session.WorkspaceState][]session.WorkspaceLivecycleHook, location string) *WorkspaceProvider {
+	return &WorkspaceProvider{
+		hooks:    hooks,
+		Location: location,
+	}
+}
+
+func (wf *WorkspaceProvider) Create(ctx context.Context, instanceID, location string, create session.WorkspaceFactory) (ws *session.Workspace, err error) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "Store.NewWorkspace")
+	tracing.ApplyOWI(span, log.OWI("", "", instanceID))
+	defer tracing.FinishSpan(span, &err)
+
+	ws, err = create(ctx, location)
+	if err != nil {
+		return nil, err
+	}
+
+	err = wf.runLifecycleHooks(ctx, ws, session.WorkspaceInitializing)
+	if err != nil {
+		return nil, err
+	}
+
+	return nil, nil
+}
+
+func (wf *WorkspaceProvider) Get(ctx context.Context, instanceID string) (*session.Workspace, error) {
+	path := filepath.Join(wf.Location, fmt.Sprintf("%s.workspace.json", instanceID))
+	return loadWorkspace(ctx, path)
+}
+
+func (s *WorkspaceProvider) runLifecycleHooks(ctx context.Context, ws *session.Workspace, state session.WorkspaceState) error {
+	hooks := s.hooks[state]
+	log.WithFields(ws.OWI()).WithField("state", state).WithField("hooks", len(hooks)).Debug("running lifecycle hooks")
+
+	for _, h := range hooks {
+		err := h(ctx, ws)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func loadWorkspace(ctx context.Context, path string) (ws *session.Workspace, err error) {
+	span, _ := opentracing.StartSpanFromContext(ctx, "loadWorkspace")
+	defer tracing.FinishSpan(span, &err)
+
+	fc, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("cannot load session file: %w", err)
+	}
+
+	err = json.Unmarshal(fc, ws)
+	if err != nil {
+		return nil, fmt.Errorf("cannot unmarshal session file: %w", err)
+	}
+
+	return ws, nil
+}

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

@@ -35,7 +35,6 @@ import (
 	"github.com/gitpod-io/gitpod/ws-daemon/pkg/cpulimit"
 	"github.com/gitpod-io/gitpod/ws-daemon/pkg/diskguard"
 	"github.com/gitpod-io/gitpod/ws-daemon/pkg/dispatch"
-	"github.com/gitpod-io/gitpod/ws-daemon/pkg/internal/session"
 	"github.com/gitpod-io/gitpod/ws-daemon/pkg/iws"
 	"github.com/gitpod-io/gitpod/ws-daemon/pkg/netlimit"
 	"github.com/gitpod-io/gitpod/ws-daemon/pkg/quota"
@@ -193,18 +192,26 @@ func NewDaemon(config Config) (*Daemon, error) {
 			return nil, err
 		}
 
-		store, err := session.NewStore(context.Background(), contentCfg.WorkingArea, content.WorkspaceLifecycleHooks(
+		hooks := content.WorkspaceLifecycleHooks(
 			contentCfg,
 			func(instanceID string) bool { return true },
 			&iws.Uidmapper{Config: config.Uidmapper, Runtime: containerRuntime},
 			xfs,
 			config.CPULimit.CGroupBasePath,
-		))
-		if err != nil {
-			return nil, err
-		}
-
-		workspaceOps, err := controller.NewWorkspaceOperations(contentCfg, store, wrappedReg)
+		)
+
+		// store, err := session.NewStore(context.Background(), contentCfg.WorkingArea, content.WorkspaceLifecycleHooks(
+		// 	contentCfg,
+		// 	func(instanceID string) bool { return true },
+		// 	&iws.Uidmapper{Config: config.Uidmapper, Runtime: containerRuntime},
+		// 	xfs,
+		// 	config.CPULimit.CGroupBasePath,
+		// ))
+		// if err != nil {
+		// 	return nil, err
+		// }
+
+		workspaceOps, err := controller.NewWorkspaceOperations(contentCfg, controller.NewWorkspaceProvider(hooks, contentCfg.WorkingArea), wrappedReg)
 		if err != nil {
 			return nil, err
 		}