TopologyReconciler

api/v1beta1/binding_types.go

@@ -81,6 +81,14 @@ func (b *Binding) GroupResource() schema.GroupResource {
 	}
 }
 
+func (b *Binding) RabbitReference() RabbitmqClusterReference {
+	return b.Spec.RabbitmqClusterReference
+}
+
+func (b *Binding) SetStatusConditions(c []Condition) {
+	b.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Binding{}, &BindingList{})
 }

api/v1beta1/exchange_types.go

@@ -77,6 +77,14 @@ func (e *Exchange) GroupResource() schema.GroupResource {
 	}
 }
 
+func (e *Exchange) RabbitReference() RabbitmqClusterReference {
+	return e.Spec.RabbitmqClusterReference
+}
+
+func (e *Exchange) SetStatusConditions(c []Condition) {
+	e.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Exchange{}, &ExchangeList{})
 }

api/v1beta1/federation_types.go

@@ -75,6 +75,14 @@ func (f *Federation) GroupResource() schema.GroupResource {
 	}
 }
 
+func (f *Federation) RabbitReference() RabbitmqClusterReference {
+	return f.Spec.RabbitmqClusterReference
+}
+
+func (f *Federation) SetStatusConditions(c []Condition) {
+	f.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Federation{}, &FederationList{})
 }

api/v1beta1/permission_types.go

@@ -74,6 +74,14 @@ func (p *Permission) GroupResource() schema.GroupResource {
 	}
 }
 
+func (p *Permission) RabbitReference() RabbitmqClusterReference {
+	return p.Spec.RabbitmqClusterReference
+}
+
+func (p *Permission) SetStatusConditions(c []Condition) {
+	p.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Permission{}, &PermissionList{})
 }

api/v1beta1/policy_types.go

@@ -77,6 +77,14 @@ func (p *Policy) GroupResource() schema.GroupResource {
 	}
 }
 
+func (p *Policy) RabbitReference() RabbitmqClusterReference {
+	return p.Spec.RabbitmqClusterReference
+}
+
+func (p *Policy) SetStatusConditions(c []Condition) {
+	p.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Policy{}, &PolicyList{})
 }

api/v1beta1/queue_types.go

@@ -81,6 +81,14 @@ func (q *Queue) GroupResource() schema.GroupResource {
 	}
 }
 
+func (q *Queue) RabbitReference() RabbitmqClusterReference {
+	return q.Spec.RabbitmqClusterReference
+}
+
+func (q *Queue) SetStatusConditions(c []Condition) {
+	q.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Queue{}, &QueueList{})
 }

api/v1beta1/schemareplication_types.go

@@ -80,13 +80,21 @@ type SchemaReplicationList struct {
 	Items           []SchemaReplication `json:"items"`
 }
 
-func init() {
-	SchemeBuilder.Register(&SchemaReplication{}, &SchemaReplicationList{})
-}
-
 func (s *SchemaReplication) GroupResource() schema.GroupResource {
 	return schema.GroupResource{
 		Group:    s.GroupVersionKind().Group,
 		Resource: s.GroupVersionKind().Kind,
 	}
 }
+
+func (s *SchemaReplication) RabbitReference() RabbitmqClusterReference {
+	return s.Spec.RabbitmqClusterReference
+}
+
+func (s *SchemaReplication) SetStatusConditions(c []Condition) {
+	s.Status.Conditions = c
+}
+
+func init() {
+	SchemeBuilder.Register(&SchemaReplication{}, &SchemaReplicationList{})
+}

api/v1beta1/shovel_types.go

@@ -88,6 +88,14 @@ func (s *Shovel) GroupResource() schema.GroupResource {
 	}
 }
 
+func (s *Shovel) RabbitReference() RabbitmqClusterReference {
+	return s.Spec.RabbitmqClusterReference
+}
+
+func (s *Shovel) SetStatusConditions(c []Condition) {
+	s.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Shovel{}, &ShovelList{})
 }

api/v1beta1/topology_resource.go

@@ -0,0 +1,20 @@
+package v1beta1
+
+import "sigs.k8s.io/controller-runtime/pkg/client"
+
+var _ TopologyResource = &Binding{}
+var _ TopologyResource = &Exchange{}
+var _ TopologyResource = &Federation{}
+var _ TopologyResource = &Permission{}
+var _ TopologyResource = &Queue{}
+var _ TopologyResource = &SchemaReplication{}
+var _ TopologyResource = &Shovel{}
+var _ TopologyResource = &User{}
+var _ TopologyResource = &Vhost{}
+
+// +k8s:deepcopy-gen=false
+type TopologyResource interface {
+	client.Object
+	RabbitReference() RabbitmqClusterReference
+	SetStatusConditions([]Condition)
+}

api/v1beta1/user_types.go

@@ -86,6 +86,14 @@ func (u *User) GroupResource() schema.GroupResource {
 	}
 }
 
+func (u *User) RabbitReference() RabbitmqClusterReference {
+	return u.Spec.RabbitmqClusterReference
+}
+
+func (u *User) SetStatusConditions(c []Condition) {
+	u.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&User{}, &UserList{})
 }

api/v1beta1/vhost_types.go

@@ -65,6 +65,14 @@ func (v *Vhost) GroupResource() schema.GroupResource {
 	}
 }
 
+func (v *Vhost) RabbitReference() RabbitmqClusterReference {
+	return v.Spec.RabbitmqClusterReference
+}
+
+func (v *Vhost) SetStatusConditions(c []Condition) {
+	v.Status.Conditions = c
+}
+
 func init() {
 	SchemeBuilder.Register(&Vhost{}, &VhostList{})
 }

controllers/binding_controller.go

@@ -13,133 +13,39 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"reflect"
 
+	"github.com/go-logr/logr"
 	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
+	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 	"github.com/rabbitmq/messaging-topology-operator/internal"
 	"github.com/rabbitmq/messaging-topology-operator/rabbitmqclient"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/client-go/tools/record"
-	clientretry "k8s.io/client-go/util/retry"
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-
-	"github.com/go-logr/logr"
-	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	topology "github.com/rabbitmq/messaging-topology-operator/api/v1beta1"
 )
 
-// BindingReconciler reconciles a Binding object
-type BindingReconciler struct {
-	client.Client
-	Log                     logr.Logger
-	Scheme                  *runtime.Scheme
-	Recorder                record.EventRecorder
-	RabbitmqClientFactory   rabbitmqclient.Factory
-	KubernetesClusterDomain string
-}
-
 // +kubebuilder:rbac:groups=rabbitmq.com,resources=bindings,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=rabbitmq.com,resources=bindings/finalizers,verbs=update
 // +kubebuilder:rbac:groups=rabbitmq.com,resources=bindings/status,verbs=get;update;patch
 
-func (r *BindingReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	logger := ctrl.LoggerFrom(ctx)
-
-	binding := &topology.Binding{}
-	if err := r.Get(ctx, req.NamespacedName, binding); err != nil {
-		return reconcile.Result{}, client.IgnoreNotFound(err)
-	}
-
-	systemCertPool, err := extractSystemCertPool(ctx, r.Recorder, binding)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
-
-	credsProvider, tlsEnabled, err := rabbitmqclient.ParseReference(ctx, r.Client, binding.Spec.RabbitmqClusterReference, binding.Namespace, r.KubernetesClusterDomain)
-	if err != nil {
-		return handleRMQReferenceParseError(ctx, r.Client, r.Recorder, binding, &binding.Status.Conditions, err)
-	}
-
-	rabbitClient, err := r.RabbitmqClientFactory(credsProvider, tlsEnabled, systemCertPool)
-	if err != nil {
-		logger.Error(err, failedGenerateRabbitClient)
-		return reconcile.Result{}, err
-	}
-
-	if !binding.ObjectMeta.DeletionTimestamp.IsZero() {
-		logger.Info("Deleting")
-		return ctrl.Result{}, r.deleteBinding(ctx, rabbitClient, binding)
-	}
-
-	if err := addFinalizerIfNeeded(ctx, r.Client, binding); err != nil {
-		return ctrl.Result{}, err
-	}
-	spec, err := json.Marshal(binding.Spec)
-	if err != nil {
-		logger.Error(err, failedMarshalSpec)
-	}
-
-	logger.Info("Start reconciling",
-		"spec", string(spec))
-
-	if err := r.declareBinding(ctx, rabbitClient, binding); err != nil {
-		// Set Condition 'Ready' to false with message
-		binding.Status.Conditions = []topology.Condition{
-			topology.NotReady(err.Error(), binding.Status.Conditions),
-		}
-		if writerErr := clientretry.RetryOnConflict(clientretry.DefaultRetry, func() error {
-			return r.Status().Update(ctx, binding)
-		}); writerErr != nil {
-			logger.Error(writerErr, failedStatusUpdate, "status", binding.Status)
-		}
-		return ctrl.Result{}, err
-	}
-
-	binding.Status.Conditions = []topology.Condition{topology.Ready(binding.Status.Conditions)}
-	binding.Status.ObservedGeneration = binding.GetGeneration()
-	if writerErr := clientretry.RetryOnConflict(clientretry.DefaultRetry, func() error {
-		return r.Status().Update(ctx, binding)
-	}); writerErr != nil {
-		logger.Error(writerErr, failedStatusUpdate, "status", binding.Status)
-	}
-	logger.Info("Finished reconciling")
-
-	return ctrl.Result{}, nil
-}
-
-func (r *BindingReconciler) declareBinding(ctx context.Context, client rabbitmqclient.Client, binding *topology.Binding) error {
-	logger := ctrl.LoggerFrom(ctx)
+type BindingReconciler struct{}
 
+func (r *BindingReconciler) DeclareFunc(ctx context.Context, client rabbitmqclient.Client, obj topology.TopologyResource) error {
+	binding := obj.(*topology.Binding)
 	info, err := internal.GenerateBindingInfo(binding)
 	if err != nil {
-		msg := "failed to generate binding info"
-		r.Recorder.Event(binding, corev1.EventTypeWarning, "FailedDeclare", msg)
-		logger.Error(err, msg)
-		return err
+		return fmt.Errorf("failed to generate binding info: %w", err)
 	}
-
-	if err := validateResponse(client.DeclareBinding(binding.Spec.Vhost, *info)); err != nil {
-		msg := "failed to declare binding"
-		r.Recorder.Event(binding, corev1.EventTypeWarning, "FailedDeclare", msg)
-		logger.Error(err, msg)
-		return err
-	}
-
-	logger.Info("Successfully declared binding")
-	r.Recorder.Event(binding, corev1.EventTypeNormal, "SuccessfulDeclare", "Successfully declared binding")
-	return nil
+	return validateResponse(client.DeclareBinding(binding.Spec.Vhost, *info))
 }
 
 // deletes binding from rabbitmq server; bindings have no name; server needs BindingInfo to delete them
 // when server responds with '404' Not Found, it logs and does not requeue on error
 // if no binding argument is set, generating properties key by using internal.GeneratePropertiesKey
 // if binding arguments are set, list all bindings between source/destination to find the binding; if it failed to find corresponding binding, it assumes that the binding is already deleted and returns no error
-func (r *BindingReconciler) deleteBinding(ctx context.Context, client rabbitmqclient.Client, binding *topology.Binding) error {
+func (r *BindingReconciler) DeleteFunc(ctx context.Context, client rabbitmqclient.Client, obj topology.TopologyResource) error {
 	logger := ctrl.LoggerFrom(ctx)
-
+	binding := obj.(*topology.Binding)
 	var info *rabbithole.BindingInfo
 	var err error
 	if binding.Spec.Arguments != nil {
@@ -149,15 +55,12 @@ func (r *BindingReconciler) deleteBinding(ctx context.Context, client rabbitmqcl
 		}
 		if info == nil {
 			logger.Info("cannot find the corresponding binding info in rabbitmq server; binding already deleted")
-			return removeFinalizer(ctx, r.Client, binding)
+			return nil
 		}
 	} else {
 		info, err = internal.GenerateBindingInfo(binding)
 		if err != nil {
-			msg := "failed to generate binding info"
-			r.Recorder.Event(binding, corev1.EventTypeWarning, "FailedDelete", msg)
-			logger.Error(err, msg)
-			return err
+			return fmt.Errorf("failed to generate binding info: %w", err)
 		}
 		info.PropertiesKey = internal.GeneratePropertiesKey(binding)
 	}
@@ -166,25 +69,18 @@ func (r *BindingReconciler) deleteBinding(ctx context.Context, client rabbitmqcl
 	if errors.Is(err, NotFound) {
 		logger.Info("cannot find binding in rabbitmq server; already deleted")
 	} else if err != nil {
-		msg := "failed to delete binding"
-		r.Recorder.Event(binding, corev1.EventTypeWarning, "FailedDelete", msg)
-		logger.Error(err, msg)
 		return err
 	}
 
-	logger.Info("Successfully deleted binding")
-	r.Recorder.Event(binding, corev1.EventTypeNormal, "SuccessfulDelete", "successfully deleted binding")
-	return removeFinalizer(ctx, r.Client, binding)
+	return nil
 }
 
 func (r *BindingReconciler) findBindingInfo(logger logr.Logger, binding *topology.Binding, client rabbitmqclient.Client) (*rabbithole.BindingInfo, error) {
 	logger.Info("binding arguments set; listing bindings from server to complete deletion")
 	arguments := make(map[string]interface{})
 	if binding.Spec.Arguments != nil {
 		if err := json.Unmarshal(binding.Spec.Arguments.Raw, &arguments); err != nil {
-			msg := "failed to unmarshall binding arguments"
-			r.Recorder.Event(binding, corev1.EventTypeWarning, "FailedDelete", msg)
-			logger.Error(err, msg)
+			logger.Error(err, "failed to unmarshall binding arguments")
 			return nil, err
 		}
 	}
@@ -196,9 +92,7 @@ func (r *BindingReconciler) findBindingInfo(logger logr.Logger, binding *topolog
 		bindingInfos, err = client.ListExchangeBindingsBetween(binding.Spec.Vhost, binding.Spec.Source, binding.Spec.Destination)
 	}
 	if err != nil {
-		msg := "failed to list binding infos"
-		r.Recorder.Event(binding, corev1.EventTypeWarning, "FailedDelete", msg)
-		logger.Error(err, msg)
+		logger.Error(err, "failed to list binding infos")
 		return nil, err
 	}
 	var info *rabbithole.BindingInfo
@@ -209,13 +103,3 @@ func (r *BindingReconciler) findBindingInfo(logger logr.Logger, binding *topolog
 	}
 	return info, nil
 }
-
-func (r *BindingReconciler) SetInternalDomainName(clusterDomain string) {
-	r.KubernetesClusterDomain = clusterDomain
-}
-
-func (r *BindingReconciler) SetupWithManager(mgr ctrl.Manager) error {
-	return ctrl.NewControllerManagedBy(mgr).
-		For(&topology.Binding{}).
-		Complete(r)
-}