Bount the source of subscribers to an informer

Author: aledbf

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

@@ -0,0 +1,94 @@
+// Copyright (c) 2022 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 controllers
+
+import (
+	"context"
+	"os"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+
+	config "github.com/gitpod-io/gitpod/ws-manager/api/config"
+	workspacev1 "github.com/gitpod-io/gitpod/ws-manager/api/crd/v1"
+)
+
+func NewSubscriberReconciler(c client.Client, scheme *runtime.Scheme, recorder record.EventRecorder, cfg *config.Configuration) (*SubscriberReconciler, error) {
+	reconciler := &SubscriberReconciler{
+		Client:   c,
+		Scheme:   scheme,
+		Config:   cfg,
+		Recorder: recorder,
+	}
+
+	return reconciler, nil
+}
+
+type SubscriberReconciler struct {
+	client.Client
+	Scheme *runtime.Scheme
+
+	Config      *config.Configuration
+	Recorder    record.EventRecorder
+	OnReconcile func(ctx context.Context, ws *workspacev1.Workspace)
+}
+
+// Reconcile is part of the main kubernetes reconciliation loop which aims to
+// move the current state of the cluster closer to the desired state.
+// Modify the Reconcile function to compare the state specified by
+// the Workspace object against the actual cluster state, and then
+// perform operations to make the cluster state reflect the state specified by
+// the user.
+//
+// For more details, check Reconcile and its Result here:
+// - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/reconcile
+func (r *SubscriberReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := log.FromContext(ctx)
+
+	var workspace workspacev1.Workspace
+	if err := r.Get(ctx, req.NamespacedName, &workspace); err != nil {
+		if !errors.IsNotFound(err) {
+			log.Error(err, "unable to fetch workspace")
+		}
+
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+
+	if r.OnReconcile != nil {
+		// Publish to subscribers in a goroutine, to prevent blocking the main reconcile loop.
+		ws := workspace.DeepCopy()
+		go func() {
+			r.OnReconcile(ctx, ws)
+		}()
+	}
+
+	return reconcile.Result{}, nil
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *SubscriberReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
+	c, err := controller.NewUnmanaged("maintenance-controller", mgr, controller.Options{Reconciler: r})
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		err = c.Start(ctx)
+		if err != nil {
+			log.FromContext(ctx).Error(err, "cannot start maintenance reconciler")
+			os.Exit(1)
+		}
+	}()
+
+	return c.Watch(source.Kind(mgr.GetCache(), &workspacev1.Workspace{}), &handler.EnqueueRequestForObject{})
+}

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

@@ -71,7 +71,6 @@ type WorkspaceReconciler struct {
 	metrics     *controllerMetrics
 	maintenance maintenance.Maintenance
 	Recorder    record.EventRecorder
-	OnReconcile func(ctx context.Context, ws *workspacev1.Workspace)
 }
 
 //+kubebuilder:rbac:groups=workspace.gitpod.io,resources=workspaces,verbs=get;list;watch;create;update;patch;delete
@@ -107,14 +106,6 @@ func (r *WorkspaceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		workspace.Status.Conditions = []metav1.Condition{}
 	}
 
-	if r.OnReconcile != nil {
-		// Publish to subscribers in a goroutine, to prevent blocking the main reconcile loop.
-		ws := workspace.DeepCopy()
-		go func() {
-			r.OnReconcile(ctx, ws)
-		}()
-	}
-
 	log.Info("reconciling workspace", "workspace", req.NamespacedName, "phase", workspace.Status.Phase)
 
 	var workspacePods corev1.PodList

components/ws-manager-mk2/main.go

@@ -6,14 +6,11 @@ package main
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"flag"
 	"fmt"
 	"net"
 	"os"
-	"sync/atomic"
-	"time"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
@@ -61,8 +58,6 @@ var (
 
 	scheme   = runtime.NewScheme()
 	setupLog = ctrl.Log.WithName("setup")
-
-	LeaderInstance atomic.Bool
 )
 
 func init() {
@@ -142,11 +137,6 @@ func main() {
 
 	mgrCtx := ctrl.SetupSignalHandler()
 
-	go func() {
-		<-mgr.Elected()
-		LeaderInstance.Store(true)
-	}()
-
 	maintenanceReconciler, err := controllers.NewMaintenanceReconciler(mgr.GetClient())
 	if err != nil {
 		setupLog.Error(err, "unable to create maintenance controller", "controller", "Maintenance")
@@ -160,19 +150,6 @@ func main() {
 		os.Exit(1)
 	}
 
-	go func() {
-		for {
-			select {
-			case <-mgrCtx.Done():
-				return
-			case <-mgr.Elected():
-				now := time.Now()
-				setupLog.Info("updating activity started time", "now", now)
-				return
-			}
-		}
-	}()
-
 	timeoutReconciler, err := controllers.NewTimeoutReconciler(mgr.GetClient(), mgr.GetEventRecorderFor("workspace"), cfg.Manager, maintenanceReconciler)
 	if err != nil {
 		setupLog.Error(err, "unable to create timeout controller", "controller", "Timeout")
@@ -185,7 +162,18 @@ func main() {
 		os.Exit(1)
 	}
 
-	workspaceReconciler.OnReconcile = wsmanService.OnWorkspaceReconcile
+	subscriberReconciler, err := controllers.NewSubscriberReconciler(mgr.GetClient(), mgr.GetScheme(), mgr.GetEventRecorderFor("subscribers"), &cfg.Manager)
+	if err != nil {
+		setupLog.Error(err, "unable to create timeout controller", "controller", "Timeout")
+		os.Exit(1)
+	}
+
+	subscriberReconciler.OnReconcile = wsmanService.OnWorkspaceReconcile
+
+	if err = subscriberReconciler.SetupWithManager(mgrCtx, mgr); err != nil {
+		setupLog.Error(err, "unable to setup workspace controller with manager", "controller", "Subscribers")
+		os.Exit(1)
+	}
 
 	if err = workspaceReconciler.SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to setup workspace controller with manager", "controller", "Workspace")
@@ -238,27 +226,8 @@ func setupGRPCService(cfg *config.ServiceConfiguration, k8s client.Client, maint
 	metrics.Registry.MustRegister(grpcMetrics)
 
 	grpcOpts := common_grpc.ServerOptionsWithInterceptors(
-		[]grpc.StreamServerInterceptor{
-			grpcMetrics.StreamServerInterceptor(),
-			func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
-				if LeaderInstance.Load() {
-					return handler(srv, ss)
-				}
-
-				return fmt.Errorf("Rejecting connection due leader election")
-			},
-		},
-		[]grpc.UnaryServerInterceptor{
-			grpcMetrics.UnaryServerInterceptor(),
-			ratelimits.UnaryInterceptor(),
-			func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
-				if LeaderInstance.Load() {
-					return handler(ctx, req)
-				}
-
-				return nil, fmt.Errorf("Rejecting connection due leader election")
-			},
-		},
+		[]grpc.StreamServerInterceptor{grpcMetrics.StreamServerInterceptor()},
+		[]grpc.UnaryServerInterceptor{grpcMetrics.UnaryServerInterceptor(), ratelimits.UnaryInterceptor()},
 	)
 	if cfg.RPCServer.TLS.CA != "" && cfg.RPCServer.TLS.Certificate != "" && cfg.RPCServer.TLS.PrivateKey != "" {
 		tlsConfig, err := common_grpc.ClientAuthTLSConfig(