Make generic controller for cluster-based resources

[andrewstucki]

This tries to extract the core of the Users controller into a generic controller implementation that wraps a ResourceReconciler interface that controls the guts of the three operations that we pretty much need to do for every cluster-based resource:

1. Patching finalizers (this could probably be removed if we want to do it in an even more generic way)
2. Syncing (upserting) a resource
3. Deleting a resource

Hence the interface looks like this:

	FinalizerPatch(request ResourceRequest[T]) client.Patch
	SyncResource(ctx context.Context, request ResourceRequest[T]) (client.Patch, error)
	DeleteResource(ctx context.Context, request ResourceRequest[T]) error

Most of the sync/delete logic can either then be inlined into the controllers, or, ideally wrapped in a "high-level" client/synchronizer (see the ACL synchronization code) in the clients package.

I'm intending on using this for CRD-defined schemas, and if y'all like the structure, would want to take a pass at refactoring TopicReconciler to use this as well for consistency.

This allows us to follow a really simple pattern for any additional CRDs. Basically anything that we're controlling via CRD and putting in a cluster should:

1. Add a ClusterSource field.
2. Generate some applyconfiguration structures
3. If it's complex, implement a high-level synchronizer that handles the heavy lifting of the Redpanda operations
4. Add a reconciler that implements the above interface

Doing so would give you:

1. Consistent status conditions
2. Consistent ways of connecting to a referenced cluster
3. Server-side apply pretty much everywhere
4. Similarly structured code patterns

[RafalKorepta]

I didn't finish review.

[RafalKorepta]

Would terminal client error prevent from further reconciliation? What action should be taken by the user of the controller/custom resource? Could the error constant have brief description or explanation when they should be used?

Maybe I'm overexaggerate over the word Terminal.

[andrewstucki]

Maybe, I use the notion of a "terminal" error in two places currently, which is how the UserReconciler works:

1. redpanda-operator/operator/internal/controller/redpanda/resource_controller.go

   Lines 122 to 137 in f61b187

   	func ignoreAllConnectionErrors(logger logr.Logger, err error) error {
   	// If we have known errors where we're unable to actually establish
   	// a connection to the cluster due to say, invalid connection parameters
   	// we're going to just skip the cleanup phase since we likely won't be
   	// able to clean ourselves up anyway.
   	if internalclient.IsTerminalClientError(err) ||
   	internalclient.IsConfigurationError(err) ||
   	internalclient.IsInvalidClusterError(err) {
   	// We use Info rather than Error here because we don't want
   	// to ignore the verbosity settings. This is really only for
   	// debugging purposes.
   	logger.V(2).Info("Ignoring non-retryable client error", "error", err)
   	return nil
   	}
   	return err
   	}

2. redpanda-operator/operator/internal/controller/redpanda/resource_controller.go

   Lines 139 to 153 in f61b187

   	func handleResourceSyncErrors(err error) (metav1.Condition, error) {
   	// If we have a known terminal error, just set the sync condition and don't re-run reconciliation.
   	if internalclient.IsInvalidClusterError(err) {
   	return redpandav1alpha2.ResourceNotSyncedCondition(redpandav1alpha2.ResourceConditionReasonClusterRefInvalid, err), nil
   	}
   	if internalclient.IsConfigurationError(err) {
   	return redpandav1alpha2.ResourceNotSyncedCondition(redpandav1alpha2.ResourceConditionReasonConfigurationInvalid, err), nil
   	}
   	if internalclient.IsTerminalClientError(err) {
   	return redpandav1alpha2.ResourceNotSyncedCondition(redpandav1alpha2.ResourceConditionReasonTerminalClientError, err), nil
   	}
   	// otherwise, set a generic unexpected error and return an error so we can re-reconcile.
   	return redpandav1alpha2.ResourceNotSyncedCondition(redpandav1alpha2.ResourceConditionReasonUnexpectedError, err), err
   	}

The first is basically if you want to ignore errors that shouldn't ever be retried (i.e. you have a non-existent cluster that's been blown away and you're trying to delete yourself from it, cluster configuration now has SASL auth disabled and you're trying to delete a user, etc.). This is used during the cleanup routine to do "best-effort" style cleanup, if something is "terminal" we basically just say, "oh well" and let the resource be GC'd if we hit one of these.

The second is in the case of you trying to sync a resource. In that case we just set the status as having encountered a terminal error with the error message in the status condition message and then stop reconciling because this isn't an "ephemeral"-type error.

[andrewstucki]

Also 👍 on documenting this stuff more.

[RafalKorepta]

NIT: For me the o and l is hard to read. Maybe the following suggestion could make it more readable 👇

Suggested change

	func registerClusterSourceIndex[T client.Object, U clientList[T]](ctx context.Context, mgr ctrl.Manager, name string, o T, l U) (handler.EventHandler, error) {
	func registerClusterSourceIndex[T client.Object, U clientList[T]](ctx context.Context, mgr ctrl.Manager, name string, obj T, list U) (handler.EventHandler, error) {

[RafalKorepta]

Not related to this PR

	role := labels["app.kubernetes.io/name"]
	if !slices.Contains([]string{"redpanda", "console"}, role) {
		return types.NamespacedName{}, false
	}

Should include connectors too.

[RafalKorepta]

NIT: Maybe make the period configurable.

[RafalKorepta]

LGTM

[RafalKorepta]

NIT: As a follow up we should bump this version in all places in our test suite

[andrewstucki]

Going to go ahead and merge and then I'll address some of the documentation in a follow-up.