controller.go

// Copyright 2021 - 2024 Crunchy Data Solutions, Inc.
//
// SPDX-License-Identifier: Apache-2.0

package postgrescluster

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

	cmv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
	"github.com/pkg/errors"
	"go.opentelemetry.io/otel/trace"
	appsv1 "k8s.io/api/apps/v1"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	policyv1 "k8s.io/api/policy/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	k8sruntime "k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/validation/field"
	"k8s.io/client-go/discovery"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/record"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	"sigs.k8s.io/controller-runtime/pkg/source"

	"github.com/percona/percona-postgresql-operator/v3/internal/config"
	"github.com/percona/percona-postgresql-operator/v3/internal/controller/runtime"
	pgbruntime "github.com/percona/percona-postgresql-operator/v3/internal/controller/runtime/pgbouncer"
	"github.com/percona/percona-postgresql-operator/v3/internal/initialize"
	"github.com/percona/percona-postgresql-operator/v3/internal/logging"
	"github.com/percona/percona-postgresql-operator/v3/internal/logicalreplica"
	"github.com/percona/percona-postgresql-operator/v3/internal/naming"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgaudit"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgbackrest"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgbouncer"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgcron"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgmonitor"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgstatmonitor"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgstatstatements"
	"github.com/percona/percona-postgresql-operator/v3/internal/pgtde"
	"github.com/percona/percona-postgresql-operator/v3/internal/pki"
	"github.com/percona/percona-postgresql-operator/v3/internal/pmm"
	"github.com/percona/percona-postgresql-operator/v3/internal/postgres"
	"github.com/percona/percona-postgresql-operator/v3/internal/registration"
	"github.com/percona/percona-postgresql-operator/v3/internal/setuser"
	"github.com/percona/percona-postgresql-operator/v3/percona/certmanager"
	"github.com/percona/percona-postgresql-operator/v3/percona/k8s"
	"github.com/percona/percona-postgresql-operator/v3/pkg/apis/upstream.pgv2.percona.com/v1beta1"
)

const (
	// ControllerName is the name of the PostgresCluster controller
	ControllerName = "postgrescluster-controller"
)

// Reconciler holds resources for the PostgresCluster reconciler
type Reconciler struct {
	Client client.Client
	// K8SPG-992: APIReader reads directly from the API server, bypassing the cache
	APIReader                    client.Reader
	Scheme                       *k8sruntime.Scheme
	DiscoveryClient              *discovery.DiscoveryClient
	IsOpenShift                  bool
	Owner                        client.FieldOwner
	PodExec                      runtime.PodExecutor
	Recorder                     record.EventRecorder
	Registration                 registration.Registration
	Tracer                       trace.Tracer
	CertManagerCtrlFunc          certmanager.NewControllerFunc
	RestConfig                   *rest.Config
	Controller                   controller.Controller
	Cache                        cache.Cache
	certManagerWatchesRegistered atomic.Bool
	newPGBouncerAdmin            func(opts pgbruntime.AdminClientOptions) (pgbruntime.AdminClient, error)
}

func (r *Reconciler) apiReader() client.Reader {
	if r.APIReader != nil {
		return r.APIReader
	}
	return r.Client
}

// +kubebuilder:rbac:groups="",resources="events",verbs={create,patch}
// +kubebuilder:rbac:groups="upstream.pgv2.percona.com",resources="postgresclusters",verbs={get,list,watch}
// +kubebuilder:rbac:groups="upstream.pgv2.percona.com",resources="postgresclusters/status",verbs={patch}

// Reconcile reconciles a ConfigMap in a namespace managed by the PostgreSQL Operator
func (r *Reconciler) Reconcile(
	ctx context.Context, request reconcile.Request) (reconcile.Result, error,
) {
	ctx, span := r.Tracer.Start(ctx, "Reconcile")
	log := logging.FromContext(ctx)
	defer span.End()

	// get the postgrescluster from the cache
	cluster := &v1beta1.PostgresCluster{}
	if err := r.Client.Get(ctx, request.NamespacedName, cluster); err != nil {
		// NotFound cannot be fixed by requeuing so ignore it. During background
		// deletion, we receive delete events from cluster's dependents after
		// cluster is deleted.
		if err = client.IgnoreNotFound(err); err != nil {
			log.Error(err, "unable to fetch PostgresCluster")
			span.RecordError(err)
		}
		return runtime.ErrorWithBackoff(err)
	}

	// Set any defaults that may not have been stored in the API. No DeepCopy
	// is necessary because controller-runtime makes a copy before returning
	// from its cache.
	_ = cluster.Default(ctx, nil)

	if cluster.Spec.OpenShift == nil {
		cluster.Spec.OpenShift = &r.IsOpenShift
	}

	// Keep a copy of cluster prior to any manipulations.
	before := cluster.DeepCopy()

	// NOTE(cbandy): When a namespace is deleted, objects owned by a
	// PostgresCluster may be deleted before the PostgresCluster is deleted.
	// When this happens, any attempt to reconcile those objects is rejected
	// as Forbidden: "unable to create new content in namespace … because it is
	// being terminated".

	// Check for and handle deletion of cluster. Return early if it is being
	// deleted or there was an error.
	if result, err := r.handleDelete(ctx, cluster); err != nil {
		span.RecordError(err)
		log.Error(err, "deleting")
		return runtime.ErrorWithBackoff(err)

	} else if result != nil {
		if log := log.V(1); log.Enabled() {
			log.Info("deleting", "result", fmt.Sprintf("%+v", *result))
		}
		return *result, nil
	}

	// Perform initial validation on a cluster
	// TODO: Move this to a defaulting (mutating admission) webhook
	// to leverage regular validation.

	// verify all needed image values are defined
	if err := config.VerifyImageValues(cluster); err != nil {
		// warning event with missing image information
		r.Recorder.Event(cluster, corev1.EventTypeWarning, "MissingRequiredImage",
			err.Error())
		// specifically allow reconciliation if the cluster is shutdown to
		// facilitate upgrades, otherwise return
		if !initialize.FromPointer(cluster.Spec.Shutdown) {
			return runtime.ErrorWithBackoff(err)
		}
	}
	// Issue Warning Event if postgres version is EOL according to PostgreSQL:
	// https://www.postgresql.org/support/versioning/
	currentTime := time.Now()
	if postgres.ReleaseIsFinal(cluster.Spec.PostgresVersion, currentTime) {
		r.Recorder.Eventf(cluster, corev1.EventTypeWarning, "EndOfLifePostgresVersion",
			"The last minor version of Postgres %[1]v has been released."+
				" PG %[1]v will no longer receive updates. We recommend upgrading."+
				" See https://www.postgresql.org/support/versioning",
			cluster.Spec.PostgresVersion)
	}

	if cluster.Spec.Standby != nil &&
		cluster.Spec.Standby.Enabled &&
		cluster.Spec.Standby.Host == "" &&
		cluster.Spec.Standby.RepoName == "" {
		// When a standby cluster is requested but a repoName or host is not provided
		// the cluster will be created as a non-standby. Reject any clusters with
		// this configuration and provide an event
		path := field.NewPath("spec", "standby")
		err := field.Invalid(path, cluster.Name, "Standby requires a host or repoName to be enabled")
		r.Recorder.Event(cluster, corev1.EventTypeWarning, "InvalidStandbyConfiguration", err.Error())
		return runtime.ErrorWithBackoff(err)
	}

	var (
		clusterConfigMap             *corev1.ConfigMap
		clusterReplicationSecret     *corev1.Secret
		clusterPodService            *corev1.Service
		clusterVolumes               []corev1.PersistentVolumeClaim
		instanceServiceAccount       *corev1.ServiceAccount
		instances                    *observedInstances
		patroniLeaderService         *corev1.Service
		primaryCertificate           *corev1.SecretProjection
		caBundle                     *corev1.SecretProjection
		primaryService               *corev1.Service
		replicaService               *corev1.Service
		rootCA                       *pki.RootCertificateAuthority
		monitoringSecret             *corev1.Secret
		exporterQueriesConfig        *corev1.ConfigMap
		exporterWebConfig            *corev1.ConfigMap
		err                          error
		backupsSpecFound             bool
		backupsReconciliationAllowed bool
		dedicatedSnapshotPVC         *corev1.PersistentVolumeClaim
	)

	patchClusterStatus := func() error {
		if !equality.Semantic.DeepEqual(before.Status, cluster.Status) {
			// NOTE(cbandy): Kubernetes prior to v1.16.10 and v1.17.6 does not track
			// managed fields on the status subresource: https://issue.k8s.io/88901
			if err := r.Client.Status().Patch(
				ctx, cluster, client.MergeFrom(before), r.Owner); err != nil {
				log.Error(err, "patching cluster status")
				return err
			}
			log.V(1).Info("patched cluster status")
		}
		return nil
	}

	if r.Registration != nil && r.Registration.Required(r.Recorder, cluster, &cluster.Status.Conditions) {
		registration.SetAdvanceWarning(r.Recorder, cluster, &cluster.Status.Conditions)
	}
	cluster.Status.RegistrationRequired = nil
	cluster.Status.TokenRequired = ""

	// if the cluster is paused, set a condition and return
	if cluster.Spec.Paused != nil && *cluster.Spec.Paused {
		meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
			Type:    v1beta1.PostgresClusterProgressing,
			Status:  metav1.ConditionFalse,
			Reason:  "Paused",
			Message: "No spec changes will be applied and no other statuses will be updated.",

			ObservedGeneration: cluster.GetGeneration(),
		})
		return runtime.ErrorWithBackoff(patchClusterStatus())
	} else {
		meta.RemoveStatusCondition(&cluster.Status.Conditions, v1beta1.PostgresClusterProgressing)
	}

	if err == nil {
		backupsSpecFound, backupsReconciliationAllowed, err = r.BackupsEnabled(ctx, cluster)

		// If we cannot reconcile because the backup reconciliation is paused, set a condition and exit
		if !backupsReconciliationAllowed {
			meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
				Type:   v1beta1.PostgresClusterProgressing,
				Status: metav1.ConditionFalse,
				Reason: "Paused",
				Message: "Reconciliation is paused: please fill in spec.backups " +
					"or add the postgres-operator.crunchydata.com/authorizeBackupRemoval " +
					"annotation to authorize backup removal.",

				ObservedGeneration: cluster.GetGeneration(),
			})
			return runtime.ErrorWithBackoff(patchClusterStatus())
		} else {
			meta.RemoveStatusCondition(&cluster.Status.Conditions, v1beta1.PostgresClusterProgressing)
		}
	}

	// K8SPG-1045
	if err == nil {
		if err = r.reconcileTLSCondition(ctx, cluster); err != nil {
			return runtime.ErrorWithBackoff(err)
		}

		if meta.IsStatusConditionPresentAndEqual(cluster.Status.Conditions, v1beta1.ConditionTypeTLSSecretsReady, metav1.ConditionFalse) {
			meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
				Type:               v1beta1.PostgresClusterProgressing,
				Status:             metav1.ConditionFalse,
				Reason:             "Paused",
				Message:            "Reconciliation is paused. Check `TLSSecretsReady` condition",
				ObservedGeneration: cluster.GetGeneration(),
			})
			return runtime.ErrorWithBackoff(patchClusterStatus())
		}
		meta.RemoveStatusCondition(&cluster.Status.Conditions, v1beta1.PostgresClusterProgressing)
	}

	pgHBAs := postgres.NewHBAs()
	pmm.PostgreSQLHBAs(cluster, &pgHBAs)
	pgmonitor.PostgreSQLHBAs(cluster, &pgHBAs)
	pgbouncer.PostgreSQL(cluster, &pgHBAs)
	logicalreplica.PostgreSQLHBAs(cluster, &pgHBAs)

	// K8SPG-554
	if cluster.Spec.TLSOnly {
		for i := range pgHBAs.Mandatory {
			pgHBAs.Mandatory[i].TLSOnly()
		}
		for i := range pgHBAs.Default {
			pgHBAs.Default[i].TLSOnly()
		}
	}

	pgParameters := postgres.NewParameters()
	// K8SPG-577
	// K8SPG-884: pg_stat_statements must come before pg_stat_monitor
	if cluster.Spec.Extensions.PGStatStatements {
		pgstatstatements.PostgreSQLParameters(&pgParameters)
	}
	// K8SPG-375
	if cluster.Spec.Extensions.PGStatMonitor {
		pgstatmonitor.PostgreSQLParameters(&pgParameters)
	}
	if cluster.Spec.Extensions.PGAudit {
		pgaudit.PostgreSQLParameters(&pgParameters)
	}
	if cluster.Spec.Extensions.PGCron {
		pgcron.PostgreSQLParameters(&pgParameters)
	}
	if cluster.Spec.Extensions.SetUser {
		setuser.PostgreSQLParameters(&pgParameters)
	}

	// pg_tde should be removed from shared libraries only after extension is dropped
	if cluster.Spec.Extensions.PGTDE.Enabled || meta.IsStatusConditionTrue(cluster.Status.Conditions, v1beta1.PGTDEEnabled) {
		pgtde.PostgreSQLParameters(cluster, &pgParameters)
	}

	pgbackrest.PostgreSQL(cluster, &pgParameters, backupsSpecFound)
	pgmonitor.PostgreSQLParameters(cluster, &pgParameters)

	// Set huge_pages = try if a hugepages resource limit > 0, otherwise set "off"
	postgres.SetHugePages(cluster, &pgParameters)

	if err == nil {
		rootCA, err = r.reconcileRootCertificate(ctx, cluster)
	}

	if err == nil {
		certManagerManaged, certErr := r.isRootCACertManagerManaged(ctx, cluster)
		if certErr != nil {
			log.V(1).Info("failed to check if root CA is cert-manager managed, will retry on next reconcile",
				"error", certErr)
		} else if certManagerManaged {
			r.registerCertManagerWatches(ctx)
		}
	}

	if err == nil {
		// Since any existing data directories must be moved prior to bootstrapping the
		// cluster, further reconciliation will not occur until the directory move Jobs
		// (if configured) have completed. Func reconcileDirMoveJobs() will therefore
		// return a bool indicating that the controller should return early while any
		// required Jobs are running, after which it will indicate that an early
		// return is no longer needed, and reconciliation can proceed normally.
		returnEarly, err := r.reconcileDirMoveJobs(ctx, cluster)
		if err != nil || returnEarly {
			if patchErr := patchClusterStatus(); patchErr != nil {
				if err == nil {
					err = patchErr
				} else {
					log.Error(patchErr, "Failed to patch cluster status")
				}
			}
			return runtime.ErrorWithBackoff(err)
		}
	}
	if err == nil {
		clusterVolumes, err = r.observePersistentVolumeClaims(ctx, cluster)
	}
	if err == nil {
		clusterVolumes, err = r.configureExistingPVCs(ctx, cluster, clusterVolumes)
	}
	if err == nil {
		instances, err = r.observeInstances(ctx, cluster)
	}

	result := reconcile.Result{}

	if err == nil {
		var requeue time.Duration
		if requeue, err = r.reconcilePatroniStatus(ctx, cluster, instances); err == nil && requeue > 0 {
			result.RequeueAfter = requeue
		}
	}
	if err == nil {
		err = r.reconcilePatroniSwitchover(ctx, cluster, instances)
	}
	// reconcile the Pod service before reconciling any data source in case it is necessary
	// to start Pods during data source reconciliation that require network connections (e.g.
	// if it is necessary to start a dedicated repo host to bootstrap a new cluster using its
	// own existing backups).
	if err == nil {
		clusterPodService, err = r.reconcileClusterPodService(ctx, cluster)
	}
	// reconcile the RBAC resources before reconciling any data source in case
	// restore/move Job pods require the ServiceAccount to access any data source.
	// e.g., we are restoring from an S3 source using an IAM for access
	// - https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts-technical-overview.html
	if err == nil {
		instanceServiceAccount, err = r.reconcileRBACResources(ctx, cluster)
	}
	// First handle reconciling any data source configured for the PostgresCluster.  This includes
	// reconciling the data source defined to bootstrap a new cluster, as well as a reconciling
	// a data source to perform restore in-place and re-bootstrap the cluster.
	if err == nil {
		// Since the PostgreSQL data source needs to be populated prior to bootstrapping the
		// cluster, further reconciliation will not occur until the data source (if configured) is
		// initialized.  Func reconcileDataSource() will therefore return a bool indicating that
		// the controller should return early while data initialization is in progress, after
		// which it will indicate that an early return is no longer needed, and reconciliation
		// can proceed normally.
		returnEarly, err := r.reconcileDataSource(ctx, cluster, instances, clusterVolumes, rootCA, backupsSpecFound)
		if err != nil || returnEarly {
			if patchErr := patchClusterStatus(); patchErr != nil {
				if err == nil {
					err = patchErr
				} else {
					log.Error(patchErr, "Failed to patch cluster status")
				}
			}
			return runtime.ErrorWithBackoff(err)
		}
	}
	if err == nil {
		clusterConfigMap, err = r.reconcileClusterConfigMap(ctx, cluster, pgHBAs, pgParameters)
	}
	if err == nil {
		clusterReplicationSecret, err = r.reconcileReplicationSecret(ctx, cluster, rootCA)
	}
	if err == nil {
		patroniLeaderService, err = r.reconcilePatroniLeaderLease(ctx, cluster)
	}
	if err == nil {
		primaryService, err = r.reconcileClusterPrimaryService(ctx, cluster, patroniLeaderService)
	}
	if err == nil {
		replicaService, err = r.reconcileClusterReplicaService(ctx, cluster)
	}
	if err == nil {
		primaryCertificate, err = r.reconcileClusterCertificate(ctx, rootCA, cluster, primaryService, replicaService)
	}
	if err == nil {
		// After the cluster and replication certificates, whose ca.crt this
		// merges with the CAs from spec.tls.additionalTrustedCAs.
		caBundle, err = r.reconcileCABundleSecret(ctx, cluster)
	}
	if err == nil {
		err = r.reconcilePatroniDistributedConfiguration(ctx, cluster)
	}
	if err == nil {
		err = r.reconcilePatroniDynamicConfiguration(ctx, cluster, instances, pgHBAs, pgParameters)
	}
	if err == nil {
		monitoringSecret, err = r.reconcileMonitoringSecret(ctx, cluster)
	}
	if err == nil {
		exporterQueriesConfig, err = r.reconcileExporterQueriesConfig(ctx, cluster)
	}
	if err == nil {
		exporterWebConfig, err = r.reconcileExporterWebConfig(ctx, cluster)
	}
	if err == nil {
		err = r.reconcileInstanceSets(
			ctx, cluster, clusterConfigMap, clusterReplicationSecret, rootCA,
			clusterPodService, instanceServiceAccount, instances, patroniLeaderService,
			primaryCertificate, caBundle, clusterVolumes, exporterQueriesConfig, exporterWebConfig,
			backupsSpecFound,
		)
	}

	if err == nil {
		err = r.reconcilePostgresDatabases(ctx, cluster, instances, patchClusterStatus)
	}
	// K8SPG-911: the two reconcilers around this one need a writable instance.
	// A standby has none, so its pg_tde status comes from what it reports.
	if err == nil {
		r.reconcilePGTDEStandby(ctx, cluster, instances)
	}
	if err == nil {
		err = r.reconcilePGTDEProviders(ctx, cluster, instances, patchClusterStatus)
	}
	if err == nil {
		err = r.reconcilePostgresUsers(ctx, cluster, instances)
	}

	if err == nil {
		var next reconcile.Result
		if next, err = r.reconcilePGBackRest(ctx, cluster,
			instances, rootCA, backupsSpecFound); err == nil && next.RequeueAfter > 0 {
			if result.RequeueAfter == 0 || next.RequeueAfter < result.RequeueAfter {
				result.RequeueAfter = next.RequeueAfter
			}
		}
	}
	if err == nil {
		dedicatedSnapshotPVC, err = r.reconcileDedicatedSnapshotVolume(ctx, cluster, clusterVolumes)
	}
	if err == nil {
		err = r.reconcileVolumeSnapshots(ctx, cluster, dedicatedSnapshotPVC)
	}
	if err == nil {
		err = r.reconcilePGBouncer(ctx, cluster, instances, primaryCertificate, caBundle, rootCA)
	}
	if err == nil {
		err = r.reconcilePGMonitor(ctx, cluster, instances, monitoringSecret)
	}
	if err == nil {
		err = r.reconcileDatabaseInitSQL(ctx, cluster, instances)
	}
	if err == nil {
		err = r.reconcilePGAdmin(ctx, cluster)
	}
	if err == nil {
		// This is after [Reconciler.rolloutInstances] to ensure that recreating
		// Pods takes precedence.
		err = r.handlePatroniRestarts(ctx, cluster, instances)
	}

	// at this point everything reconciled successfully, and we can update the
	// observedGeneration
	cluster.Status.ObservedGeneration = cluster.GetGeneration()

	log.V(1).Info("reconciled cluster")

	if patchErr := patchClusterStatus(); patchErr != nil {
		if err != nil {
			log.Error(patchErr, "Failed to patch cluster status")
		} else {
			err = errors.Wrap(patchErr, "failed to patch cluster status")
		}
	}

	return result, err
}

// deleteControlled safely deletes object when it is controlled by cluster.
func (r *Reconciler) deleteControlled(
	ctx context.Context, cluster *v1beta1.PostgresCluster, object client.Object,
) error {
	if metav1.IsControlledBy(object, cluster) {
		uid := object.GetUID()
		version := object.GetResourceVersion()
		exactly := client.Preconditions{UID: &uid, ResourceVersion: &version}

		return r.Client.Delete(ctx, object, exactly)
	}

	return nil
}

// patch sends patch to object's endpoint in the Kubernetes API and updates
// object with any returned content. The fieldManager is set to r.Owner, but
// can be overridden in options.
// - https://docs.k8s.io/reference/using-api/server-side-apply/#managers
func (r *Reconciler) patch(
	ctx context.Context, object client.Object,
	patch client.Patch, options ...client.PatchOption,
) error {
	options = append([]client.PatchOption{r.Owner}, options...)
	return r.Client.Patch(ctx, object, patch, options...)
}

// The owner reference created by controllerutil.SetControllerReference blocks
// deletion. The OwnerReferencesPermissionEnforcement plugin requires that the
// creator of such a reference have either "delete" permission on the owner or
// "update" permission on the owner's "finalizers" subresource.
// - https://docs.k8s.io/reference/access-authn-authz/admission-controllers/
// +kubebuilder:rbac:groups="upstream.pgv2.percona.com",resources="postgresclusters/finalizers",verbs={update}

// setControllerReference sets owner as a Controller OwnerReference on controlled.
// Only one OwnerReference can be a controller, so it returns an error if another
// is already set.
func (r *Reconciler) setControllerReference(
	owner *v1beta1.PostgresCluster, controlled client.Object,
) error {
	return controllerutil.SetControllerReference(owner, controlled, r.Client.Scheme())
}

// setOwnerReference sets an OwnerReference on the object without setting the
// owner as a controller. This allows for multiple OwnerReferences on an object.
func (r *Reconciler) setOwnerReference(
	owner *v1beta1.PostgresCluster, controlled client.Object,
) error {
	return controllerutil.SetOwnerReference(owner, controlled, r.Client.Scheme())
}

// +kubebuilder:rbac:groups="",resources="configmaps",verbs={get,list,watch}
// +kubebuilder:rbac:groups="",resources="endpoints",verbs={get,list,watch}
// +kubebuilder:rbac:groups="",resources="persistentvolumeclaims",verbs={get,list,watch}
// +kubebuilder:rbac:groups="",resources="secrets",verbs={get,list,watch}
// +kubebuilder:rbac:groups="",resources="services",verbs={get,list,watch}
// +kubebuilder:rbac:groups="",resources="serviceaccounts",verbs={get,list,watch}
// +kubebuilder:rbac:groups="apps",resources="deployments",verbs={get,list,watch}
// +kubebuilder:rbac:groups="apps",resources="statefulsets",verbs={get,list,watch}
// +kubebuilder:rbac:groups="batch",resources="jobs",verbs={get,list,watch}
// +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources="roles",verbs={get,list,watch}
// +kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources="rolebindings",verbs={get,list,watch}
// +kubebuilder:rbac:groups="batch",resources="cronjobs",verbs={get,list,watch}
// +kubebuilder:rbac:groups="policy",resources="poddisruptionbudgets",verbs={get,list,watch}

// SetupWithManager adds the PostgresCluster controller to the provided runtime manager
func (r *Reconciler) SetupWithManager(mgr manager.Manager) error {
	if r.PodExec == nil {
		var err error
		r.PodExec, err = runtime.NewPodExecutor(mgr.GetConfig())
		if err != nil {
			return err
		}
	}

	if r.newPGBouncerAdmin == nil {
		r.newPGBouncerAdmin = func(o pgbruntime.AdminClientOptions) (pgbruntime.AdminClient, error) {
			return pgbruntime.NewAdminClient(o)
		}
	}

	if r.DiscoveryClient == nil {
		var err error
		r.DiscoveryClient, err = discovery.NewDiscoveryClientForConfig(mgr.GetConfig())
		if err != nil {
			return err
		}
	}

	r.Cache = mgr.GetCache()

	if err := mgr.GetFieldIndexer().IndexField(
		context.Background(),
		&v1beta1.PostgresCluster{},
		v1beta1.IndexFieldPGBouncerUserSecrets,
		v1beta1.PGBouncerUserSecretsIndexerFunc,
	); err != nil {
		return err
	}

	if err := mgr.GetFieldIndexer().IndexField(
		context.Background(),
		&v1beta1.PostgresCluster{},
		v1beta1.IndexFieldAdditionalTrustedCASecrets,
		v1beta1.AdditionalTrustedCASecretsIndexerFunc,
	); err != nil {
		return err
	}

	// K8SPG-712: Allow overriding default configurations
	configMapPredicate := builder.WithPredicates(predicate.Funcs{
		UpdateFunc: func(e event.UpdateEvent) bool {
			configMap, ok := e.ObjectNew.(*corev1.ConfigMap)
			if !ok {
				return true
			}
			// Skip reconciliation if the ConfigMap has the specified annotation
			_, hasAnnotation := configMap.Annotations[naming.OverrideConfigAnnotation]
			return !hasAnnotation
		},
	})

	bldr := builder.ControllerManagedBy(mgr).
		For(&v1beta1.PostgresCluster{}).
		Owns(&corev1.ConfigMap{}, configMapPredicate). // K8SPG-712
		Owns(&corev1.Endpoints{}).                     //nolint:staticcheck // SA1019: matches production code
		Owns(&corev1.PersistentVolumeClaim{}).
		Owns(&corev1.Secret{}).
		Owns(&corev1.Service{}).
		Owns(&corev1.ServiceAccount{}).
		Owns(&appsv1.Deployment{}).
		Owns(&appsv1.StatefulSet{}).
		Owns(&batchv1.Job{}).
		Owns(&rbacv1.Role{}).
		Owns(&rbacv1.RoleBinding{}).
		Owns(&batchv1.CronJob{}).
		Owns(&policyv1.PodDisruptionBudget{}).
		Watches(&corev1.Secret{}, r.watchClusterSecrets(), builder.WithPredicates(predicate.NewPredicateFuncs(func(obj client.Object) bool {
			_, hasCluster := obj.GetLabels()[naming.LabelCluster]
			return hasCluster
		}))).
		Watches(&corev1.Pod{}, r.watchPods()).
		Watches(&corev1.Secret{}, r.watchPGBouncerUserSecrets()).
		Watches(&corev1.Secret{}, r.watchAdditionalTrustedCASecrets()).
		Watches(&appsv1.StatefulSet{},
			r.controllerRefHandlerFuncs()) // watch all StatefulSets

	// When cert-manager is installed, watch Certificate resources owned by
	// PostgresCluster and cert-manager-issued Secrets (which are owned by
	// Certificate, not PostgresCluster) so that deletions or renewals trigger
	// an immediate reconcile rather than waiting for the next resync.
	certManagerExists, err := k8s.GroupVersionKindExists(r.DiscoveryClient, "cert-manager.io/v1", "Certificate")
	if err != nil {
		return err
	}
	if certManagerExists {
		certManagerSecretPredicate := builder.WithPredicates(predicate.NewPredicateFuncs(func(obj client.Object) bool {
			_, hasCluster := obj.GetLabels()[naming.LabelCluster]
			_, hasCertAnnotation := obj.GetAnnotations()["cert-manager.io/certificate-name"]
			return hasCluster && hasCertAnnotation
		}))
		bldr.Owns(&cmv1.Certificate{}).
			Owns(&cmv1.Issuer{}).
			Watches(&corev1.Secret{}, r.watchCertManagerSecrets(), certManagerSecretPredicate)
	}

	return bldr.Complete(r)
}

// registerCertManagerWatches dynamically registers watches for cert-manager
// Certificate, Issuer, and cert-manager-issued Secret resources for
// case where cert-manager is installed after the operator starts, so the
// watches were not registered in SetupWithManager.
func (r *Reconciler) registerCertManagerWatches(ctx context.Context) {
	if r.Controller == nil || r.Cache == nil {
		return
	}

	if r.certManagerWatchesRegistered.Load() {
		return
	}

	log := logging.FromContext(ctx)

	certHandler := handler.TypedEnqueueRequestForOwner[*cmv1.Certificate](
		r.Scheme, r.Client.RESTMapper(),
		&v1beta1.PostgresCluster{},
		handler.OnlyControllerOwner(),
	)
	if err := r.Controller.Watch(source.Kind(
		r.Cache, &cmv1.Certificate{}, certHandler,
	)); err != nil {
		log.Error(err, "failed to register dynamic watch for Certificates")
		return
	}

	issuerHandler := handler.TypedEnqueueRequestForOwner[*cmv1.Issuer](
		r.Scheme, r.Client.RESTMapper(),
		&v1beta1.PostgresCluster{},
		handler.OnlyControllerOwner(),
	)
	if err := r.Controller.Watch(source.Kind(
		r.Cache, &cmv1.Issuer{}, issuerHandler,
	)); err != nil {
		log.Error(err, "failed to register dynamic watch for Issuers")
		return
	}
	secretHandler := handler.TypedEnqueueRequestsFromMapFunc(
		func(ctx context.Context, secret *corev1.Secret) []reconcile.Request {
			cluster := secret.GetLabels()[naming.LabelCluster]
			if len(cluster) > 0 {
				return []reconcile.Request{
					{NamespacedName: client.ObjectKey{
						Namespace: secret.GetNamespace(),
						Name:      cluster,
					}},
				}
			}
			return nil
		},
	)
	certManagerSecretPredicate := predicate.NewTypedPredicateFuncs(func(secret *corev1.Secret) bool {
		_, hasCluster := secret.GetLabels()[naming.LabelCluster]
		_, hasCertAnnotation := secret.GetAnnotations()["cert-manager.io/certificate-name"]
		return hasCluster && hasCertAnnotation
	})
	if err := r.Controller.Watch(source.Kind(
		r.Cache, &corev1.Secret{}, secretHandler, certManagerSecretPredicate,
	)); err != nil {
		log.Error(err, "failed to register dynamic watch for cert-manager Secrets")
		return
	}

	r.certManagerWatchesRegistered.Store(true)

	log.Info("dynamically registered cert-manager watches")
}