Filter subscriber reconciler events (v2) (#18696)

* Filter subscriber reconciler events (v2)

* Ignore LastActivity

Author: aledbf

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

@@ -150,6 +150,22 @@ type PortSpec struct {
 	Protocol PortProtocol `json:"protocol"`
 }
 
+func (ps PortSpec) Equal(other PortSpec) bool {
+	if ps.Port != other.Port {
+		return false
+	}
+
+	if ps.Visibility != other.Visibility {
+		return false
+	}
+
+	if ps.Protocol != other.Protocol {
+		return false
+	}
+
+	return true
+}
+
 // WorkspaceStatus defines the observed state of Workspace
 type WorkspaceStatus struct {
 	PodStarts  int    `json:"podStarts"`

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

@@ -8,13 +8,17 @@ import (
 	"context"
 	"os"
 
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	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/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	config "github.com/gitpod-io/gitpod/ws-manager/api/config"
@@ -76,5 +80,24 @@ func (r *SubscriberReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Ma
 		}
 	}()
 
-	return c.Watch(source.Kind(mgr.GetCache(), &workspacev1.Workspace{}), &handler.EnqueueRequestForObject{})
+	// we need several reconciliation loops during a workspace creation until it reaches a stable state.
+	// this introduces the side effect of multiple notifications to the subscribers with partial information.
+	// the filterByUpdate predicate acts as a filter to avoid this
+	filterByUpdate := predicate.Funcs{
+		CreateFunc: func(ce event.CreateEvent) bool {
+			return true
+		},
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			old := e.ObjectOld.(*workspacev1.Workspace)
+			new := e.ObjectNew.(*workspacev1.Workspace)
+
+			if !cmp.Equal(old.Spec.Ports, new.Spec.Ports) {
+				return true
+			}
+
+			return !cmp.Equal(old.Status, new.Status, cmpopts.IgnoreFields(workspacev1.WorkspaceStatus{}, "LastActivity"))
+		},
+	}
+
+	return c.Watch(source.Kind(mgr.GetCache(), &workspacev1.Workspace{}), &handler.EnqueueRequestForObject{}, filterByUpdate)
 }