Refactor to use in-memory ws activity map

Author: WVerlaek

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

@@ -147,9 +147,6 @@ const (
 	// UserActivity is the time when MarkActive was first called on the workspace
 	WorkspaceConditionFirstUserActivity WorkspaceCondition = "FirstUserActivity"
 
-	// UserActivity is the most recent time when MarkActive was called on the workspace
-	WorkspaceConditionUserActivity WorkspaceCondition = "UserActivity"
-
 	// Closed indicates that a workspace is marked as closed. This will shorten its timeout.
 	WorkspaceConditionClosed WorkspaceCondition = "Closed"
 

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

@@ -13,6 +13,7 @@ import (
 
 	"github.com/gitpod-io/gitpod/common-go/util"
 
+	wsactivity "github.com/gitpod-io/gitpod/ws-manager-mk2/pkg/activity"
 	"github.com/gitpod-io/gitpod/ws-manager/api/config"
 	workspacev1 "github.com/gitpod-io/gitpod/ws-manager/api/crd/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -309,8 +310,10 @@ const (
 )
 
 // isWorkspaceTimedOut determines if a workspace is timed out based on the manager configuration and state the pod is in.
-// This function does NOT use the workspaceTimedoutAnnotation, but rather is used to set that annotation in the first place.
-func isWorkspaceTimedOut(ws *workspacev1.Workspace, pod *corev1.Pod, timeouts config.WorkspaceTimeoutConfiguration) (reason string, err error) {
+// This function does NOT use the Timeout condition, but rather is used to set that condition in the first place.
+//
+//nolint:unused,deadcode TODO: Remove nolint
+func isWorkspaceTimedOut(ws *workspacev1.Workspace, pod *corev1.Pod, timeouts config.WorkspaceTimeoutConfiguration, act *wsactivity.WorkspaceActivity) (reason string, err error) {
 	// workspaceID := ws.Spec.Ownership.WorkspaceID
 	phase := ws.Status.Phase
 
@@ -326,7 +329,7 @@ func isWorkspaceTimedOut(ws *workspacev1.Workspace, pod *corev1.Pod, timeouts co
 
 	// TODO: Use ws or pod's CreationTimestamp?
 	start := ws.ObjectMeta.CreationTimestamp.Time
-	lastActivity := getWorkspaceActivity(ws)
+	lastActivity := act.GetLastActivity(ws.Spec.Ownership.WorkspaceID)
 	isClosed := conditionPresentAndTrue(ws.Status.Conditions, string(workspacev1.WorkspaceConditionClosed))
 
 	switch phase {
@@ -385,15 +388,7 @@ func isWorkspaceTimedOut(ws *workspacev1.Workspace, pod *corev1.Pod, timeouts co
 	}
 }
 
-func getWorkspaceActivity(ws *workspacev1.Workspace) *time.Time {
-	for _, c := range ws.Status.Conditions {
-		if c.Type == string(workspacev1.WorkspaceConditionUserActivity) {
-			return &c.LastTransitionTime.Time
-		}
-	}
-	return nil
-}
-
+//nolint:unused,deadcode TODO: remove nolint
 func formatDuration(d time.Duration) string {
 	d = d.Round(time.Minute)
 	h := d / time.Hour

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

@@ -16,6 +16,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
+	wsactivity "github.com/gitpod-io/gitpod/ws-manager-mk2/pkg/activity"
 	config "github.com/gitpod-io/gitpod/ws-manager/api/config"
 	workspacev1 "github.com/gitpod-io/gitpod/ws-manager/api/crd/v1"
 	"github.com/prometheus/client_golang/prometheus"
@@ -28,11 +29,12 @@ const (
 	kubernetesOperationTimeout = 5 * time.Second
 )
 
-func NewWorkspaceReconciler(c client.Client, scheme *runtime.Scheme, cfg config.Configuration, reg prometheus.Registerer) (*WorkspaceReconciler, error) {
+func NewWorkspaceReconciler(c client.Client, scheme *runtime.Scheme, cfg config.Configuration, reg prometheus.Registerer, activity *wsactivity.WorkspaceActivity) (*WorkspaceReconciler, error) {
 	reconciler := &WorkspaceReconciler{
-		Client: c,
-		Scheme: scheme,
-		Config: cfg,
+		Client:   c,
+		Scheme:   scheme,
+		Config:   cfg,
+		activity: activity,
 	}
 
 	metrics, err := newControllerMetrics(reconciler)
@@ -52,6 +54,7 @@ type WorkspaceReconciler struct {
 
 	Config      config.Configuration
 	metrics     *controllerMetrics
+	activity    *wsactivity.WorkspaceActivity
 	OnReconcile func(ctx context.Context, ws *workspacev1.Workspace)
 }
 

components/ws-manager-mk2/main.go

@@ -39,6 +39,7 @@ import (
 	"github.com/gitpod-io/gitpod/common-go/pprof"
 	regapi "github.com/gitpod-io/gitpod/registry-facade/api"
 	"github.com/gitpod-io/gitpod/ws-manager-mk2/controllers"
+	"github.com/gitpod-io/gitpod/ws-manager-mk2/pkg/activity"
 	"github.com/gitpod-io/gitpod/ws-manager-mk2/service"
 	wsmanapi "github.com/gitpod-io/gitpod/ws-manager/api"
 	config "github.com/gitpod-io/gitpod/ws-manager/api/config"
@@ -97,13 +98,14 @@ func main() {
 		os.Exit(1)
 	}
 
-	reconciler, err := controllers.NewWorkspaceReconciler(mgr.GetClient(), mgr.GetScheme(), cfg.Manager, metrics.Registry)
+	activity := &activity.WorkspaceActivity{}
+	reconciler, err := controllers.NewWorkspaceReconciler(mgr.GetClient(), mgr.GetScheme(), cfg.Manager, metrics.Registry, activity)
 	if err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Workspace")
 		os.Exit(1)
 	}
 
-	wsmanService, err := setupGRPCService(cfg, mgr.GetClient())
+	wsmanService, err := setupGRPCService(cfg, mgr.GetClient(), activity)
 	if err != nil {
 		setupLog.Error(err, "unable to start manager service")
 		os.Exit(1)
@@ -137,7 +139,7 @@ func main() {
 	}
 }
 
-func setupGRPCService(cfg *config.ServiceConfiguration, k8s client.Client) (*service.WorkspaceManagerServer, error) {
+func setupGRPCService(cfg *config.ServiceConfiguration, k8s client.Client, activity *activity.WorkspaceActivity) (*service.WorkspaceManagerServer, error) {
 	// TODO(cw): remove use of common-go/log
 
 	if len(cfg.RPCServer.RateLimits) > 0 {
@@ -170,7 +172,7 @@ func setupGRPCService(cfg *config.ServiceConfiguration, k8s client.Client) (*ser
 
 	grpcOpts = append(grpcOpts, grpc.UnknownServiceHandler(proxy.TransparentHandler(imagebuilderDirector(cfg.ImageBuilderProxy.TargetAddr))))
 
-	srv := service.NewWorkspaceManagerServer(k8s, &cfg.Manager, metrics.Registry)
+	srv := service.NewWorkspaceManagerServer(k8s, &cfg.Manager, metrics.Registry, activity)
 
 	grpcServer := grpc.NewServer(grpcOpts...)
 	grpc_prometheus.Register(grpcServer)

components/ws-manager-mk2/pkg/activity/activity.go

@@ -0,0 +1,29 @@
+// 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 activity
+
+import (
+	"sync"
+	"time"
+)
+
+// WorkspaceActivity is used to track the last user activity per workspace. This is
+// stored in memory instead of on the Workspace resource to limit load on the k8s API,
+// as this value will update often for each workspace.
+type WorkspaceActivity struct {
+	m sync.Map
+}
+
+func (w *WorkspaceActivity) Store(workspaceId string, lastActivity time.Time) {
+	w.m.Store(workspaceId, &lastActivity)
+}
+
+func (w *WorkspaceActivity) GetLastActivity(workspaceId string) *time.Time {
+	lastActivity, ok := w.m.Load(workspaceId)
+	if ok {
+		return lastActivity.(*time.Time)
+	}
+	return nil
+}

components/ws-manager-mk2/service/manager.go

@@ -24,6 +24,7 @@ import (
 	wsk8s "github.com/gitpod-io/gitpod/common-go/kubernetes"
 	"github.com/gitpod-io/gitpod/common-go/log"
 	"github.com/gitpod-io/gitpod/common-go/tracing"
+	"github.com/gitpod-io/gitpod/ws-manager-mk2/pkg/activity"
 	"github.com/gitpod-io/gitpod/ws-manager/api"
 	wsmanapi "github.com/gitpod-io/gitpod/ws-manager/api"
 	"github.com/gitpod-io/gitpod/ws-manager/api/config"
@@ -41,24 +42,26 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func NewWorkspaceManagerServer(clnt client.Client, cfg *config.Configuration, reg prometheus.Registerer) *WorkspaceManagerServer {
+func NewWorkspaceManagerServer(clnt client.Client, cfg *config.Configuration, reg prometheus.Registerer, activity *activity.WorkspaceActivity) *WorkspaceManagerServer {
 	metrics := newWorkspaceMetrics()
 	reg.MustRegister(metrics)
 
 	return &WorkspaceManagerServer{
-		Client:  clnt,
-		Config:  cfg,
-		metrics: metrics,
+		Client:   clnt,
+		Config:   cfg,
+		metrics:  metrics,
+		activity: activity,
 		subs: subscriptions{
 			subscribers: make(map[string]chan *wsmanapi.SubscribeResponse),
 		},
 	}
 }
 
 type WorkspaceManagerServer struct {
-	Client  client.Client
-	Config  *config.Configuration
-	metrics *workspaceMetrics
+	Client   client.Client
+	Config   *config.Configuration
+	metrics  *workspaceMetrics
+	activity *activity.WorkspaceActivity
 
 	subs subscriptions
 	wsmanapi.UnimplementedWorkspaceManagerServer
@@ -328,21 +331,10 @@ func (wsm *WorkspaceManagerServer) MarkActive(ctx context.Context, req *wsmanapi
 		return &api.MarkActiveResponse{}, nil
 	}
 
-	// We do not keep the last activity as annotation on the workspace to limit the load we're placing
+	// We do not keep the last activity in the workspace resource to limit the load we're placing
 	// on the K8S master in check. Thus, this state lives locally in a map.
-	// TODO: Check impact on k8s api, moving to CRD condition.
 	now := time.Now().UTC()
-	if err = wsm.modifyWorkspace(ctx, req.Id, true, func(ws *workspacev1.Workspace) error {
-		ws.Status.Conditions = addUniqueCondition(ws.Status.Conditions, metav1.Condition{
-			Type:               string(workspacev1.WorkspaceConditionUserActivity),
-			Status:             metav1.ConditionTrue,
-			LastTransitionTime: metav1.NewTime(now),
-		})
-		return nil
-	}); err != nil {
-		log.WithError(err).WithFields(log.OWI("", "", workspaceID)).Warn("was unable to set UserActivity condition on workspace")
-		return nil, err
-	}
+	wsm.activity.Store(req.Id, now)
 
 	// We do however maintain the the "closed" flag as annotation on the workspace. This flag should not change
 	// very often and provides a better UX if it persists across ws-manager restarts.
@@ -375,7 +367,7 @@ func (wsm *WorkspaceManagerServer) MarkActive(ctx context.Context, req *wsmanapi
 		log.WithError(err).WithFields(log.OWI("", "", workspaceID)).Warn("was unable to mark workspace properly")
 	}
 
-	// If it's the first call: Mark the pod with firstUserActivityAnnotation
+	// If it's the first call: Mark the pod with FirstUserActivity condition.
 	if firstUserActivity == nil {
 		err := wsm.modifyWorkspace(ctx, req.Id, true, func(ws *workspacev1.Workspace) error {
 			ws.Status.Conditions = append(ws.Status.Conditions, metav1.Condition{