From b143884b62bfcccd4b3248ca8fb4f80b85db4ea2 Mon Sep 17 00:00:00 2001 From: yoav-katz Date: Wed, 5 Aug 2026 18:02:49 +0300 Subject: [PATCH 1/5] feat(dcs-package) --- .../controller/postgrescluster/cluster.go | 54 +-- .../controller/postgrescluster/instance.go | 6 +- .../controller/postgrescluster/patroni.go | 154 +------- .../postgrescluster/patroni_test.go | 207 ---------- internal/controller/postgrescluster/rbac.go | 3 +- internal/patroni/config.go | 110 ++---- internal/patroni/config_test.go | 126 +++--- internal/patroni/dcs/dcs.go | 101 +++++ internal/patroni/dcs/kubernetes.go | 335 ++++++++++++++++ internal/patroni/dcs/kubernetes_test.go | 360 ++++++++++++++++++ internal/patroni/rbac.go | 55 +-- internal/patroni/rbac_test.go | 84 +--- internal/patroni/reconcile.go | 22 +- internal/patroni/reconcile_test.go | 64 ++-- 14 files changed, 1022 insertions(+), 659 deletions(-) create mode 100644 internal/patroni/dcs/dcs.go create mode 100644 internal/patroni/dcs/kubernetes.go create mode 100644 internal/patroni/dcs/kubernetes_test.go diff --git a/internal/controller/postgrescluster/cluster.go b/internal/controller/postgrescluster/cluster.go index 1bba315bf3..7c9c32a643 100644 --- a/internal/controller/postgrescluster/cluster.go +++ b/internal/controller/postgrescluster/cluster.go @@ -17,6 +17,7 @@ import ( "github.com/percona/percona-postgresql-operator/v2/internal/initialize" "github.com/percona/percona-postgresql-operator/v2/internal/naming" "github.com/percona/percona-postgresql-operator/v2/internal/patroni" + "github.com/percona/percona-postgresql-operator/v2/internal/patroni/dcs" "github.com/percona/percona-postgresql-operator/v2/internal/pki" "github.com/percona/percona-postgresql-operator/v2/internal/postgres" "github.com/percona/percona-postgresql-operator/v2/internal/util" @@ -44,7 +45,7 @@ func (r *Reconciler) reconcileClusterConfigMap( if err == nil { err = patroni.ClusterConfigMap(ctx, cluster, pgHBAs, pgParameters, - clusterConfigMap) + dcs.For(cluster).ClusterYAML(cluster), clusterConfigMap) } if err == nil { err = errors.WithStack(r.apply(ctx, clusterConfigMap)) @@ -117,43 +118,20 @@ func (r *Reconciler) generateClusterPrimaryService( err := errors.WithStack(r.setControllerReference(cluster, service)) - // Endpoints for a Service have the same name as the Service. Copy labels, - // annotations, and ownership, too. - endpoints := &corev1.Endpoints{} - service.ObjectMeta.DeepCopyInto(&endpoints.ObjectMeta) - endpoints.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Endpoints")) - - if leader == nil { - // TODO(cbandy): We need to build a different kind of Service here. - return nil, nil, errors.New("Patroni DCS other than Kubernetes Endpoints is not implemented") + spec, subset, backendErr := dcs.For(cluster).PrimaryService(cluster, leader) + if backendErr != nil { + return nil, nil, backendErr } - - // Allocate no IP address (headless) and manage the Endpoints ourselves. - // - https://docs.k8s.io/concepts/services-networking/service/#headless-services - // - https://docs.k8s.io/concepts/services-networking/service/#services-without-selectors - service.Spec.ClusterIP = corev1.ClusterIPNone - service.Spec.Selector = nil - - service.Spec.Ports = []corev1.ServicePort{{ - Name: naming.PortPostgreSQL, - Port: *cluster.Spec.Port, - Protocol: corev1.ProtocolTCP, - TargetPort: intstr.FromString(naming.PortPostgreSQL), - }} - - // Resolve to the ClusterIP for which Patroni has configured the Endpoints. - endpoints.Subsets = []corev1.EndpointSubset{{ - Addresses: []corev1.EndpointAddress{{IP: leader.Spec.ClusterIP}}, - }} - - // Copy the EndpointPorts from the ServicePorts. - for _, sp := range service.Spec.Ports { - endpoints.Subsets[0].Ports = append(endpoints.Subsets[0].Ports, - corev1.EndpointPort{ - Name: sp.Name, - Port: sp.Port, - Protocol: sp.Protocol, - }) + service.Spec = spec + + var endpoints *corev1.Endpoints + if subset != nil { + // Endpoints for a Service have the same name as the Service. Copy labels, + // annotations, and ownership, too. + endpoints = &corev1.Endpoints{} + service.ObjectMeta.DeepCopyInto(&endpoints.ObjectMeta) + endpoints.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Endpoints")) + endpoints.Subsets = []corev1.EndpointSubset{*subset} } return service, endpoints, err @@ -177,7 +155,7 @@ func (r *Reconciler) reconcileClusterPrimaryService( if err == nil { err = errors.WithStack(r.apply(ctx, service)) } - if err == nil { + if err == nil && endpoints != nil { err = errors.WithStack(r.apply(ctx, endpoints)) } return service, err diff --git a/internal/controller/postgrescluster/instance.go b/internal/controller/postgrescluster/instance.go index ea67429f6d..0938356c39 100644 --- a/internal/controller/postgrescluster/instance.go +++ b/internal/controller/postgrescluster/instance.go @@ -36,6 +36,7 @@ import ( "github.com/percona/percona-postgresql-operator/v2/internal/logging" "github.com/percona/percona-postgresql-operator/v2/internal/naming" "github.com/percona/percona-postgresql-operator/v2/internal/patroni" + "github.com/percona/percona-postgresql-operator/v2/internal/patroni/dcs" "github.com/percona/percona-postgresql-operator/v2/internal/pgbackrest" "github.com/percona/percona-postgresql-operator/v2/internal/pgtde" "github.com/percona/percona-postgresql-operator/v2/internal/pki" @@ -1277,8 +1278,9 @@ func (r *Reconciler) reconcileInstance( return errors.Wrap(err, "failed to determine initial init image") } + dcsEnvVars := dcs.For(cluster).InstanceEnvVars(cluster, patroniLeaderService, instance.Spec.Template.Spec.Containers) err = patroni.InstancePod( - ctx, cluster, clusterConfigMap, clusterPodService, patroniLeaderService, + ctx, cluster, clusterConfigMap, clusterPodService, dcsEnvVars, spec, instanceCertificates, instanceConfigMap, &instance.Spec.Template, initImage) // K8SPG-708 } @@ -1548,7 +1550,7 @@ func (r *Reconciler) reconcileInstanceConfigMap( }, cluster.Name, "pg", cluster.Labels[naming.LabelVersion])) if err == nil { - err = patroni.InstanceConfigMap(ctx, cluster, spec, instanceConfigMap) + err = patroni.InstanceConfigMap(ctx, cluster, spec, dcs.For(cluster).InstanceYAML(cluster), instanceConfigMap) } if err == nil { err = errors.WithStack(r.apply(ctx, instanceConfigMap)) diff --git a/internal/controller/postgrescluster/patroni.go b/internal/controller/postgrescluster/patroni.go index ec61fece40..0d6f4b1e0e 100644 --- a/internal/controller/postgrescluster/patroni.go +++ b/internal/controller/postgrescluster/patroni.go @@ -12,13 +12,12 @@ import ( "github.com/pkg/errors" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/util/intstr" "sigs.k8s.io/controller-runtime/pkg/client" - "github.com/percona/percona-postgresql-operator/v2/internal/initialize" "github.com/percona/percona-postgresql-operator/v2/internal/logging" "github.com/percona/percona-postgresql-operator/v2/internal/naming" "github.com/percona/percona-postgresql-operator/v2/internal/patroni" + "github.com/percona/percona-postgresql-operator/v2/internal/patroni/dcs" "github.com/percona/percona-postgresql-operator/v2/internal/pki" "github.com/percona/percona-postgresql-operator/v2/internal/postgres" "github.com/percona/percona-postgresql-operator/v2/percona/certmanager" @@ -30,22 +29,7 @@ import ( func (r *Reconciler) deletePatroniArtifacts( ctx context.Context, cluster *v1beta1.PostgresCluster, ) error { - // TODO(cbandy): This could also be accomplished by adopting the Endpoints - // as Patroni creates them. Would their events cause too many reconciles? - // Foreground deletion may force us to adopt and set finalizers anyway. - - selector, err := naming.AsSelector(naming.ClusterPatronis(cluster)) - if err == nil { - err = errors.WithStack( - r.Client.DeleteAllOf( - ctx, &corev1.Endpoints{}, - client.InNamespace(cluster.Namespace), - client.MatchingLabelsSelector{Selector: selector}, - ), - ) - } - - return err + return dcs.For(cluster).Delete(ctx, r.Client, cluster) } func (r *Reconciler) handlePatroniRestarts( @@ -147,15 +131,12 @@ func (r *Reconciler) handlePatroniRestarts( func (r *Reconciler) reconcilePatroniDistributedConfiguration( ctx context.Context, cluster *v1beta1.PostgresCluster, ) error { - // When using Endpoints for DCS, Patroni needs a Service to ensure that the - // Endpoints object is not removed by Kubernetes at startup. Patroni will - // create this object if it has permission to do so, but it won't set any - // ownership. - // - https://releases.k8s.io/v1.16.0/pkg/controller/endpoint/endpoints_controller.go#L547 - // - https://releases.k8s.io/v1.20.0/pkg/controller/endpoint/endpoints_controller.go#L580 - // - https://github.com/zalando/patroni/blob/v2.0.1/patroni/dcs/kubernetes.py#L865-L881 - dcsService := &corev1.Service{ObjectMeta: naming.PatroniDistributedConfiguration(cluster)} - dcsService.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Service")) + // The DCS backend may not own any Kubernetes object for its distributed + // configuration (e.g. an external DCS). + dcsService := dcs.For(cluster).DistributedConfigurationService(cluster) + if dcsService == nil { + return nil + } err := errors.WithStack(r.setControllerReference(cluster, dcsService)) @@ -170,11 +151,6 @@ func (r *Reconciler) reconcilePatroniDistributedConfiguration( }, cluster.Name, "", cluster.Labels[naming.LabelVersion]), ) - // Allocate no IP address (headless) and create no Endpoints. - // - https://docs.k8s.io/concepts/services-networking/service/#headless-services - dcsService.Spec.ClusterIP = corev1.ClusterIPNone - dcsService.Spec.Selector = nil - if err == nil { err = errors.WithStack(r.apply(ctx, dcsService)) } @@ -233,80 +209,6 @@ func (r *Reconciler) reconcilePatroniDynamicConfiguration( ) } -// generatePatroniLeaderLeaseService returns a v1.Service that exposes the -// Patroni leader when Patroni is using Endpoints for its leader elections. -func (r *Reconciler) generatePatroniLeaderLeaseService( - cluster *v1beta1.PostgresCluster) (*corev1.Service, error, -) { - service := &corev1.Service{ObjectMeta: naming.PatroniLeaderEndpoints(cluster)} - service.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Service")) - - service.Annotations = naming.Merge( - cluster.Spec.Metadata.GetAnnotationsOrNil(), - ) - service.Labels = naming.Merge( - cluster.Spec.Metadata.GetLabelsOrNil(), - ) - - if spec := cluster.Spec.Service; spec != nil { - service.Annotations = naming.Merge(service.Annotations, - spec.Metadata.GetAnnotationsOrNil()) - service.Labels = naming.Merge(service.Labels, - spec.Metadata.GetLabelsOrNil()) - } - - // add our labels last so they aren't overwritten - service.Labels = naming.Merge(service.Labels, - naming.WithPerconaLabels(map[string]string{ // K8SPG-430 - naming.LabelCluster: cluster.Name, - naming.LabelPatroni: naming.PatroniScope(cluster), - }, cluster.Name, "", cluster.Labels[naming.LabelVersion])) - - // Allocate an IP address and/or node port and let Patroni manage the Endpoints. - // Patroni will ensure that they always route to the elected leader. - // - https://docs.k8s.io/concepts/services-networking/service/#services-without-selectors - service.Spec.Selector = nil - - // The TargetPort must be the name (not the number) of the PostgreSQL - // ContainerPort. This name allows the port number to differ between - // instances, which can happen during a rolling update. - servicePort := corev1.ServicePort{ - Name: naming.PortPostgreSQL, - Port: *cluster.Spec.Port, - Protocol: corev1.ProtocolTCP, - TargetPort: intstr.FromString(naming.PortPostgreSQL), - } - - if spec := cluster.Spec.Service; spec == nil { - service.Spec.Type = corev1.ServiceTypeClusterIP - } else { - service.Spec.Type = corev1.ServiceType(spec.Type) - // K8SPG-389 - service.Spec.LoadBalancerSourceRanges = spec.LoadBalancerSourceRanges - - if spec.NodePort != nil { - if service.Spec.Type == corev1.ServiceTypeClusterIP { - // The NodePort can only be set when the Service type is NodePort or - // LoadBalancer. However, due to a known issue prior to Kubernetes - // 1.20, we clear these errors during our apply. To preserve the - // appropriate behavior, we log an Event and return an error. - // TODO(tjmoore4): Once Validation Rules are available, this check - // and event could potentially be removed in favor of that validation - r.Recorder.Eventf(cluster, corev1.EventTypeWarning, "MisconfiguredClusterIP", - "NodePort cannot be set with type ClusterIP on Service %q", service.Name) - return nil, errors.Errorf("NodePort cannot be set with type ClusterIP on Service %q", service.Name) - } - servicePort.NodePort = *spec.NodePort - } - service.Spec.ExternalTrafficPolicy = initialize.FromPointer(spec.ExternalTrafficPolicy) - service.Spec.InternalTrafficPolicy = spec.InternalTrafficPolicy - } - service.Spec.Ports = []corev1.ServicePort{servicePort} - - err := errors.WithStack(r.setControllerReference(cluster, service)) - return service, err -} - // +kubebuilder:rbac:groups="",resources="services",verbs={create,patch} // reconcilePatroniLeaderLease sets labels and ownership on the objects Patroni @@ -315,12 +217,11 @@ func (r *Reconciler) generatePatroniLeaderLeaseService( func (r *Reconciler) reconcilePatroniLeaderLease( ctx context.Context, cluster *v1beta1.PostgresCluster, ) (*corev1.Service, error) { - // When using Endpoints for DCS, Patroni needs a Service to ensure that the - // Endpoints object is not removed by Kubernetes at startup. - // - https://releases.k8s.io/v1.16.0/pkg/controller/endpoint/endpoints_controller.go#L547 - // - https://releases.k8s.io/v1.20.0/pkg/controller/endpoint/endpoints_controller.go#L580 - service, err := r.generatePatroniLeaderLeaseService(cluster) - if err == nil { + service, err := dcs.For(cluster).LeaderLeaseService(cluster, r.Recorder) + if err == nil && service != nil { + err = errors.WithStack(r.setControllerReference(cluster, service)) + } + if err == nil && service != nil { err = errors.WithStack(r.apply(ctx, service)) } return service, err @@ -333,9 +234,6 @@ func (r *Reconciler) reconcilePatroniStatus( ctx context.Context, cluster *v1beta1.PostgresCluster, observedInstances *observedInstances, ) (time.Duration, error) { - var requeue time.Duration - log := logging.FromContext(ctx) - var readyInstance bool for _, instance := range observedInstances.forCluster { if r, _ := instance.IsReady(); r { @@ -343,29 +241,13 @@ func (r *Reconciler) reconcilePatroniStatus( } } - dcs := &corev1.Endpoints{ObjectMeta: naming.PatroniDistributedConfiguration(cluster)} - err := errors.WithStack(client.IgnoreNotFound( - r.Client.Get(ctx, client.ObjectKeyFromObject(dcs), dcs), - )) - - if err == nil { - if dcs.Annotations["initialize"] != "" { - // After bootstrap, Patroni writes the cluster system identifier to DCS. - cluster.Status.Patroni.SystemIdentifier = dcs.Annotations["initialize"] - } else if readyInstance { - // While we typically expect a value for the initialize key to be present in the - // Endpoints above by the time the StatefulSet for any instance indicates "ready" - // (since Patroni writes this value after successful cluster bootstrap, at which time - // the initial primary should transition to "ready"), sometimes this is not the case - // and the "initialize" key is not yet present. Therefore, if a "ready" instance - // is detected in the cluster we assume this is the case, and simply log a message and - // requeue in order to try again until the expected value is found. - log.Info("detected ready instance but no initialize value") - requeue = time.Second - } + observation, err := dcs.For(cluster).Observe(ctx, r.Client, cluster, readyInstance) + if err == nil && observation.SystemIdentifier != "" { + // After bootstrap, the DCS backend reports the cluster system identifier. + cluster.Status.Patroni.SystemIdentifier = observation.SystemIdentifier } - return requeue, err + return observation.RequeueAfter, err } // reconcileReplicationSecret creates a secret containing the TLS diff --git a/internal/controller/postgrescluster/patroni_test.go b/internal/controller/postgrescluster/patroni_test.go index 6ceb8069dc..248a76aa8a 100644 --- a/internal/controller/postgrescluster/patroni_test.go +++ b/internal/controller/postgrescluster/patroni_test.go @@ -21,220 +21,13 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/tools/record" "sigs.k8s.io/controller-runtime/pkg/client" "github.com/percona/percona-postgresql-operator/v2/internal/naming" - "github.com/percona/percona-postgresql-operator/v2/internal/testing/cmp" "github.com/percona/percona-postgresql-operator/v2/internal/testing/require" "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" ) -func TestGeneratePatroniLeaderLeaseService(t *testing.T) { - _, cc := setupKubernetes(t) - require.ParallelCapacity(t, 0) - - reconciler := &Reconciler{ - Client: cc, - Recorder: new(record.FakeRecorder), - } - - cluster := &v1beta1.PostgresCluster{} - cluster.Namespace = "ns1" - cluster.Name = "pg2" - cluster.Spec.Port = new(int32(9876)) - - cluster.Labels = map[string]string{ - naming.LabelVersion: "2.3.0", - } - - alwaysExpect := func(t testing.TB, service *corev1.Service) { - assert.Assert(t, cmp.MarshalMatches(service.TypeMeta, ` -apiVersion: v1 -kind: Service - `)) - assert.Assert(t, cmp.MarshalMatches(service.ObjectMeta, ` -labels: - app.kubernetes.io/instance: pg2 - app.kubernetes.io/managed-by: percona-postgresql-operator - app.kubernetes.io/name: percona-postgresql - app.kubernetes.io/part-of: percona-postgresql - postgres-operator.crunchydata.com/cluster: pg2 - postgres-operator.crunchydata.com/patroni: pg2-ha -name: pg2-ha -namespace: ns1 -ownerReferences: -- apiVersion: upstream.pgv2.percona.com/v1beta1 - blockOwnerDeletion: true - controller: true - kind: PostgresCluster - name: pg2 - uid: "" - `)) - - // Always gets a ClusterIP (never None). - assert.Equal(t, service.Spec.ClusterIP, "") - assert.Assert(t, service.Spec.Selector == nil, - "got %v", service.Spec.Selector) - } - - t.Run("NoServiceSpec", func(t *testing.T) { - service, err := reconciler.generatePatroniLeaderLeaseService(cluster) - assert.NilError(t, err) - alwaysExpect(t, service) - // Defaults to ClusterIP. - assert.Equal(t, service.Spec.Type, corev1.ServiceTypeClusterIP) - assert.Assert(t, cmp.MarshalMatches(service.Spec.Ports, ` -- name: postgres - port: 9876 - protocol: TCP - targetPort: postgres - `)) - }) - - t.Run("AnnotationsLabels", func(t *testing.T) { - cluster := cluster.DeepCopy() - cluster.Spec.Metadata = &v1beta1.Metadata{ - Annotations: map[string]string{"a": "v1"}, - Labels: map[string]string{"b": "v2"}, - } - cluster.Labels = map[string]string{ - naming.LabelVersion: "2.3.0", - } - - service, err := reconciler.generatePatroniLeaderLeaseService(cluster) - assert.NilError(t, err) - - // Annotations present in the metadata. - assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ - "a": "v1", - }) - - // Labels present in the metadata. - assert.DeepEqual(t, service.ObjectMeta.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ - "b": "v2", - "postgres-operator.crunchydata.com/cluster": "pg2", - "postgres-operator.crunchydata.com/patroni": "pg2-ha", - }, "pg2", "", "2.3.0"))) - - // Labels not in the selector. - assert.Assert(t, service.Spec.Selector == nil, - "got %v", service.Spec.Selector) - - // Add metadata to individual service - cluster.Spec.Service = &v1beta1.ServiceSpec{ - Metadata: &v1beta1.Metadata{ - Annotations: map[string]string{"c": "v3"}, - Labels: map[string]string{"d": "v4", - "postgres-operator.crunchydata.com/cluster": "wrongName"}, - }, - } - - service, err = reconciler.generatePatroniLeaderLeaseService(cluster) - assert.NilError(t, err) - - // Annotations present in the metadata. - assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ - "a": "v1", - "c": "v3", - }) - - // Labels present in the metadata. - assert.DeepEqual(t, service.ObjectMeta.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ - "b": "v2", - "d": "v4", - "postgres-operator.crunchydata.com/cluster": "pg2", - "postgres-operator.crunchydata.com/patroni": "pg2-ha", - }, "pg2", "", "2.3.0"))) - - // Labels not in the selector. - assert.Assert(t, service.Spec.Selector == nil, - "got %v", service.Spec.Selector) - }) - - types := []struct { - Type string - Expect func(testing.TB, *corev1.Service) - }{ - {Type: "ClusterIP", Expect: func(t testing.TB, service *corev1.Service) { - assert.Equal(t, service.Spec.Type, corev1.ServiceTypeClusterIP) - }}, - {Type: "NodePort", Expect: func(t testing.TB, service *corev1.Service) { - assert.Equal(t, service.Spec.Type, corev1.ServiceTypeNodePort) - }}, - {Type: "LoadBalancer", Expect: func(t testing.TB, service *corev1.Service) { - assert.Equal(t, service.Spec.Type, corev1.ServiceTypeLoadBalancer) - }}, - } - - for _, test := range types { - t.Run(test.Type, func(t *testing.T) { - cluster := cluster.DeepCopy() - cluster.Spec.Service = &v1beta1.ServiceSpec{Type: test.Type} - - service, err := reconciler.generatePatroniLeaderLeaseService(cluster) - assert.NilError(t, err) - alwaysExpect(t, service) - test.Expect(t, service) - assert.Assert(t, cmp.MarshalMatches(service.Spec.Ports, ` -- name: postgres - port: 9876 - protocol: TCP - targetPort: postgres - `)) - }) - } - - typesAndPort := []struct { - Description string - Type string - NodePort *int32 - Expect func(testing.TB, *corev1.Service, error) - }{ - {Description: "ClusterIP with Port 32000", Type: "ClusterIP", - NodePort: new(int32(32000)), Expect: func(t testing.TB, service *corev1.Service, err error) { - assert.ErrorContains(t, err, "NodePort cannot be set with type ClusterIP on Service \"pg2-ha\"") - assert.Assert(t, service == nil) - }}, - {Description: "NodePort with Port 32001", Type: "NodePort", - NodePort: new(int32(32001)), Expect: func(t testing.TB, service *corev1.Service, err error) { - assert.NilError(t, err) - alwaysExpect(t, service) - assert.Equal(t, service.Spec.Type, corev1.ServiceTypeNodePort) - assert.Assert(t, cmp.MarshalMatches(service.Spec.Ports, ` -- name: postgres - nodePort: 32001 - port: 9876 - protocol: TCP - targetPort: postgres -`)) - }}, - {Description: "LoadBalancer with Port 32002", Type: "LoadBalancer", - NodePort: new(int32(32002)), Expect: func(t testing.TB, service *corev1.Service, err error) { - assert.Equal(t, service.Spec.Type, corev1.ServiceTypeLoadBalancer) - assert.NilError(t, err) - alwaysExpect(t, service) - assert.Assert(t, cmp.MarshalMatches(service.Spec.Ports, ` -- name: postgres - nodePort: 32002 - port: 9876 - protocol: TCP - targetPort: postgres -`)) - }}, - } - - for _, test := range typesAndPort { - t.Run(test.Description, func(t *testing.T) { - cluster := cluster.DeepCopy() - cluster.Spec.Service = &v1beta1.ServiceSpec{Type: test.Type, NodePort: test.NodePort} - - service, err := reconciler.generatePatroniLeaderLeaseService(cluster) - test.Expect(t, service, err) - }) - } -} - func TestReconcilePatroniLeaderLease(t *testing.T) { ctx := context.Background() _, cc := setupKubernetes(t) diff --git a/internal/controller/postgrescluster/rbac.go b/internal/controller/postgrescluster/rbac.go index 73817e310e..ca992a7d59 100644 --- a/internal/controller/postgrescluster/rbac.go +++ b/internal/controller/postgrescluster/rbac.go @@ -13,6 +13,7 @@ import ( "github.com/percona/percona-postgresql-operator/v2/internal/naming" "github.com/percona/percona-postgresql-operator/v2/internal/patroni" + "github.com/percona/percona-postgresql-operator/v2/internal/patroni/dcs" "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" ) @@ -79,7 +80,7 @@ func (r *Reconciler) reconcileInstanceRBAC( Kind: account.Kind, Name: account.Name, }} - role.Rules = patroni.Permissions(cluster) + role.Rules = append(patroni.Permissions(cluster), dcs.For(cluster).Permissions(cluster)...) if err == nil { err = errors.WithStack(r.apply(ctx, account)) diff --git a/internal/patroni/config.go b/internal/patroni/config.go index 848ac0cee6..498875c18b 100644 --- a/internal/patroni/config.go +++ b/internal/patroni/config.go @@ -42,38 +42,20 @@ func quoteShellWord(s string) string { } // clusterYAML returns Patroni settings that apply to the entire cluster. +// dcsYAML contains the DCS backend's config additions (see internal/patroni/dcs); +// its top-level keys are merged into the result, except "postgresql" which is +// shallow-merged into the "postgresql" section below instead of replacing it. func clusterYAML( cluster *v1beta1.PostgresCluster, pgHBAs postgres.HBAs, pgParameters postgres.Parameters, + dcsYAML map[string]any, ) (string, error) { - labels := map[string]string{naming.LabelCluster: cluster.Name} - if cluster.CompareVersion("2.9.0") >= 0 { - labels = naming.Merge(cluster.Spec.Metadata.GetLabelsOrNil(), labels) - } - root := map[string]any{ // The cluster identifier. This value cannot change during the cluster's // lifetime. "scope": naming.PatroniScope(cluster), - // Use Kubernetes Endpoints for the distributed configuration store (DCS). - // These values cannot change during the cluster's lifetime. - // - // NOTE(cbandy): It *might* be possible to *carefully* change the role and - // scope labels, but there is no way to reconfigure all instances at once. - "kubernetes": map[string]any{ - "namespace": cluster.Namespace, - "role_label": naming.LabelRole, - "scope_label": naming.LabelPatroni, - "use_endpoints": true, - - // In addition to "scope_label" above, Patroni will add the following to - // every object it creates. It will also use these as filters when doing - // any lookups. - "labels": labels, - }, - "postgresql": map[string]any{ // TODO(cbandy): "callbacks" @@ -187,10 +169,30 @@ func clusterYAML( root["postgresql"].(map[string]any)["remove_data_directory_on_diverged_timelines"] = cluster.Spec.Patroni.RemoveDataDirectoryOnDivergedTimelines } + mergeDCSYAML(root, dcsYAML) + b, err := yaml.Marshal(root) return string(append([]byte(yamlGeneratedWarning), b...)), err } +// mergeDCSYAML merges dcsYAML's top-level keys into root. A "postgresql" key +// is shallow-merged into root's existing "postgresql" section instead of +// replacing it, so a DCS backend can contribute settings (e.g. callbacks) +// without clobbering what's already there. Every other key is set directly. +func mergeDCSYAML(root, dcsYAML map[string]any) { + for k, v := range dcsYAML { + if k == "postgresql" { + if addition, ok := v.(map[string]any); ok { + if postgresql, ok := root["postgresql"].(map[string]any); ok { + maps.Copy(postgresql, addition) + continue + } + } + } + root[k] = v + } +} + // DynamicConfiguration combines configuration with some PostgreSQL settings // and returns a value that can be marshaled to JSON. func DynamicConfiguration( @@ -333,12 +335,12 @@ func DynamicConfiguration( } // instanceEnvironment returns the environment variables needed by Patroni's -// instance container. +// instance container. dcsEnvVars are the DCS backend's additions (see +// internal/patroni/dcs), e.g. PATRONI_KUBERNETES_*. func instanceEnvironment( cluster *v1beta1.PostgresCluster, clusterPodService *corev1.Service, - leaderService *corev1.Service, - podContainers []corev1.Container, + dcsEnvVars []corev1.EnvVar, ) []corev1.EnvVar { var ( patroniPort = *cluster.Spec.Patroni.Port @@ -346,31 +348,13 @@ func instanceEnvironment( podSubdomain = clusterPodService.Name ) - // Gather Endpoint ports for any Container ports that match the leader - // Service definition. - ports := []corev1.EndpointPort{} - for _, sp := range leaderService.Spec.Ports { - for i := range podContainers { - for _, cp := range podContainers[i].Ports { - if sp.TargetPort.StrVal == cp.Name { - ports = append(ports, corev1.EndpointPort{ - Name: sp.Name, - Port: cp.ContainerPort, - Protocol: cp.Protocol, - }) - } - } - } - } - portsYAML, _ := yaml.Marshal(ports) - // NOTE(cbandy): Patroni consumes and then removes environment variables // starting with "PATRONI_". // - https://github.com/zalando/patroni/blob/v2.0.2/patroni/config.py#L247 // - https://github.com/zalando/patroni/blob/v2.0.2/patroni/postgresql/postmaster.py#L215-L216 variables := []corev1.EnvVar{ - // Set "name" to the v1.Pod's name. Required when using Kubernetes for DCS. + // Set "name" to the v1.Pod's name. Required for Patroni's node identity. // Patroni must be restarted when changing this value. { Name: "PATRONI_NAME", @@ -380,27 +364,6 @@ func instanceEnvironment( }}, }, - // Set "kubernetes.pod_ip" to the v1.Pod's primary IP address. - // Patroni must be restarted when changing this value. - { - Name: "PATRONI_KUBERNETES_POD_IP", - ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{ - APIVersion: "v1", - FieldPath: "status.podIP", - }}, - }, - - // When using Endpoints for DCS, Patroni needs to replicate the leader - // ServicePort definitions. Set "kubernetes.ports" to the YAML of this - // Pod's equivalent EndpointPort definitions. - // - // This is connascent with PATRONI_POSTGRESQL_CONNECT_ADDRESS below. - // Patroni must be restarted when changing this value. - { - Name: "PATRONI_KUBERNETES_PORTS", - Value: string(portsYAML), - }, - // Set "postgresql.connect_address" using the Pod's stable DNS name. // PostgreSQL must be restarted when changing this value. { @@ -454,7 +417,7 @@ func instanceEnvironment( }, } - return variables + return append(variables, dcsEnvVars...) } // instanceConfigFiles returns projections of Patroni's configuration files @@ -486,25 +449,18 @@ func instanceConfigFiles(cluster, instance *corev1.ConfigMap) []corev1.VolumePro } } -// instanceYAML returns Patroni settings that apply to instance. +// instanceYAML returns Patroni settings that apply to instance. dcsYAML +// contains the DCS backend's config additions (see internal/patroni/dcs). func instanceYAML( cluster *v1beta1.PostgresCluster, instance *v1beta1.PostgresInstanceSetSpec, pgbackrestReplicaCreateCommand []string, + dcsYAML map[string]any, ) (string, error) { root := map[string]any{ // Missing here is "name" which cannot be known until the instance Pod is // created. That value should be injected using the downward API and the // PATRONI_NAME environment variable. - "kubernetes": map[string]any{ - // Missing here is "pod_ip" which cannot be known until the instance Pod is - // created. That value should be injected using the downward API and the - // PATRONI_KUBERNETES_POD_IP environment variable. - - // Missing here is "ports" which is is connascent with "postgresql.connect_address". - // See the PATRONI_KUBERNETES_PORTS env variable. - }, - "restapi": map[string]any{ // Missing here is "connect_address" which cannot be known until the // instance Pod is created. That value should be injected using the downward @@ -665,6 +621,8 @@ func instanceYAML( } } + mergeDCSYAML(root, dcsYAML) + b, err := yaml.Marshal(root) return string(append([]byte(yamlGeneratedWarning), b...)), err } diff --git a/internal/patroni/config_test.go b/internal/patroni/config_test.go index 5cfc026544..2ea7419dac 100644 --- a/internal/patroni/config_test.go +++ b/internal/patroni/config_test.go @@ -27,6 +27,24 @@ import ( "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" ) +// kubernetesClusterYAML stands in for dcs.For(cluster).ClusterYAML(cluster) +// (see internal/patroni/dcs) without importing that package from here. +func kubernetesClusterYAML(cluster *v1beta1.PostgresCluster) map[string]any { + labels := map[string]string{naming.LabelCluster: cluster.Name} + if cluster.CompareVersion("2.9.0") >= 0 { + labels = naming.Merge(cluster.Spec.Metadata.GetLabelsOrNil(), labels) + } + return map[string]any{ + "kubernetes": map[string]any{ + "namespace": cluster.Namespace, + "role_label": naming.LabelRole, + "scope_label": naming.LabelPatroni, + "use_endpoints": true, + "labels": labels, + }, + } +} + func TestClusterYAML(t *testing.T) { t.Parallel() @@ -37,7 +55,7 @@ func TestClusterYAML(t *testing.T) { cluster.Namespace = "some-namespace" cluster.Name = "cluster-name" - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) assert.Equal(t, data, strings.TrimSpace(` # Generated by postgres-operator. DO NOT EDIT UNLESS YOU KNOW WHAT YOU'RE DOING. @@ -104,7 +122,7 @@ watchdog: }, } - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -134,7 +152,7 @@ watchdog: } cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -156,7 +174,7 @@ watchdog: } cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -181,7 +199,7 @@ watchdog: } cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -203,7 +221,7 @@ watchdog: cluster.Spec.PostgresVersion = 17 cluster.Spec.Extensions.PGTDE.Enabled = true - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -230,7 +248,7 @@ watchdog: cluster.Spec.Patroni = &v1beta1.PatroniSpec{} cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -250,7 +268,7 @@ watchdog: cluster.Name = "cluster-name" cluster.Spec.PostgresVersion = 14 - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) assert.NilError(t, err) assert.Equal(t, data, strings.TrimSpace(` # Generated by postgres-operator. DO NOT EDIT UNLESS YOU KNOW WHAT YOU'RE DOING. @@ -300,6 +318,44 @@ watchdog: mode: "off" `)+"\n") }) + + t.Run("dcsYAML merge", func(t *testing.T) { + cluster := new(v1beta1.PostgresCluster) + err := cluster.Default(context.Background(), nil) + assert.NilError(t, err) + cluster.Namespace = "some-namespace" + cluster.Name = "cluster-name" + + // A top-level key is set directly; a "postgresql" key is + // shallow-merged into the existing "postgresql" section instead of + // replacing it (see mergeDCSYAML). + dcsYAML := map[string]any{ + "etcd3": map[string]any{"hosts": []string{"etcd:2379"}}, + "postgresql": map[string]any{ + "callbacks": map[string]any{"on_start": "/some/script.sh"}, + }, + } + + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcsYAML) + assert.NilError(t, err) + + var parsed map[string]any + assert.NilError(t, yaml.Unmarshal([]byte(data), &parsed)) + + etcd3, ok := parsed["etcd3"].(map[string]any) + assert.Assert(t, ok, "expected etcd3 section") + assert.DeepEqual(t, etcd3["hosts"], []any{"etcd:2379"}) + + pgSection, ok := parsed["postgresql"].(map[string]any) + assert.Assert(t, ok, "expected postgresql section") + callbacks, ok := pgSection["callbacks"].(map[string]any) + assert.Assert(t, ok, "expected postgresql.callbacks section") + assert.Equal(t, callbacks["on_start"], "/some/script.sh") + + // Existing postgresql settings survive the shallow-merge. + _, ok = pgSection["authentication"] + assert.Assert(t, ok, "expected existing postgresql.authentication to survive the merge") + }) } func TestDynamicConfiguration(t *testing.T) { @@ -1059,6 +1115,10 @@ func TestInstanceConfigFiles(t *testing.T) { `)) } +// TestInstanceEnvironment covers the generic env vars produced regardless of +// DCS backend, and that a backend's dcsEnvVars pass through unchanged. +// Backend-specific env var content (e.g. PATRONI_KUBERNETES_*) is covered in +// internal/patroni/dcs. func TestInstanceEnvironment(t *testing.T) { t.Parallel() @@ -1066,11 +1126,10 @@ func TestInstanceEnvironment(t *testing.T) { err := cluster.Default(context.Background(), nil) assert.NilError(t, err) cluster.Spec.PostgresVersion = 12 - leaderService := new(corev1.Service) podService := new(corev1.Service) podService.Name = "pod-dns" - vars := instanceEnvironment(cluster, podService, leaderService, nil) + vars := instanceEnvironment(cluster, podService, nil) assert.Assert(t, cmp.MarshalMatches(vars, ` - name: PATRONI_NAME @@ -1078,14 +1137,6 @@ func TestInstanceEnvironment(t *testing.T) { fieldRef: apiVersion: v1 fieldPath: metadata.name -- name: PATRONI_KUBERNETES_POD_IP - valueFrom: - fieldRef: - apiVersion: v1 - fieldPath: status.podIP -- name: PATRONI_KUBERNETES_PORTS - value: | - [] - name: PATRONI_POSTGRESQL_CONNECT_ADDRESS value: $(PATRONI_NAME).pod-dns:5432 - name: PATRONI_POSTGRESQL_LISTEN @@ -1102,15 +1153,13 @@ func TestInstanceEnvironment(t *testing.T) { value: /etc/patroni `)) - t.Run("MatchingPorts", func(t *testing.T) { - leaderService.Spec.Ports = []corev1.ServicePort{{Name: "postgres"}} - leaderService.Spec.Ports[0].TargetPort.StrVal = "postgres" - containers := []corev1.Container{{Name: "okay"}} - containers[0].Ports = []corev1.ContainerPort{{ - Name: "postgres", ContainerPort: 9999, Protocol: corev1.ProtocolTCP, - }} + t.Run("dcsEnvVars pass through", func(t *testing.T) { + dcsEnvVars := []corev1.EnvVar{ + {Name: "PATRONI_KUBERNETES_POD_IP", Value: "1.2.3.4"}, + {Name: "PATRONI_KUBERNETES_PORTS", Value: "[]"}, + } - vars := instanceEnvironment(cluster, podService, leaderService, containers) + vars := instanceEnvironment(cluster, podService, dcsEnvVars) assert.Assert(t, cmp.MarshalMatches(vars, ` - name: PATRONI_NAME @@ -1118,16 +1167,6 @@ func TestInstanceEnvironment(t *testing.T) { fieldRef: apiVersion: v1 fieldPath: metadata.name -- name: PATRONI_KUBERNETES_POD_IP - valueFrom: - fieldRef: - apiVersion: v1 - fieldPath: status.podIP -- name: PATRONI_KUBERNETES_PORTS - value: | - - name: postgres - port: 9999 - protocol: TCP - name: PATRONI_POSTGRESQL_CONNECT_ADDRESS value: $(PATRONI_NAME).pod-dns:5432 - name: PATRONI_POSTGRESQL_LISTEN @@ -1142,6 +1181,10 @@ func TestInstanceEnvironment(t *testing.T) { value: '*:8008' - name: PATRONICTL_CONFIG_FILE value: /etc/patroni +- name: PATRONI_KUBERNETES_POD_IP + value: 1.2.3.4 +- name: PATRONI_KUBERNETES_PORTS + value: '[]' `)) }) } @@ -1159,7 +1202,7 @@ func TestInstanceYAML(t *testing.T) { } instance := new(v1beta1.PostgresInstanceSetSpec) - data, err := instanceYAML(cluster, instance, nil) + data, err := instanceYAML(cluster, instance, nil, nil) assert.NilError(t, err) assert.Equal(t, data, strings.Trim(` # Generated by postgres-operator. DO NOT EDIT UNLESS YOU KNOW WHAT YOU'RE DOING. @@ -1170,7 +1213,6 @@ bootstrap: - encoding=UTF8 - waldir=/pgdata/pg12_wal method: initdb -kubernetes: {} postgresql: basebackup: - waldir=/pgdata/pg12_wal @@ -1182,7 +1224,7 @@ restapi: {} tags: {} `, "\t\n")+"\n") - dataWithReplicaCreate, err := instanceYAML(cluster, instance, []string{"some", "backrest", "cmd"}) + dataWithReplicaCreate, err := instanceYAML(cluster, instance, []string{"some", "backrest", "cmd"}, nil) assert.NilError(t, err) assert.Equal(t, dataWithReplicaCreate, strings.Trim(` # Generated by postgres-operator. DO NOT EDIT UNLESS YOU KNOW WHAT YOU'RE DOING. @@ -1193,7 +1235,6 @@ bootstrap: - encoding=UTF8 - waldir=/pgdata/pg12_wal method: initdb -kubernetes: {} postgresql: basebackup: - waldir=/pgdata/pg12_wal @@ -1214,7 +1255,7 @@ tags: {} cluster.Spec.Patroni = &v1beta1.PatroniSpec{} cluster.Spec.Patroni.CreateReplicaMethods = []v1beta1.CreateReplicaMethod{"basebackup", "pgbackrest"} - dataWithCustomMethods, err := instanceYAML(cluster, instance, nil) + dataWithCustomMethods, err := instanceYAML(cluster, instance, nil, nil) assert.NilError(t, err) assert.Equal(t, dataWithCustomMethods, strings.Trim(` # Generated by postgres-operator. DO NOT EDIT UNLESS YOU KNOW WHAT YOU'RE DOING. @@ -1225,7 +1266,6 @@ bootstrap: - encoding=UTF8 - waldir=/pgdata/pg12_wal method: initdb -kubernetes: {} postgresql: basebackup: - waldir=/pgdata/pg12_wal @@ -1252,7 +1292,7 @@ func TestPGBackRestCreateReplicaCommand(t *testing.T) { } instance := new(v1beta1.PostgresInstanceSetSpec) - data, err := instanceYAML(cluster, instance, []string{"some", "backrest", "cmd"}) + data, err := instanceYAML(cluster, instance, []string{"some", "backrest", "cmd"}, nil) assert.NilError(t, err) var parsed struct { diff --git a/internal/patroni/dcs/dcs.go b/internal/patroni/dcs/dcs.go new file mode 100644 index 0000000000..4d07b2595c --- /dev/null +++ b/internal/patroni/dcs/dcs.go @@ -0,0 +1,101 @@ +// Copyright 2021 - 2024 Crunchy Data Solutions, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +// Package dcs owns everything specific to the distributed configuration +// store (DCS) backend Patroni uses. Generic Patroni logic lives in +// internal/patroni; DCS-specific behavior belongs here. +package dcs + +import ( + "context" + "time" + + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" +) + +// Backend describes the behavior a Patroni DCS backend must provide. +// Implementations are stateless; any dependency they need (a client, an +// executor, an event recorder) is passed as a parameter. +type Backend interface { + // --- Patroni configuration additions --- + + // ClusterYAML returns top-level Patroni config keys owned by this + // backend (e.g. "kubernetes"), merged into the cluster-wide config. + ClusterYAML(cluster *v1beta1.PostgresCluster) map[string]any + + // InstanceYAML returns top-level Patroni config keys owned by this + // backend, merged into each instance's config. + InstanceYAML(cluster *v1beta1.PostgresCluster) map[string]any + + // --- Pod additions --- + + // InstanceEnvVars returns backend-specific environment variables for + // Patroni's instance container. + InstanceEnvVars(cluster *v1beta1.PostgresCluster, + leaderService *corev1.Service, podContainers []corev1.Container) []corev1.EnvVar + + // --- RBAC --- + + // Permissions returns backend-specific RBAC rules for Patroni's + // ServiceAccount, in addition to the generic rules patroni.Permissions + // always grants. + Permissions(cluster *v1beta1.PostgresCluster) []rbacv1.PolicyRule + + // --- Kubernetes objects this backend owns for its own bookkeeping --- + + // DistributedConfigurationService returns the Service this backend needs + // to protect its distributed-configuration objects, or nil when it owns + // no such object. + DistributedConfigurationService(cluster *v1beta1.PostgresCluster) *corev1.Service + + // LeaderLeaseService returns the Service that resolves to the elected + // Patroni leader, or nil when this backend owns no such object. + LeaderLeaseService(cluster *v1beta1.PostgresCluster, + recorder record.EventRecorder) (*corev1.Service, error) + + // PrimaryService returns the ServiceSpec and, if this backend needs the + // operator to manage them itself, EndpointSubset that route traffic to + // cluster's current PostgreSQL primary. leaderService is this backend's + // own LeaderLeaseService result (nil if it has none, or hasn't been + // created yet). endpointSubset is nil when the ServiceSpec's Selector + // already routes traffic on its own (e.g. a future backend using pod + // labels), in which case the operator manages no Endpoints for this + // Service. + PrimaryService(cluster *v1beta1.PostgresCluster, leaderService *corev1.Service) ( + spec corev1.ServiceSpec, endpointSubset *corev1.EndpointSubset, err error) + + // --- Runtime observation --- + + // Observe reports what this backend can tell us about Patroni's runtime + // state this reconcile. readyInstance tells the backend whether any + // instance is currently Ready, since "not bootstrapped yet" vs. + // "bootstrapped but our signal is delayed" needs different requeue + // handling and is backend-specific policy. + Observe(ctx context.Context, cli client.Client, cluster *v1beta1.PostgresCluster, + readyInstance bool) (Observation, error) + + // Delete removes any backend-owned objects during cluster teardown. + Delete(ctx context.Context, cli client.Client, cluster *v1beta1.PostgresCluster) error +} + +// Observation is what a backend learned about Patroni's runtime state on a +// single reconcile pass. +type Observation struct { + // SystemIdentifier is "" when not yet known. + SystemIdentifier string + + // RequeueAfter is 0 when no explicit requeue is needed. + RequeueAfter time.Duration +} + +// For selects the DCS backend for cluster. Only Kubernetes is implemented +// today. +func For(cluster *v1beta1.PostgresCluster) Backend { + return kubernetesBackend{} +} diff --git a/internal/patroni/dcs/kubernetes.go b/internal/patroni/dcs/kubernetes.go new file mode 100644 index 0000000000..c44f820c66 --- /dev/null +++ b/internal/patroni/dcs/kubernetes.go @@ -0,0 +1,335 @@ +// Copyright 2021 - 2024 Crunchy Data Solutions, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +package dcs + +import ( + "context" + "time" + + "github.com/pkg/errors" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/yaml" + + "github.com/percona/percona-postgresql-operator/v2/internal/initialize" + "github.com/percona/percona-postgresql-operator/v2/internal/logging" + "github.com/percona/percona-postgresql-operator/v2/internal/naming" + "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" +) + +// kubernetesBackend uses Kubernetes Endpoints as Patroni's DCS. +type kubernetesBackend struct{} + +func (kubernetesBackend) ClusterYAML(cluster *v1beta1.PostgresCluster) map[string]any { + labels := map[string]string{naming.LabelCluster: cluster.Name} + if cluster.CompareVersion("2.9.0") >= 0 { + labels = naming.Merge(cluster.Spec.Metadata.GetLabelsOrNil(), labels) + } + + return map[string]any{ + // Use Kubernetes Endpoints for the distributed configuration store (DCS). + // These values cannot change during the cluster's lifetime. + // + // NOTE(cbandy): It *might* be possible to *carefully* change the role and + // scope labels, but there is no way to reconfigure all instances at once. + "kubernetes": map[string]any{ + "namespace": cluster.Namespace, + "role_label": naming.LabelRole, + "scope_label": naming.LabelPatroni, + "use_endpoints": true, + + // In addition to "scope_label" above, Patroni will add the following to + // every object it creates. It will also use these as filters when doing + // any lookups. + "labels": labels, + }, + } +} + +func (kubernetesBackend) InstanceYAML(*v1beta1.PostgresCluster) map[string]any { + return nil +} + +func (kubernetesBackend) InstanceEnvVars( + _ *v1beta1.PostgresCluster, leaderService *corev1.Service, podContainers []corev1.Container, +) []corev1.EnvVar { + // "kubernetes.pod_ip" and "kubernetes.ports" cannot be known until the + // instance Pod is created, so they aren't set in InstanceYAML. Instead + // they're injected using the downward API via the + // PATRONI_KUBERNETES_POD_IP and PATRONI_KUBERNETES_PORTS env vars below. + // Gather Endpoint ports for any Container ports that match the leader + // Service definition. + ports := []corev1.EndpointPort{} + for _, sp := range leaderService.Spec.Ports { + for i := range podContainers { + for _, cp := range podContainers[i].Ports { + if sp.TargetPort.StrVal == cp.Name { + ports = append(ports, corev1.EndpointPort{ + Name: sp.Name, + Port: cp.ContainerPort, + Protocol: cp.Protocol, + }) + } + } + } + } + portsYAML, _ := yaml.Marshal(ports) + + return []corev1.EnvVar{ + // Set "kubernetes.pod_ip" to the v1.Pod's primary IP address. + // Patroni must be restarted when changing this value. + { + Name: "PATRONI_KUBERNETES_POD_IP", + ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{ + APIVersion: "v1", + FieldPath: "status.podIP", + }}, + }, + + // When using Endpoints for DCS, Patroni needs to replicate the leader + // ServicePort definitions. Set "kubernetes.ports" to the YAML of this + // Pod's equivalent EndpointPort definitions. + // + // This is connascent with PATRONI_POSTGRESQL_CONNECT_ADDRESS. + // Patroni must be restarted when changing this value. + { + Name: "PATRONI_KUBERNETES_PORTS", + Value: string(portsYAML), + }, + } +} + +// TODO(cbandy): Separate these so that one can choose ConfigMap over Endpoints. + +// When using Endpoints for DCS, "create", "list", "patch", and "watch" are +// required. Include "get" for good measure. The `patronictl scaffold` and +// `patronictl remove` commands require "deletecollection". +// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={get} +// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={create,deletecollection} +// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={list,watch} +// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={patch} +// +kubebuilder:rbac:namespace=patroni,groups="",resources="services",verbs={create} + +// The OpenShift RestrictedEndpointsAdmission plugin requires special +// authorization to create Endpoints that contain Pod IPs. +// - https://github.com/openshift/origin/pull/9383 +// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints/restricted",verbs={create} + +func (kubernetesBackend) Permissions(cluster *v1beta1.PostgresCluster) []rbacv1.PolicyRule { + rules := make([]rbacv1.PolicyRule, 0, 3) + + rules = append(rules, rbacv1.PolicyRule{ + APIGroups: []string{corev1.SchemeGroupVersion.Group}, + Resources: []string{"endpoints"}, + Verbs: []string{"create", "deletecollection", "get", "list", "patch", "watch"}, + }) + + if cluster.Spec.OpenShift != nil && *cluster.Spec.OpenShift { + rules = append(rules, rbacv1.PolicyRule{ + APIGroups: []string{corev1.SchemeGroupVersion.Group}, + Resources: []string{"endpoints/restricted"}, + Verbs: []string{"create"}, + }) + } + + // When using Endpoints for DCS, Patroni tries to create the "{scope}-config" service. + // NOTE(cbandy): The PostgresCluster controller already creates this Service; + // it might be possible to eliminate this permission if it also created the + // Endpoints. + rules = append(rules, rbacv1.PolicyRule{ + APIGroups: []string{corev1.SchemeGroupVersion.Group}, + Resources: []string{"services"}, + Verbs: []string{"create"}, + }) + + return rules +} + +func (kubernetesBackend) DistributedConfigurationService(cluster *v1beta1.PostgresCluster) *corev1.Service { + // When using Endpoints for DCS, Patroni needs a Service to ensure that the + // Endpoints object is not removed by Kubernetes at startup. Patroni will + // create this object if it has permission to do so, but it won't set any + // ownership. + // - https://releases.k8s.io/v1.16.0/pkg/controller/endpoint/endpoints_controller.go#L547 + // - https://releases.k8s.io/v1.20.0/pkg/controller/endpoint/endpoints_controller.go#L580 + // - https://github.com/zalando/patroni/blob/v2.0.1/patroni/dcs/kubernetes.py#L865-L881 + service := &corev1.Service{ObjectMeta: naming.PatroniDistributedConfiguration(cluster)} + service.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Service")) + + // Allocate no IP address (headless) and create no Endpoints. + // - https://docs.k8s.io/concepts/services-networking/service/#headless-services + service.Spec.ClusterIP = corev1.ClusterIPNone + service.Spec.Selector = nil + + return service +} + +func (kubernetesBackend) LeaderLeaseService( + cluster *v1beta1.PostgresCluster, recorder record.EventRecorder, +) (*corev1.Service, error) { + service := &corev1.Service{ObjectMeta: naming.PatroniLeaderEndpoints(cluster)} + service.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("Service")) + + service.Annotations = naming.Merge( + cluster.Spec.Metadata.GetAnnotationsOrNil(), + ) + service.Labels = naming.Merge( + cluster.Spec.Metadata.GetLabelsOrNil(), + ) + + if spec := cluster.Spec.Service; spec != nil { + service.Annotations = naming.Merge(service.Annotations, + spec.Metadata.GetAnnotationsOrNil()) + service.Labels = naming.Merge(service.Labels, + spec.Metadata.GetLabelsOrNil()) + } + + // add our labels last so they aren't overwritten + service.Labels = naming.Merge(service.Labels, + naming.WithPerconaLabels(map[string]string{ // K8SPG-430 + naming.LabelCluster: cluster.Name, + naming.LabelPatroni: naming.PatroniScope(cluster), + }, cluster.Name, "", cluster.Labels[naming.LabelVersion])) + + // Allocate an IP address and/or node port and let Patroni manage the Endpoints. + // Patroni will ensure that they always route to the elected leader. + // - https://docs.k8s.io/concepts/services-networking/service/#services-without-selectors + service.Spec.Selector = nil + + // The TargetPort must be the name (not the number) of the PostgreSQL + // ContainerPort. This name allows the port number to differ between + // instances, which can happen during a rolling update. + servicePort := corev1.ServicePort{ + Name: naming.PortPostgreSQL, + Port: *cluster.Spec.Port, + Protocol: corev1.ProtocolTCP, + TargetPort: intstr.FromString(naming.PortPostgreSQL), + } + + if spec := cluster.Spec.Service; spec == nil { + service.Spec.Type = corev1.ServiceTypeClusterIP + } else { + service.Spec.Type = corev1.ServiceType(spec.Type) + // K8SPG-389 + service.Spec.LoadBalancerSourceRanges = spec.LoadBalancerSourceRanges + + if spec.NodePort != nil { + if service.Spec.Type == corev1.ServiceTypeClusterIP { + // The NodePort can only be set when the Service type is NodePort or + // LoadBalancer. However, due to a known issue prior to Kubernetes + // 1.20, we clear these errors during our apply. To preserve the + // appropriate behavior, we log an Event and return an error. + // TODO(tjmoore4): Once Validation Rules are available, this check + // and event could potentially be removed in favor of that validation + recorder.Eventf(cluster, corev1.EventTypeWarning, "MisconfiguredClusterIP", + "NodePort cannot be set with type ClusterIP on Service %q", service.Name) + return nil, errors.Errorf("NodePort cannot be set with type ClusterIP on Service %q", service.Name) + } + servicePort.NodePort = *spec.NodePort + } + service.Spec.ExternalTrafficPolicy = initialize.FromPointer(spec.ExternalTrafficPolicy) + service.Spec.InternalTrafficPolicy = spec.InternalTrafficPolicy + } + service.Spec.Ports = []corev1.ServicePort{servicePort} + + return service, nil +} + +func (kubernetesBackend) PrimaryService( + cluster *v1beta1.PostgresCluster, leader *corev1.Service, +) (corev1.ServiceSpec, *corev1.EndpointSubset, error) { + // We want to name and label our primary Service consistently. When Patroni is + // using Endpoints for its DCS, however, they and any Service that uses them + // must use the same name as the Patroni "scope" which has its own constraints. + // + // To stay free from those constraints, our primary Service resolves to the + // ClusterIP of the Service created in Reconciler.reconcilePatroniLeaderLease + // when Patroni is using Endpoints. + if leader == nil { + // TODO(cbandy): We need to build a different kind of Service here. + return corev1.ServiceSpec{}, nil, errors.New("Patroni DCS other than Kubernetes Endpoints is not implemented") + } + + // Allocate no IP address (headless) and manage the Endpoints ourselves. + // - https://docs.k8s.io/concepts/services-networking/service/#headless-services + // - https://docs.k8s.io/concepts/services-networking/service/#services-without-selectors + spec := corev1.ServiceSpec{ + ClusterIP: corev1.ClusterIPNone, + Selector: nil, + Ports: []corev1.ServicePort{{ + Name: naming.PortPostgreSQL, + Port: *cluster.Spec.Port, + Protocol: corev1.ProtocolTCP, + TargetPort: intstr.FromString(naming.PortPostgreSQL), + }}, + } + + // Resolve to the ClusterIP for which Patroni has configured the Endpoints. + subset := &corev1.EndpointSubset{ + Addresses: []corev1.EndpointAddress{{IP: leader.Spec.ClusterIP}}, + } + + // Copy the EndpointPorts from the ServicePorts. + for _, sp := range spec.Ports { + subset.Ports = append(subset.Ports, corev1.EndpointPort{ + Name: sp.Name, + Port: sp.Port, + Protocol: sp.Protocol, + }) + } + + return spec, subset, nil +} + +func (kubernetesBackend) Observe( + ctx context.Context, cli client.Client, cluster *v1beta1.PostgresCluster, readyInstance bool, +) (Observation, error) { + var observation Observation + + dcs := &corev1.Endpoints{ObjectMeta: naming.PatroniDistributedConfiguration(cluster)} + err := errors.WithStack(client.IgnoreNotFound( + cli.Get(ctx, client.ObjectKeyFromObject(dcs), dcs), + )) + + if err == nil { + if dcs.Annotations["initialize"] != "" { + // After bootstrap, Patroni writes the cluster system identifier to DCS. + observation.SystemIdentifier = dcs.Annotations["initialize"] + } else if readyInstance { + // While we typically expect a value for the initialize key to be present in the + // Endpoints above by the time the StatefulSet for any instance indicates "ready" + // (since Patroni writes this value after successful cluster bootstrap, at which time + // the initial primary should transition to "ready"), sometimes this is not the case + // and the "initialize" key is not yet present. Therefore, if a "ready" instance + // is detected in the cluster we assume this is the case, and simply log a message and + // requeue in order to try again until the expected value is found. + logging.FromContext(ctx).Info("detected ready instance but no initialize value") + observation.RequeueAfter = time.Second + } + } + + return observation, err +} + +func (kubernetesBackend) Delete(ctx context.Context, cli client.Client, cluster *v1beta1.PostgresCluster) error { + // TODO(cbandy): This could also be accomplished by adopting the Endpoints + // as Patroni creates them. Would their events cause too many reconciles? + // Foreground deletion may force us to adopt and set finalizers anyway. + selector, err := naming.AsSelector(naming.ClusterPatronis(cluster)) + if err == nil { + err = errors.WithStack( + cli.DeleteAllOf( + ctx, &corev1.Endpoints{}, + client.InNamespace(cluster.Namespace), + client.MatchingLabelsSelector{Selector: selector}, + ), + ) + } + + return err +} diff --git a/internal/patroni/dcs/kubernetes_test.go b/internal/patroni/dcs/kubernetes_test.go new file mode 100644 index 0000000000..a60ac2566e --- /dev/null +++ b/internal/patroni/dcs/kubernetes_test.go @@ -0,0 +1,360 @@ +// Copyright 2021 - 2024 Crunchy Data Solutions, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +package dcs + +import ( + "context" + "testing" + "time" + + "gotest.tools/v3/assert" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/percona/percona-postgresql-operator/v2/internal/naming" + "github.com/percona/percona-postgresql-operator/v2/internal/testing/cmp" + "github.com/percona/percona-postgresql-operator/v2/internal/testing/require" + "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" +) + +func TestKubernetesClusterYAML(t *testing.T) { + cluster := new(v1beta1.PostgresCluster) + assert.NilError(t, cluster.Default(context.Background(), nil)) + cluster.Namespace = "some-namespace" + cluster.Name = "cluster-name" + + dcsYAML := (kubernetesBackend{}).ClusterYAML(cluster) + assert.Assert(t, cmp.MarshalMatches(dcsYAML, ` +kubernetes: + labels: + postgres-operator.crunchydata.com/cluster: cluster-name + namespace: some-namespace + role_label: postgres-operator.crunchydata.com/role + scope_label: postgres-operator.crunchydata.com/patroni + use_endpoints: true + `)) +} + +func TestKubernetesInstanceYAML(t *testing.T) { + dcsYAML := (kubernetesBackend{}).InstanceYAML(new(v1beta1.PostgresCluster)) + assert.Assert(t, dcsYAML == nil) +} + +func TestKubernetesInstanceEnvVars(t *testing.T) { + leaderService := new(corev1.Service) + leaderService.Spec.Ports = []corev1.ServicePort{{Name: "postgres"}} + leaderService.Spec.Ports[0].TargetPort.StrVal = "postgres" + containers := []corev1.Container{{Name: "okay"}} + containers[0].Ports = []corev1.ContainerPort{{ + Name: "postgres", ContainerPort: 9999, Protocol: corev1.ProtocolTCP, + }} + + vars := (kubernetesBackend{}).InstanceEnvVars(new(v1beta1.PostgresCluster), leaderService, containers) + + assert.Assert(t, cmp.MarshalMatches(vars, ` +- name: PATRONI_KUBERNETES_POD_IP + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: status.podIP +- name: PATRONI_KUBERNETES_PORTS + value: | + - name: postgres + port: 9999 + protocol: TCP + `)) +} + +func TestKubernetesPermissions(t *testing.T) { + cluster := new(v1beta1.PostgresCluster) + + t.Run("Upstream", func(t *testing.T) { + permissions := (kubernetesBackend{}).Permissions(cluster) + assert.Assert(t, cmp.MarshalMatches(permissions, ` +- apiGroups: + - "" + resources: + - endpoints + verbs: + - create + - deletecollection + - get + - list + - patch + - watch +- apiGroups: + - "" + resources: + - services + verbs: + - create + `)) + }) + + t.Run("OpenShift", func(t *testing.T) { + cluster := cluster.DeepCopy() + cluster.Spec.OpenShift = new(bool) + *cluster.Spec.OpenShift = true + + permissions := (kubernetesBackend{}).Permissions(cluster) + assert.Assert(t, cmp.MarshalMatches(permissions, ` +- apiGroups: + - "" + resources: + - endpoints + verbs: + - create + - deletecollection + - get + - list + - patch + - watch +- apiGroups: + - "" + resources: + - endpoints/restricted + verbs: + - create +- apiGroups: + - "" + resources: + - services + verbs: + - create + `)) + }) +} + +func TestKubernetesDistributedConfigurationService(t *testing.T) { + cluster := new(v1beta1.PostgresCluster) + cluster.Namespace = "ns1" + cluster.Name = "pg1" + + service := (kubernetesBackend{}).DistributedConfigurationService(cluster) + assert.Assert(t, service != nil) + assert.Equal(t, service.Namespace, "ns1") + assert.Equal(t, service.Name, naming.PatroniScope(cluster)+"-config") + assert.Equal(t, service.Spec.ClusterIP, corev1.ClusterIPNone) + assert.Assert(t, service.Spec.Selector == nil, "got %v", service.Spec.Selector) +} + +func TestKubernetesLeaderLeaseService(t *testing.T) { + cluster := &v1beta1.PostgresCluster{} + cluster.Namespace = "ns1" + cluster.Name = "pg2" + cluster.Spec.Port = new(int32(9876)) + cluster.Labels = map[string]string{ + naming.LabelVersion: "2.3.0", + } + + alwaysExpect := func(t testing.TB, service *corev1.Service) { + assert.Assert(t, cmp.MarshalMatches(service.TypeMeta, ` +apiVersion: v1 +kind: Service + `)) + assert.Equal(t, service.Name, "pg2-ha") + assert.Equal(t, service.Namespace, "ns1") + + // Always gets a ClusterIP (never None). + assert.Equal(t, service.Spec.ClusterIP, "") + assert.Assert(t, service.Spec.Selector == nil, + "got %v", service.Spec.Selector) + } + + t.Run("NoServiceSpec", func(t *testing.T) { + service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + assert.NilError(t, err) + alwaysExpect(t, service) + // Defaults to ClusterIP. + assert.Equal(t, service.Spec.Type, corev1.ServiceTypeClusterIP) + assert.Assert(t, cmp.MarshalMatches(service.Spec.Ports, ` +- name: postgres + port: 9876 + protocol: TCP + targetPort: postgres + `)) + }) + + t.Run("AnnotationsLabels", func(t *testing.T) { + cluster := cluster.DeepCopy() + cluster.Spec.Metadata = &v1beta1.Metadata{ + Annotations: map[string]string{"a": "v1"}, + Labels: map[string]string{"b": "v2"}, + } + + service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + assert.NilError(t, err) + + assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ + "a": "v1", + }) + assert.DeepEqual(t, service.ObjectMeta.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ + "b": "v2", + "postgres-operator.crunchydata.com/cluster": "pg2", + "postgres-operator.crunchydata.com/patroni": "pg2-ha", + }, "pg2", "", "2.3.0"))) + + // Add metadata to individual service + cluster.Spec.Service = &v1beta1.ServiceSpec{ + Metadata: &v1beta1.Metadata{ + Annotations: map[string]string{"c": "v3"}, + Labels: map[string]string{"d": "v4", + "postgres-operator.crunchydata.com/cluster": "wrongName"}, + }, + } + + service, err = (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + assert.NilError(t, err) + + assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ + "a": "v1", + "c": "v3", + }) + assert.DeepEqual(t, service.ObjectMeta.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ + "b": "v2", + "d": "v4", + "postgres-operator.crunchydata.com/cluster": "pg2", + "postgres-operator.crunchydata.com/patroni": "pg2-ha", + }, "pg2", "", "2.3.0"))) + }) + + types := []struct { + Type string + Expect func(testing.TB, *corev1.Service) + }{ + {Type: "ClusterIP", Expect: func(t testing.TB, service *corev1.Service) { + assert.Equal(t, service.Spec.Type, corev1.ServiceTypeClusterIP) + }}, + {Type: "NodePort", Expect: func(t testing.TB, service *corev1.Service) { + assert.Equal(t, service.Spec.Type, corev1.ServiceTypeNodePort) + }}, + {Type: "LoadBalancer", Expect: func(t testing.TB, service *corev1.Service) { + assert.Equal(t, service.Spec.Type, corev1.ServiceTypeLoadBalancer) + }}, + } + + for _, test := range types { + t.Run(test.Type, func(t *testing.T) { + cluster := cluster.DeepCopy() + cluster.Spec.Service = &v1beta1.ServiceSpec{Type: test.Type} + + service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + assert.NilError(t, err) + alwaysExpect(t, service) + test.Expect(t, service) + }) + } + + t.Run("NodePortWithClusterIP", func(t *testing.T) { + cluster := cluster.DeepCopy() + cluster.Spec.Service = &v1beta1.ServiceSpec{Type: "ClusterIP", NodePort: new(int32(32000))} + + recorder := new(record.FakeRecorder) + service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, recorder) + assert.ErrorContains(t, err, `NodePort cannot be set with type ClusterIP on Service "pg2-ha"`) + assert.Assert(t, service == nil) + }) +} + +func TestKubernetesPrimaryService(t *testing.T) { + cluster := new(v1beta1.PostgresCluster) + cluster.Spec.Port = new(int32(2600)) + + t.Run("NoLeader", func(t *testing.T) { + spec, subset, err := (kubernetesBackend{}).PrimaryService(cluster, nil) + assert.ErrorContains(t, err, "not implemented") + assert.DeepEqual(t, spec, corev1.ServiceSpec{}) + assert.Assert(t, subset == nil) + }) + + t.Run("ResolvesToLeaderClusterIP", func(t *testing.T) { + leader := &corev1.Service{} + leader.Spec.ClusterIP = "1.9.8.3" + + spec, subset, err := (kubernetesBackend{}).PrimaryService(cluster, leader) + assert.NilError(t, err) + + assert.Equal(t, spec.ClusterIP, corev1.ClusterIPNone) + assert.Assert(t, spec.Selector == nil, "got %v", spec.Selector) + assert.Assert(t, cmp.MarshalMatches(spec.Ports, ` +- name: postgres + port: 2600 + protocol: TCP + targetPort: postgres + `)) + + assert.Assert(t, subset != nil) + assert.Assert(t, cmp.MarshalMatches(subset, ` +addresses: +- ip: 1.9.8.3 +ports: +- name: postgres + port: 2600 + protocol: TCP + `)) + }) +} + +func TestKubernetesObserve(t *testing.T) { + _, cc := require.Kubernetes2(t) + require.ParallelCapacity(t, 0) + ns := require.Namespace(t, cc) + ctx := context.Background() + + cluster := new(v1beta1.PostgresCluster) + cluster.Namespace = ns.Name + cluster.Name = "observe-test" + + t.Run("NotFound, not ready", func(t *testing.T) { + observation, err := (kubernetesBackend{}).Observe(ctx, cc, cluster, false) + assert.NilError(t, err) + assert.Equal(t, observation.SystemIdentifier, "") + assert.Equal(t, observation.RequeueAfter, time.Duration(0)) + }) + + t.Run("NotFound, ready", func(t *testing.T) { + observation, err := (kubernetesBackend{}).Observe(ctx, cc, cluster, true) + assert.NilError(t, err) + assert.Equal(t, observation.SystemIdentifier, "") + assert.Equal(t, observation.RequeueAfter, time.Second) + }) + + t.Run("initialize annotation present", func(t *testing.T) { + endpoints := &corev1.Endpoints{ObjectMeta: naming.PatroniDistributedConfiguration(cluster)} + endpoints.Annotations = map[string]string{"initialize": "123456"} + assert.NilError(t, cc.Create(ctx, endpoints)) + t.Cleanup(func() { assert.Check(t, client.IgnoreNotFound(cc.Delete(ctx, endpoints))) }) + + observation, err := (kubernetesBackend{}).Observe(ctx, cc, cluster, false) + assert.NilError(t, err) + assert.Equal(t, observation.SystemIdentifier, "123456") + assert.Equal(t, observation.RequeueAfter, time.Duration(0)) + }) +} + +func TestKubernetesDelete(t *testing.T) { + _, cc := require.Kubernetes2(t) + require.ParallelCapacity(t, 0) + ns := require.Namespace(t, cc) + ctx := context.Background() + + cluster := new(v1beta1.PostgresCluster) + cluster.Namespace = ns.Name + cluster.Name = "delete-test" + + endpoints := &corev1.Endpoints{ObjectMeta: naming.PatroniDistributedConfiguration(cluster)} + endpoints.Labels = map[string]string{ + naming.LabelCluster: cluster.Name, + naming.LabelPatroni: naming.PatroniScope(cluster), + } + assert.NilError(t, cc.Create(ctx, endpoints)) + + assert.NilError(t, (kubernetesBackend{}).Delete(ctx, cc, cluster)) + + err := cc.Get(ctx, client.ObjectKeyFromObject(endpoints), endpoints) + assert.Assert(t, apierrors.IsNotFound(err), "expected the Endpoints to be deleted, got %v", err) +} diff --git a/internal/patroni/rbac.go b/internal/patroni/rbac.go index 2e19339ed9..c105636768 100644 --- a/internal/patroni/rbac.go +++ b/internal/patroni/rbac.go @@ -16,57 +16,12 @@ import ( // +kubebuilder:rbac:namespace=patroni,groups="",resources="pods",verbs={list,watch} // +kubebuilder:rbac:namespace=patroni,groups="",resources="pods",verbs={patch} -// TODO(cbandy): Separate these so that one can choose ConfigMap over Endpoints. - -// When using Endpoints for DCS, "create", "list", "patch", and "watch" are -// required. Include "get" for good measure. The `patronictl scaffold` and -// `patronictl remove` commands require "deletecollection". -// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={get} -// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={create,deletecollection} -// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={list,watch} -// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints",verbs={patch} -// +kubebuilder:rbac:namespace=patroni,groups="",resources="services",verbs={create} - -// The OpenShift RestrictedEndpointsAdmission plugin requires special -// authorization to create Endpoints that contain Pod IPs. -// - https://github.com/openshift/origin/pull/9383 -// +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints/restricted",verbs={create} - -// Permissions returns the RBAC rules Patroni needs for cluster. -func Permissions(cluster *v1beta1.PostgresCluster) []rbacv1.PolicyRule { - // TODO(cbandy): This must change when using ConfigMaps for DCS. - - rules := make([]rbacv1.PolicyRule, 0, 4) - - rules = append(rules, rbacv1.PolicyRule{ - APIGroups: []string{corev1.SchemeGroupVersion.Group}, - Resources: []string{"endpoints"}, - Verbs: []string{"create", "deletecollection", "get", "list", "patch", "watch"}, - }) - - if cluster.Spec.OpenShift != nil && *cluster.Spec.OpenShift { - rules = append(rules, rbacv1.PolicyRule{ - APIGroups: []string{corev1.SchemeGroupVersion.Group}, - Resources: []string{"endpoints/restricted"}, - Verbs: []string{"create"}, - }) - } - - rules = append(rules, rbacv1.PolicyRule{ +// Permissions returns the RBAC rules Patroni needs for cluster, regardless +// of DCS backend. See internal/patroni/dcs for backend-specific rules. +func Permissions(*v1beta1.PostgresCluster) []rbacv1.PolicyRule { + return []rbacv1.PolicyRule{{ APIGroups: []string{corev1.SchemeGroupVersion.Group}, Resources: []string{"pods"}, Verbs: []string{"get", "list", "patch", "watch"}, - }) - - // When using Endpoints for DCS, Patroni tries to create the "{scope}-config" service. - // NOTE(cbandy): The PostgresCluster controller already creates this Service; - // it might be possible to eliminate this permission if it also created the - // Endpoints. - rules = append(rules, rbacv1.PolicyRule{ - APIGroups: []string{corev1.SchemeGroupVersion.Group}, - Resources: []string{"services"}, - Verbs: []string{"create"}, - }) - - return rules + }} } diff --git a/internal/patroni/rbac_test.go b/internal/patroni/rbac_test.go index 4b45aa1f94..2039f498d8 100644 --- a/internal/patroni/rbac_test.go +++ b/internal/patroni/rbac_test.go @@ -27,78 +27,21 @@ func isUniqueAndSorted(slice []string) bool { return true } +// TestPermissions covers the generic RBAC rules Patroni needs regardless of +// DCS backend. See internal/patroni/dcs for backend-specific rules. func TestPermissions(t *testing.T) { cluster := new(v1beta1.PostgresCluster) err := cluster.Default(context.Background(), nil) assert.NilError(t, err) - t.Run("Upstream", func(t *testing.T) { - permissions := Permissions(cluster) - for _, rule := range permissions { - assert.Assert(t, isUniqueAndSorted(rule.APIGroups), "got %q", rule.APIGroups) - assert.Assert(t, isUniqueAndSorted(rule.Resources), "got %q", rule.Resources) - assert.Assert(t, isUniqueAndSorted(rule.Verbs), "got %q", rule.Verbs) - } - - assert.Assert(t, cmp.MarshalMatches(permissions, ` -- apiGroups: - - "" - resources: - - endpoints - verbs: - - create - - deletecollection - - get - - list - - patch - - watch -- apiGroups: - - "" - resources: - - pods - verbs: - - get - - list - - patch - - watch -- apiGroups: - - "" - resources: - - services - verbs: - - create - `)) - }) - - t.Run("OpenShift", func(t *testing.T) { - cluster.Spec.OpenShift = new(bool) - *cluster.Spec.OpenShift = true - - permissions := Permissions(cluster) - for _, rule := range permissions { - assert.Assert(t, isUniqueAndSorted(rule.APIGroups), "got %q", rule.APIGroups) - assert.Assert(t, isUniqueAndSorted(rule.Resources), "got %q", rule.Resources) - assert.Assert(t, isUniqueAndSorted(rule.Verbs), "got %q", rule.Verbs) - } + permissions := Permissions(cluster) + for _, rule := range permissions { + assert.Assert(t, isUniqueAndSorted(rule.APIGroups), "got %q", rule.APIGroups) + assert.Assert(t, isUniqueAndSorted(rule.Resources), "got %q", rule.Resources) + assert.Assert(t, isUniqueAndSorted(rule.Verbs), "got %q", rule.Verbs) + } - assert.Assert(t, cmp.MarshalMatches(permissions, ` -- apiGroups: - - "" - resources: - - endpoints - verbs: - - create - - deletecollection - - get - - list - - patch - - watch -- apiGroups: - - "" - resources: - - endpoints/restricted - verbs: - - create + assert.Assert(t, cmp.MarshalMatches(permissions, ` - apiGroups: - "" resources: @@ -108,12 +51,5 @@ func TestPermissions(t *testing.T) { - list - patch - watch -- apiGroups: - - "" - resources: - - services - verbs: - - create - `)) - }) + `)) } diff --git a/internal/patroni/reconcile.go b/internal/patroni/reconcile.go index ece6cea757..c29b0dc1e4 100644 --- a/internal/patroni/reconcile.go +++ b/internal/patroni/reconcile.go @@ -28,11 +28,14 @@ func ClusterBootstrapped(postgresCluster *v1beta1.PostgresCluster) bool { return postgresCluster.Status.Patroni.SystemIdentifier != "" } -// ClusterConfigMap populates the shared ConfigMap with fields needed to run Patroni. +// ClusterConfigMap populates the shared ConfigMap with fields needed to run +// Patroni. dcsYAML is the DCS backend's config additions (see +// internal/patroni/dcs). func ClusterConfigMap(ctx context.Context, inCluster *v1beta1.PostgresCluster, inHBAs postgres.HBAs, inParameters postgres.Parameters, + dcsYAML map[string]any, outClusterConfigMap *corev1.ConfigMap, ) error { var err error @@ -40,15 +43,18 @@ func ClusterConfigMap(ctx context.Context, initialize.Map(&outClusterConfigMap.Data) outClusterConfigMap.Data[configMapFileKey], err = clusterYAML(inCluster, inHBAs, - inParameters) + inParameters, dcsYAML) return err } -// InstanceConfigMap populates the shared ConfigMap with fields needed to run Patroni. +// InstanceConfigMap populates the shared ConfigMap with fields needed to run +// Patroni. dcsYAML is the DCS backend's config additions (see +// internal/patroni/dcs). func InstanceConfigMap(ctx context.Context, inCluster *v1beta1.PostgresCluster, inInstanceSpec *v1beta1.PostgresInstanceSetSpec, + dcsYAML map[string]any, outInstanceConfigMap *corev1.ConfigMap, ) error { var err error @@ -58,7 +64,7 @@ func InstanceConfigMap(ctx context.Context, command := pgbackrest.ReplicaCreateCommand(inCluster, inInstanceSpec) outInstanceConfigMap.Data[configMapFileKey], err = instanceYAML( - inCluster, inInstanceSpec, command) + inCluster, inInstanceSpec, command, dcsYAML) return err } @@ -81,12 +87,13 @@ func InstanceCertificates(ctx context.Context, } // InstancePod populates a PodTemplateSpec with the fields needed to run Patroni. -// The database container must already be in the template. +// The database container must already be in the template. dcsEnvVars are the +// DCS backend's additions (see internal/patroni/dcs). func InstancePod(ctx context.Context, inCluster *v1beta1.PostgresCluster, inClusterConfigMap *corev1.ConfigMap, inClusterPodService *corev1.Service, - inPatroniLeaderService *corev1.Service, + dcsEnvVars []corev1.EnvVar, inInstanceSpec *v1beta1.PostgresInstanceSetSpec, inInstanceCertificates *corev1.Secret, inInstanceConfigMap *corev1.ConfigMap, @@ -114,8 +121,7 @@ func InstancePod(ctx context.Context, } container.Env = append(container.Env, - instanceEnvironment(inCluster, inClusterPodService, inPatroniLeaderService, - outInstancePod.Spec.Containers)...) + instanceEnvironment(inCluster, inClusterPodService, dcsEnvVars)...) volume := corev1.Volume{Name: "patroni-config"} volume.Projected = new(corev1.ProjectedVolumeSource) diff --git a/internal/patroni/reconcile_test.go b/internal/patroni/reconcile_test.go index e7f2a239a5..9e74aedd38 100644 --- a/internal/patroni/reconcile_test.go +++ b/internal/patroni/reconcile_test.go @@ -28,18 +28,21 @@ func TestClusterConfigMap(t *testing.T) { pgHBAs := postgres.HBAs{} pgParameters := postgres.Parameters{} + dcsYAML := map[string]any{"kubernetes": map[string]any{"example": true}} + err := cluster.Default(context.Background(), nil) assert.NilError(t, err) config := new(corev1.ConfigMap) - assert.NilError(t, ClusterConfigMap(ctx, cluster, pgHBAs, pgParameters, config)) + assert.NilError(t, ClusterConfigMap(ctx, cluster, pgHBAs, pgParameters, dcsYAML, config)) // The output of clusterYAML should go into config. - data, _ := clusterYAML(cluster, pgHBAs, pgParameters) + data, _ := clusterYAML(cluster, pgHBAs, pgParameters, dcsYAML) assert.DeepEqual(t, config.Data["patroni.yaml"], data) + assert.Assert(t, cmp.Contains(config.Data["patroni.yaml"], "example: true")) // No change when called again. before := config.DeepCopy() - assert.NilError(t, ClusterConfigMap(ctx, cluster, pgHBAs, pgParameters, config)) + assert.NilError(t, ClusterConfigMap(ctx, cluster, pgHBAs, pgParameters, dcsYAML, config)) assert.DeepEqual(t, config, before) } @@ -138,15 +141,16 @@ func TestInstanceConfigMap(t *testing.T) { cluster := new(v1beta1.PostgresCluster) instance := new(v1beta1.PostgresInstanceSetSpec) config := new(corev1.ConfigMap) - data, _ := instanceYAML(cluster, instance, nil) + dcsYAML := map[string]any{"kubernetes": map[string]any{}} + data, _ := instanceYAML(cluster, instance, nil, dcsYAML) - assert.NilError(t, InstanceConfigMap(ctx, cluster, instance, config)) + assert.NilError(t, InstanceConfigMap(ctx, cluster, instance, dcsYAML, config)) assert.DeepEqual(t, config.Data["patroni.yaml"], data) // No change when called again. before := config.DeepCopy() - assert.NilError(t, InstanceConfigMap(ctx, cluster, instance, config)) + assert.NilError(t, InstanceConfigMap(ctx, cluster, instance, dcsYAML, config)) assert.DeepEqual(t, config, before) } @@ -171,14 +175,6 @@ containers: fieldRef: apiVersion: v1 fieldPath: metadata.name - - name: PATRONI_KUBERNETES_POD_IP - valueFrom: - fieldRef: - apiVersion: v1 - fieldPath: status.podIP - - name: PATRONI_KUBERNETES_PORTS - value: | - [] - name: PATRONI_POSTGRESQL_CONNECT_ADDRESS value: $(PATRONI_NAME).:5432 - name: PATRONI_POSTGRESQL_LISTEN @@ -193,6 +189,14 @@ containers: value: '*:8008' - name: PATRONICTL_CONFIG_FILE value: /etc/patroni + - name: PATRONI_KUBERNETES_POD_IP + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: status.podIP + - name: PATRONI_KUBERNETES_PORTS + value: | + [] livenessProbe: exec: command: @@ -285,14 +289,6 @@ containers: fieldRef: apiVersion: v1 fieldPath: metadata.name - - name: PATRONI_KUBERNETES_POD_IP - valueFrom: - fieldRef: - apiVersion: v1 - fieldPath: status.podIP - - name: PATRONI_KUBERNETES_PORTS - value: | - [] - name: PATRONI_POSTGRESQL_CONNECT_ADDRESS value: $(PATRONI_NAME).:5432 - name: PATRONI_POSTGRESQL_LISTEN @@ -307,6 +303,14 @@ containers: value: '*:8008' - name: PATRONICTL_CONFIG_FILE value: /etc/patroni + - name: PATRONI_KUBERNETES_POD_IP + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: status.podIP + - name: PATRONI_KUBERNETES_PORTS + value: | + [] livenessProbe: failureThreshold: 3 httpGet: @@ -369,14 +373,26 @@ volumes: instanceCertificates := new(corev1.Secret) instanceConfigMap := new(corev1.ConfigMap) instanceSpec := new(v1beta1.PostgresInstanceSetSpec) - patroniLeaderService := new(corev1.Service) template := new(corev1.PodTemplateSpec) template.Spec.Containers = []corev1.Container{{Name: "database"}} cluster.Labels = tt.labels + // Fixture standing in for a DCS backend's env vars (see + // internal/patroni/dcs), e.g. Kubernetes' PATRONI_KUBERNETES_*. + dcsEnvVars := []corev1.EnvVar{ + { + Name: "PATRONI_KUBERNETES_POD_IP", + ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{ + APIVersion: "v1", + FieldPath: "status.podIP", + }}, + }, + {Name: "PATRONI_KUBERNETES_PORTS", Value: "[]\n"}, + } + call := func() error { return InstancePod(context.Background(), - cluster, clusterConfigMap, clusterPodService, patroniLeaderService, + cluster, clusterConfigMap, clusterPodService, dcsEnvVars, instanceSpec, instanceCertificates, instanceConfigMap, template, initImage) } assert.NilError(t, call()) From 7438cf5e1b69ea914e17f88d7d78ec1cb895e41b Mon Sep 17 00:00:00 2001 From: yoav-katz Date: Wed, 5 Aug 2026 18:25:36 +0300 Subject: [PATCH 2/5] renammed dcs backend from kubernetes to kubernetes_endpoints --- internal/patroni/dcs/dcs.go | 7 +-- ...{kubernetes.go => kubernetes_endpoints.go} | 26 +++++----- ...s_test.go => kubernetes_endpoints_test.go} | 52 +++++++++---------- 3 files changed, 43 insertions(+), 42 deletions(-) rename internal/patroni/dcs/{kubernetes.go => kubernetes_endpoints.go} (92%) rename internal/patroni/dcs/{kubernetes_test.go => kubernetes_endpoints_test.go} (81%) diff --git a/internal/patroni/dcs/dcs.go b/internal/patroni/dcs/dcs.go index 4d07b2595c..22e87b725c 100644 --- a/internal/patroni/dcs/dcs.go +++ b/internal/patroni/dcs/dcs.go @@ -94,8 +94,9 @@ type Observation struct { RequeueAfter time.Duration } -// For selects the DCS backend for cluster. Only Kubernetes is implemented -// today. +// For selects the DCS backend for cluster. Only Kubernetes Endpoints is +// implemented today; a future backend (e.g. Kubernetes ConfigMaps, etcd) +// adds a case here. func For(cluster *v1beta1.PostgresCluster) Backend { - return kubernetesBackend{} + return kubernetesEndpointsBackend{} } diff --git a/internal/patroni/dcs/kubernetes.go b/internal/patroni/dcs/kubernetes_endpoints.go similarity index 92% rename from internal/patroni/dcs/kubernetes.go rename to internal/patroni/dcs/kubernetes_endpoints.go index c44f820c66..f052e31d6a 100644 --- a/internal/patroni/dcs/kubernetes.go +++ b/internal/patroni/dcs/kubernetes_endpoints.go @@ -22,10 +22,12 @@ import ( "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" ) -// kubernetesBackend uses Kubernetes Endpoints as Patroni's DCS. -type kubernetesBackend struct{} +// kubernetesEndpointsBackend uses Kubernetes Endpoints as Patroni's DCS. +// This is distinct from a future ConfigMaps-based Kubernetes backend, which +// Patroni also supports. +type kubernetesEndpointsBackend struct{} -func (kubernetesBackend) ClusterYAML(cluster *v1beta1.PostgresCluster) map[string]any { +func (kubernetesEndpointsBackend) ClusterYAML(cluster *v1beta1.PostgresCluster) map[string]any { labels := map[string]string{naming.LabelCluster: cluster.Name} if cluster.CompareVersion("2.9.0") >= 0 { labels = naming.Merge(cluster.Spec.Metadata.GetLabelsOrNil(), labels) @@ -51,11 +53,11 @@ func (kubernetesBackend) ClusterYAML(cluster *v1beta1.PostgresCluster) map[strin } } -func (kubernetesBackend) InstanceYAML(*v1beta1.PostgresCluster) map[string]any { +func (kubernetesEndpointsBackend) InstanceYAML(*v1beta1.PostgresCluster) map[string]any { return nil } -func (kubernetesBackend) InstanceEnvVars( +func (kubernetesEndpointsBackend) InstanceEnvVars( _ *v1beta1.PostgresCluster, leaderService *corev1.Service, podContainers []corev1.Container, ) []corev1.EnvVar { // "kubernetes.pod_ip" and "kubernetes.ports" cannot be known until the @@ -104,8 +106,6 @@ func (kubernetesBackend) InstanceEnvVars( } } -// TODO(cbandy): Separate these so that one can choose ConfigMap over Endpoints. - // When using Endpoints for DCS, "create", "list", "patch", and "watch" are // required. Include "get" for good measure. The `patronictl scaffold` and // `patronictl remove` commands require "deletecollection". @@ -120,7 +120,7 @@ func (kubernetesBackend) InstanceEnvVars( // - https://github.com/openshift/origin/pull/9383 // +kubebuilder:rbac:namespace=patroni,groups="",resources="endpoints/restricted",verbs={create} -func (kubernetesBackend) Permissions(cluster *v1beta1.PostgresCluster) []rbacv1.PolicyRule { +func (kubernetesEndpointsBackend) Permissions(cluster *v1beta1.PostgresCluster) []rbacv1.PolicyRule { rules := make([]rbacv1.PolicyRule, 0, 3) rules = append(rules, rbacv1.PolicyRule{ @@ -150,7 +150,7 @@ func (kubernetesBackend) Permissions(cluster *v1beta1.PostgresCluster) []rbacv1. return rules } -func (kubernetesBackend) DistributedConfigurationService(cluster *v1beta1.PostgresCluster) *corev1.Service { +func (kubernetesEndpointsBackend) DistributedConfigurationService(cluster *v1beta1.PostgresCluster) *corev1.Service { // When using Endpoints for DCS, Patroni needs a Service to ensure that the // Endpoints object is not removed by Kubernetes at startup. Patroni will // create this object if it has permission to do so, but it won't set any @@ -169,7 +169,7 @@ func (kubernetesBackend) DistributedConfigurationService(cluster *v1beta1.Postgr return service } -func (kubernetesBackend) LeaderLeaseService( +func (kubernetesEndpointsBackend) LeaderLeaseService( cluster *v1beta1.PostgresCluster, recorder record.EventRecorder, ) (*corev1.Service, error) { service := &corev1.Service{ObjectMeta: naming.PatroniLeaderEndpoints(cluster)} @@ -240,7 +240,7 @@ func (kubernetesBackend) LeaderLeaseService( return service, nil } -func (kubernetesBackend) PrimaryService( +func (kubernetesEndpointsBackend) PrimaryService( cluster *v1beta1.PostgresCluster, leader *corev1.Service, ) (corev1.ServiceSpec, *corev1.EndpointSubset, error) { // We want to name and label our primary Service consistently. When Patroni is @@ -286,7 +286,7 @@ func (kubernetesBackend) PrimaryService( return spec, subset, nil } -func (kubernetesBackend) Observe( +func (kubernetesEndpointsBackend) Observe( ctx context.Context, cli client.Client, cluster *v1beta1.PostgresCluster, readyInstance bool, ) (Observation, error) { var observation Observation @@ -316,7 +316,7 @@ func (kubernetesBackend) Observe( return observation, err } -func (kubernetesBackend) Delete(ctx context.Context, cli client.Client, cluster *v1beta1.PostgresCluster) error { +func (kubernetesEndpointsBackend) Delete(ctx context.Context, cli client.Client, cluster *v1beta1.PostgresCluster) error { // TODO(cbandy): This could also be accomplished by adopting the Endpoints // as Patroni creates them. Would their events cause too many reconciles? // Foreground deletion may force us to adopt and set finalizers anyway. diff --git a/internal/patroni/dcs/kubernetes_test.go b/internal/patroni/dcs/kubernetes_endpoints_test.go similarity index 81% rename from internal/patroni/dcs/kubernetes_test.go rename to internal/patroni/dcs/kubernetes_endpoints_test.go index a60ac2566e..7745155eb6 100644 --- a/internal/patroni/dcs/kubernetes_test.go +++ b/internal/patroni/dcs/kubernetes_endpoints_test.go @@ -21,13 +21,13 @@ import ( "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" ) -func TestKubernetesClusterYAML(t *testing.T) { +func TestKubernetesEndpointsClusterYAML(t *testing.T) { cluster := new(v1beta1.PostgresCluster) assert.NilError(t, cluster.Default(context.Background(), nil)) cluster.Namespace = "some-namespace" cluster.Name = "cluster-name" - dcsYAML := (kubernetesBackend{}).ClusterYAML(cluster) + dcsYAML := (kubernetesEndpointsBackend{}).ClusterYAML(cluster) assert.Assert(t, cmp.MarshalMatches(dcsYAML, ` kubernetes: labels: @@ -39,12 +39,12 @@ kubernetes: `)) } -func TestKubernetesInstanceYAML(t *testing.T) { - dcsYAML := (kubernetesBackend{}).InstanceYAML(new(v1beta1.PostgresCluster)) +func TestKubernetesEndpointsInstanceYAML(t *testing.T) { + dcsYAML := (kubernetesEndpointsBackend{}).InstanceYAML(new(v1beta1.PostgresCluster)) assert.Assert(t, dcsYAML == nil) } -func TestKubernetesInstanceEnvVars(t *testing.T) { +func TestKubernetesEndpointsInstanceEnvVars(t *testing.T) { leaderService := new(corev1.Service) leaderService.Spec.Ports = []corev1.ServicePort{{Name: "postgres"}} leaderService.Spec.Ports[0].TargetPort.StrVal = "postgres" @@ -53,7 +53,7 @@ func TestKubernetesInstanceEnvVars(t *testing.T) { Name: "postgres", ContainerPort: 9999, Protocol: corev1.ProtocolTCP, }} - vars := (kubernetesBackend{}).InstanceEnvVars(new(v1beta1.PostgresCluster), leaderService, containers) + vars := (kubernetesEndpointsBackend{}).InstanceEnvVars(new(v1beta1.PostgresCluster), leaderService, containers) assert.Assert(t, cmp.MarshalMatches(vars, ` - name: PATRONI_KUBERNETES_POD_IP @@ -69,11 +69,11 @@ func TestKubernetesInstanceEnvVars(t *testing.T) { `)) } -func TestKubernetesPermissions(t *testing.T) { +func TestKubernetesEndpointsPermissions(t *testing.T) { cluster := new(v1beta1.PostgresCluster) t.Run("Upstream", func(t *testing.T) { - permissions := (kubernetesBackend{}).Permissions(cluster) + permissions := (kubernetesEndpointsBackend{}).Permissions(cluster) assert.Assert(t, cmp.MarshalMatches(permissions, ` - apiGroups: - "" @@ -100,7 +100,7 @@ func TestKubernetesPermissions(t *testing.T) { cluster.Spec.OpenShift = new(bool) *cluster.Spec.OpenShift = true - permissions := (kubernetesBackend{}).Permissions(cluster) + permissions := (kubernetesEndpointsBackend{}).Permissions(cluster) assert.Assert(t, cmp.MarshalMatches(permissions, ` - apiGroups: - "" @@ -129,12 +129,12 @@ func TestKubernetesPermissions(t *testing.T) { }) } -func TestKubernetesDistributedConfigurationService(t *testing.T) { +func TestKubernetesEndpointsDistributedConfigurationService(t *testing.T) { cluster := new(v1beta1.PostgresCluster) cluster.Namespace = "ns1" cluster.Name = "pg1" - service := (kubernetesBackend{}).DistributedConfigurationService(cluster) + service := (kubernetesEndpointsBackend{}).DistributedConfigurationService(cluster) assert.Assert(t, service != nil) assert.Equal(t, service.Namespace, "ns1") assert.Equal(t, service.Name, naming.PatroniScope(cluster)+"-config") @@ -142,7 +142,7 @@ func TestKubernetesDistributedConfigurationService(t *testing.T) { assert.Assert(t, service.Spec.Selector == nil, "got %v", service.Spec.Selector) } -func TestKubernetesLeaderLeaseService(t *testing.T) { +func TestKubernetesEndpointsLeaderLeaseService(t *testing.T) { cluster := &v1beta1.PostgresCluster{} cluster.Namespace = "ns1" cluster.Name = "pg2" @@ -166,7 +166,7 @@ kind: Service } t.Run("NoServiceSpec", func(t *testing.T) { - service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + service, err := (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) assert.NilError(t, err) alwaysExpect(t, service) // Defaults to ClusterIP. @@ -186,7 +186,7 @@ kind: Service Labels: map[string]string{"b": "v2"}, } - service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + service, err := (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) assert.NilError(t, err) assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ @@ -207,7 +207,7 @@ kind: Service }, } - service, err = (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + service, err = (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) assert.NilError(t, err) assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ @@ -242,7 +242,7 @@ kind: Service cluster := cluster.DeepCopy() cluster.Spec.Service = &v1beta1.ServiceSpec{Type: test.Type} - service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + service, err := (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) assert.NilError(t, err) alwaysExpect(t, service) test.Expect(t, service) @@ -254,18 +254,18 @@ kind: Service cluster.Spec.Service = &v1beta1.ServiceSpec{Type: "ClusterIP", NodePort: new(int32(32000))} recorder := new(record.FakeRecorder) - service, err := (kubernetesBackend{}).LeaderLeaseService(cluster, recorder) + service, err := (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, recorder) assert.ErrorContains(t, err, `NodePort cannot be set with type ClusterIP on Service "pg2-ha"`) assert.Assert(t, service == nil) }) } -func TestKubernetesPrimaryService(t *testing.T) { +func TestKubernetesEndpointsPrimaryService(t *testing.T) { cluster := new(v1beta1.PostgresCluster) cluster.Spec.Port = new(int32(2600)) t.Run("NoLeader", func(t *testing.T) { - spec, subset, err := (kubernetesBackend{}).PrimaryService(cluster, nil) + spec, subset, err := (kubernetesEndpointsBackend{}).PrimaryService(cluster, nil) assert.ErrorContains(t, err, "not implemented") assert.DeepEqual(t, spec, corev1.ServiceSpec{}) assert.Assert(t, subset == nil) @@ -275,7 +275,7 @@ func TestKubernetesPrimaryService(t *testing.T) { leader := &corev1.Service{} leader.Spec.ClusterIP = "1.9.8.3" - spec, subset, err := (kubernetesBackend{}).PrimaryService(cluster, leader) + spec, subset, err := (kubernetesEndpointsBackend{}).PrimaryService(cluster, leader) assert.NilError(t, err) assert.Equal(t, spec.ClusterIP, corev1.ClusterIPNone) @@ -299,7 +299,7 @@ ports: }) } -func TestKubernetesObserve(t *testing.T) { +func TestKubernetesEndpointsObserve(t *testing.T) { _, cc := require.Kubernetes2(t) require.ParallelCapacity(t, 0) ns := require.Namespace(t, cc) @@ -310,14 +310,14 @@ func TestKubernetesObserve(t *testing.T) { cluster.Name = "observe-test" t.Run("NotFound, not ready", func(t *testing.T) { - observation, err := (kubernetesBackend{}).Observe(ctx, cc, cluster, false) + observation, err := (kubernetesEndpointsBackend{}).Observe(ctx, cc, cluster, false) assert.NilError(t, err) assert.Equal(t, observation.SystemIdentifier, "") assert.Equal(t, observation.RequeueAfter, time.Duration(0)) }) t.Run("NotFound, ready", func(t *testing.T) { - observation, err := (kubernetesBackend{}).Observe(ctx, cc, cluster, true) + observation, err := (kubernetesEndpointsBackend{}).Observe(ctx, cc, cluster, true) assert.NilError(t, err) assert.Equal(t, observation.SystemIdentifier, "") assert.Equal(t, observation.RequeueAfter, time.Second) @@ -329,14 +329,14 @@ func TestKubernetesObserve(t *testing.T) { assert.NilError(t, cc.Create(ctx, endpoints)) t.Cleanup(func() { assert.Check(t, client.IgnoreNotFound(cc.Delete(ctx, endpoints))) }) - observation, err := (kubernetesBackend{}).Observe(ctx, cc, cluster, false) + observation, err := (kubernetesEndpointsBackend{}).Observe(ctx, cc, cluster, false) assert.NilError(t, err) assert.Equal(t, observation.SystemIdentifier, "123456") assert.Equal(t, observation.RequeueAfter, time.Duration(0)) }) } -func TestKubernetesDelete(t *testing.T) { +func TestKubernetesEndpointsDelete(t *testing.T) { _, cc := require.Kubernetes2(t) require.ParallelCapacity(t, 0) ns := require.Namespace(t, cc) @@ -353,7 +353,7 @@ func TestKubernetesDelete(t *testing.T) { } assert.NilError(t, cc.Create(ctx, endpoints)) - assert.NilError(t, (kubernetesBackend{}).Delete(ctx, cc, cluster)) + assert.NilError(t, (kubernetesEndpointsBackend{}).Delete(ctx, cc, cluster)) err := cc.Get(ctx, client.ObjectKeyFromObject(endpoints), endpoints) assert.Assert(t, apierrors.IsNotFound(err), "expected the Endpoints to be deleted, got %v", err) From acbfbc1b06f85152f416ff00ef47bb9279172b24 Mon Sep 17 00:00:00 2001 From: yoav-katz Date: Wed, 5 Aug 2026 18:40:54 +0300 Subject: [PATCH 3/5] fix(lint) --- internal/patroni/dcs/kubernetes_endpoints_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/patroni/dcs/kubernetes_endpoints_test.go b/internal/patroni/dcs/kubernetes_endpoints_test.go index 7745155eb6..231107ea38 100644 --- a/internal/patroni/dcs/kubernetes_endpoints_test.go +++ b/internal/patroni/dcs/kubernetes_endpoints_test.go @@ -189,10 +189,10 @@ kind: Service service, err := (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) assert.NilError(t, err) - assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ + assert.DeepEqual(t, service.Annotations, map[string]string{ "a": "v1", }) - assert.DeepEqual(t, service.ObjectMeta.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ + assert.DeepEqual(t, service.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ "b": "v2", "postgres-operator.crunchydata.com/cluster": "pg2", "postgres-operator.crunchydata.com/patroni": "pg2-ha", @@ -210,11 +210,11 @@ kind: Service service, err = (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) assert.NilError(t, err) - assert.DeepEqual(t, service.ObjectMeta.Annotations, map[string]string{ + assert.DeepEqual(t, service.Annotations, map[string]string{ "a": "v1", "c": "v3", }) - assert.DeepEqual(t, service.ObjectMeta.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ + assert.DeepEqual(t, service.Labels, map[string]string(naming.WithPerconaLabels(map[string]string{ "b": "v2", "d": "v4", "postgres-operator.crunchydata.com/cluster": "pg2", From ad99e0867fafdc1a8ac783a25e7b1c323a1d77af Mon Sep 17 00:00:00 2001 From: yoav-katz Date: Wed, 5 Aug 2026 22:13:07 +0300 Subject: [PATCH 4/5] fix(patroni): preserve PATRONI_KUBERNETES_* env var order to avoid unwanted restarts on upgrade --- internal/patroni/config.go | 5 ++++- internal/patroni/config_test.go | 8 ++++---- internal/patroni/reconcile_test.go | 32 +++++++++++++++--------------- 3 files changed, 24 insertions(+), 21 deletions(-) diff --git a/internal/patroni/config.go b/internal/patroni/config.go index 498875c18b..1f16c28323 100644 --- a/internal/patroni/config.go +++ b/internal/patroni/config.go @@ -8,6 +8,7 @@ import ( "fmt" "maps" "path" + "slices" "strings" "github.com/pkg/errors" @@ -353,6 +354,8 @@ func instanceEnvironment( // - https://github.com/zalando/patroni/blob/v2.0.2/patroni/config.py#L247 // - https://github.com/zalando/patroni/blob/v2.0.2/patroni/postgresql/postmaster.py#L215-L216 + // Insert after PATRONI_NAME: appending would reorder existing StatefulSet + // env and force a rolling restart. variables := []corev1.EnvVar{ // Set "name" to the v1.Pod's name. Required for Patroni's node identity. // Patroni must be restarted when changing this value. @@ -417,7 +420,7 @@ func instanceEnvironment( }, } - return append(variables, dcsEnvVars...) + return slices.Insert(variables, 1, dcsEnvVars...) } // instanceConfigFiles returns projections of Patroni's configuration files diff --git a/internal/patroni/config_test.go b/internal/patroni/config_test.go index 2ea7419dac..a3ffbb0aa4 100644 --- a/internal/patroni/config_test.go +++ b/internal/patroni/config_test.go @@ -1167,6 +1167,10 @@ func TestInstanceEnvironment(t *testing.T) { fieldRef: apiVersion: v1 fieldPath: metadata.name +- name: PATRONI_KUBERNETES_POD_IP + value: 1.2.3.4 +- name: PATRONI_KUBERNETES_PORTS + value: '[]' - name: PATRONI_POSTGRESQL_CONNECT_ADDRESS value: $(PATRONI_NAME).pod-dns:5432 - name: PATRONI_POSTGRESQL_LISTEN @@ -1181,10 +1185,6 @@ func TestInstanceEnvironment(t *testing.T) { value: '*:8008' - name: PATRONICTL_CONFIG_FILE value: /etc/patroni -- name: PATRONI_KUBERNETES_POD_IP - value: 1.2.3.4 -- name: PATRONI_KUBERNETES_PORTS - value: '[]' `)) }) } diff --git a/internal/patroni/reconcile_test.go b/internal/patroni/reconcile_test.go index 9e74aedd38..a00f559225 100644 --- a/internal/patroni/reconcile_test.go +++ b/internal/patroni/reconcile_test.go @@ -175,6 +175,14 @@ containers: fieldRef: apiVersion: v1 fieldPath: metadata.name + - name: PATRONI_KUBERNETES_POD_IP + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: status.podIP + - name: PATRONI_KUBERNETES_PORTS + value: | + [] - name: PATRONI_POSTGRESQL_CONNECT_ADDRESS value: $(PATRONI_NAME).:5432 - name: PATRONI_POSTGRESQL_LISTEN @@ -189,14 +197,6 @@ containers: value: '*:8008' - name: PATRONICTL_CONFIG_FILE value: /etc/patroni - - name: PATRONI_KUBERNETES_POD_IP - valueFrom: - fieldRef: - apiVersion: v1 - fieldPath: status.podIP - - name: PATRONI_KUBERNETES_PORTS - value: | - [] livenessProbe: exec: command: @@ -289,6 +289,14 @@ containers: fieldRef: apiVersion: v1 fieldPath: metadata.name + - name: PATRONI_KUBERNETES_POD_IP + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: status.podIP + - name: PATRONI_KUBERNETES_PORTS + value: | + [] - name: PATRONI_POSTGRESQL_CONNECT_ADDRESS value: $(PATRONI_NAME).:5432 - name: PATRONI_POSTGRESQL_LISTEN @@ -303,14 +311,6 @@ containers: value: '*:8008' - name: PATRONICTL_CONFIG_FILE value: /etc/patroni - - name: PATRONI_KUBERNETES_POD_IP - valueFrom: - fieldRef: - apiVersion: v1 - fieldPath: status.podIP - - name: PATRONI_KUBERNETES_PORTS - value: | - [] livenessProbe: failureThreshold: 3 httpGet: From 93c53a1404660306c236c54dd647a02b50664b5e Mon Sep 17 00:00:00 2001 From: yoav-katz Date: Wed, 5 Aug 2026 22:31:23 +0300 Subject: [PATCH 5/5] fix(dcs): clarify leader Service error and restore NodePort test coverage --- .../postgrescluster/cluster_test.go | 4 +- .../controller/postgrescluster/instance.go | 3 +- internal/patroni/config.go | 4 +- internal/patroni/config_test.go | 35 +++--------- internal/patroni/dcs/kubernetes_endpoints.go | 3 +- .../patroni/dcs/kubernetes_endpoints_test.go | 57 ++++++++++++++++--- 6 files changed, 64 insertions(+), 42 deletions(-) diff --git a/internal/controller/postgrescluster/cluster_test.go b/internal/controller/postgrescluster/cluster_test.go index fb6eb19a1e..f663c0e7dc 100644 --- a/internal/controller/postgrescluster/cluster_test.go +++ b/internal/controller/postgrescluster/cluster_test.go @@ -594,7 +594,7 @@ func TestGenerateClusterPrimaryService(t *testing.T) { leader.Spec.ClusterIP = "1.9.8.3" _, _, err := reconciler.generateClusterPrimaryService(cluster, nil) - assert.ErrorContains(t, err, "not implemented") + assert.ErrorContains(t, err, "not available yet") alwaysExpect := func(t testing.TB, service *corev1.Service, endpoints *corev1.Endpoints) { assert.Assert(t, cmp.MarshalMatches(service.TypeMeta, ` @@ -699,7 +699,7 @@ func TestReconcileClusterPrimaryService(t *testing.T) { assert.NilError(t, cc.Create(ctx, cluster)) _, err := reconciler.reconcileClusterPrimaryService(ctx, cluster, nil) - assert.ErrorContains(t, err, "not implemented") + assert.ErrorContains(t, err, "not available yet") leader := &corev1.Service{} leader.Spec.ClusterIP = "192.0.2.10" diff --git a/internal/controller/postgrescluster/instance.go b/internal/controller/postgrescluster/instance.go index 0938356c39..196f4b8a33 100644 --- a/internal/controller/postgrescluster/instance.go +++ b/internal/controller/postgrescluster/instance.go @@ -1273,7 +1273,8 @@ func (r *Reconciler) reconcileInstance( } // K8SPG-708 - initImage, err := k8s.InitImage(ctx, r.Client, cluster, spec) + var initImage string + initImage, err = k8s.InitImage(ctx, r.Client, cluster, spec) if err != nil { return errors.Wrap(err, "failed to determine initial init image") } diff --git a/internal/patroni/config.go b/internal/patroni/config.go index 1f16c28323..5457be2b48 100644 --- a/internal/patroni/config.go +++ b/internal/patroni/config.go @@ -354,8 +354,6 @@ func instanceEnvironment( // - https://github.com/zalando/patroni/blob/v2.0.2/patroni/config.py#L247 // - https://github.com/zalando/patroni/blob/v2.0.2/patroni/postgresql/postmaster.py#L215-L216 - // Insert after PATRONI_NAME: appending would reorder existing StatefulSet - // env and force a rolling restart. variables := []corev1.EnvVar{ // Set "name" to the v1.Pod's name. Required for Patroni's node identity. // Patroni must be restarted when changing this value. @@ -420,6 +418,8 @@ func instanceEnvironment( }, } + // Insert after PATRONI_NAME: appending would reorder existing StatefulSet + // env and force a rolling restart. return slices.Insert(variables, 1, dcsEnvVars...) } diff --git a/internal/patroni/config_test.go b/internal/patroni/config_test.go index a3ffbb0aa4..2985f26160 100644 --- a/internal/patroni/config_test.go +++ b/internal/patroni/config_test.go @@ -19,6 +19,7 @@ import ( "sigs.k8s.io/yaml" "github.com/percona/percona-postgresql-operator/v2/internal/naming" + "github.com/percona/percona-postgresql-operator/v2/internal/patroni/dcs" "github.com/percona/percona-postgresql-operator/v2/internal/postgres" "github.com/percona/percona-postgresql-operator/v2/internal/testing/cmp" "github.com/percona/percona-postgresql-operator/v2/internal/testing/require" @@ -27,24 +28,6 @@ import ( "github.com/percona/percona-postgresql-operator/v2/pkg/apis/upstream.pgv2.percona.com/v1beta1" ) -// kubernetesClusterYAML stands in for dcs.For(cluster).ClusterYAML(cluster) -// (see internal/patroni/dcs) without importing that package from here. -func kubernetesClusterYAML(cluster *v1beta1.PostgresCluster) map[string]any { - labels := map[string]string{naming.LabelCluster: cluster.Name} - if cluster.CompareVersion("2.9.0") >= 0 { - labels = naming.Merge(cluster.Spec.Metadata.GetLabelsOrNil(), labels) - } - return map[string]any{ - "kubernetes": map[string]any{ - "namespace": cluster.Namespace, - "role_label": naming.LabelRole, - "scope_label": naming.LabelPatroni, - "use_endpoints": true, - "labels": labels, - }, - } -} - func TestClusterYAML(t *testing.T) { t.Parallel() @@ -55,7 +38,7 @@ func TestClusterYAML(t *testing.T) { cluster.Namespace = "some-namespace" cluster.Name = "cluster-name" - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) assert.Equal(t, data, strings.TrimSpace(` # Generated by postgres-operator. DO NOT EDIT UNLESS YOU KNOW WHAT YOU'RE DOING. @@ -122,7 +105,7 @@ watchdog: }, } - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -152,7 +135,7 @@ watchdog: } cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -174,7 +157,7 @@ watchdog: } cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -199,7 +182,7 @@ watchdog: } cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -221,7 +204,7 @@ watchdog: cluster.Spec.PostgresVersion = 17 cluster.Spec.Extensions.PGTDE.Enabled = true - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -248,7 +231,7 @@ watchdog: cluster.Spec.Patroni = &v1beta1.PatroniSpec{} cluster.Spec.Patroni.Default() - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) var parsed map[string]any @@ -268,7 +251,7 @@ watchdog: cluster.Name = "cluster-name" cluster.Spec.PostgresVersion = 14 - data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, kubernetesClusterYAML(cluster)) + data, err := clusterYAML(cluster, postgres.HBAs{}, postgres.Parameters{}, dcs.For(cluster).ClusterYAML(cluster)) assert.NilError(t, err) assert.Equal(t, data, strings.TrimSpace(` # Generated by postgres-operator. DO NOT EDIT UNLESS YOU KNOW WHAT YOU'RE DOING. diff --git a/internal/patroni/dcs/kubernetes_endpoints.go b/internal/patroni/dcs/kubernetes_endpoints.go index f052e31d6a..c7d3755a56 100644 --- a/internal/patroni/dcs/kubernetes_endpoints.go +++ b/internal/patroni/dcs/kubernetes_endpoints.go @@ -251,8 +251,7 @@ func (kubernetesEndpointsBackend) PrimaryService( // ClusterIP of the Service created in Reconciler.reconcilePatroniLeaderLease // when Patroni is using Endpoints. if leader == nil { - // TODO(cbandy): We need to build a different kind of Service here. - return corev1.ServiceSpec{}, nil, errors.New("Patroni DCS other than Kubernetes Endpoints is not implemented") + return corev1.ServiceSpec{}, nil, errors.New("Patroni leader Service is not available yet") } // Allocate no IP address (headless) and manage the Endpoints ourselves. diff --git a/internal/patroni/dcs/kubernetes_endpoints_test.go b/internal/patroni/dcs/kubernetes_endpoints_test.go index 231107ea38..30ee16d777 100644 --- a/internal/patroni/dcs/kubernetes_endpoints_test.go +++ b/internal/patroni/dcs/kubernetes_endpoints_test.go @@ -249,15 +249,54 @@ kind: Service }) } - t.Run("NodePortWithClusterIP", func(t *testing.T) { - cluster := cluster.DeepCopy() - cluster.Spec.Service = &v1beta1.ServiceSpec{Type: "ClusterIP", NodePort: new(int32(32000))} + typesAndPort := []struct { + Description string + Type string + NodePort *int32 + Expect func(testing.TB, *corev1.Service, error) + }{ + {Description: "ClusterIP with Port 32000", Type: "ClusterIP", + NodePort: new(int32(32000)), Expect: func(t testing.TB, service *corev1.Service, err error) { + assert.ErrorContains(t, err, `NodePort cannot be set with type ClusterIP on Service "pg2-ha"`) + assert.Assert(t, service == nil) + }}, + {Description: "NodePort with Port 32001", Type: "NodePort", + NodePort: new(int32(32001)), Expect: func(t testing.TB, service *corev1.Service, err error) { + assert.NilError(t, err) + alwaysExpect(t, service) + assert.Equal(t, service.Spec.Type, corev1.ServiceTypeNodePort) + assert.Assert(t, cmp.MarshalMatches(service.Spec.Ports, ` +- name: postgres + nodePort: 32001 + port: 9876 + protocol: TCP + targetPort: postgres +`)) + }}, + {Description: "LoadBalancer with Port 32002", Type: "LoadBalancer", + NodePort: new(int32(32002)), Expect: func(t testing.TB, service *corev1.Service, err error) { + assert.Equal(t, service.Spec.Type, corev1.ServiceTypeLoadBalancer) + assert.NilError(t, err) + alwaysExpect(t, service) + assert.Assert(t, cmp.MarshalMatches(service.Spec.Ports, ` +- name: postgres + nodePort: 32002 + port: 9876 + protocol: TCP + targetPort: postgres +`)) + }}, + } - recorder := new(record.FakeRecorder) - service, err := (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, recorder) - assert.ErrorContains(t, err, `NodePort cannot be set with type ClusterIP on Service "pg2-ha"`) - assert.Assert(t, service == nil) - }) + for _, test := range typesAndPort { + t.Run(test.Description, func(t *testing.T) { + cluster := cluster.DeepCopy() + cluster.Spec.Service = &v1beta1.ServiceSpec{Type: test.Type, NodePort: test.NodePort} + + service, err := (kubernetesEndpointsBackend{}).LeaderLeaseService(cluster, new(record.FakeRecorder)) + test.Expect(t, service, err) + }) + } } func TestKubernetesEndpointsPrimaryService(t *testing.T) { @@ -266,7 +305,7 @@ func TestKubernetesEndpointsPrimaryService(t *testing.T) { t.Run("NoLeader", func(t *testing.T) { spec, subset, err := (kubernetesEndpointsBackend{}).PrimaryService(cluster, nil) - assert.ErrorContains(t, err, "not implemented") + assert.ErrorContains(t, err, "not available yet") assert.DeepEqual(t, spec, corev1.ServiceSpec{}) assert.Assert(t, subset == nil) })