LoadBalancer toggles for master and replica pooler pods (#1799)

* Add support for pooler load balancer

Signed-off-by: Sergey Shatunov <me@prok.pw>

* Rename to enable_master_pooler_load_balancer

Signed-off-by: Sergey Shatunov <me@prok.pw>

* target port should be intval
* enhance pooler e2e test
* add new options to crds.go

Co-authored-by: Sergey Shatunov <me@prok.pw>

pkg/cluster/cluster_test.go

@@ -1041,7 +1041,7 @@ func TestCrossNamespacedSecrets(t *testing.T) {
 					ConnectionPoolerDefaultCPULimit:      "100m",
 					ConnectionPoolerDefaultMemoryRequest: "100Mi",
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
-					NumberOfInstances:                    int32ToPointer(1),
+					NumberOfInstances:                    k8sutil.Int32ToPointer(1),
 				},
 				PodManagementPolicy: "ordered_ready",
 				Resources: config.Resources{

pkg/cluster/connection_pooler.go

@@ -247,7 +247,7 @@ func (c *Cluster) generateConnectionPoolerPodTemplate(role PostgresRole) (
 		},
 		{
 			Name:  "PGPORT",
-			Value: c.servicePort(role),
+			Value: fmt.Sprint(c.servicePort(role)),
 		},
 		{
 			Name: "PGUSER",
@@ -386,7 +386,7 @@ func (c *Cluster) generateConnectionPoolerService(connectionPooler *ConnectionPo
 			{
 				Name:       connectionPooler.Name,
 				Port:       pgPort,
-				TargetPort: intstr.IntOrString{StrVal: c.servicePort(connectionPooler.Role)},
+				TargetPort: intstr.IntOrString{IntVal: c.servicePort(connectionPooler.Role)},
 			},
 		},
 		Type: v1.ServiceTypeClusterIP,
@@ -395,6 +395,10 @@ func (c *Cluster) generateConnectionPoolerService(connectionPooler *ConnectionPo
 		},
 	}
 
+	if c.shouldCreateLoadBalancerForPoolerService(connectionPooler.Role, spec) {
+		c.configureLoadBalanceService(&serviceSpec, spec.AllowedSourceRanges)
+	}
+
 	service := &v1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        connectionPooler.Name,
@@ -415,6 +419,29 @@ func (c *Cluster) generateConnectionPoolerService(connectionPooler *ConnectionPo
 	return service
 }
 
+func (c *Cluster) shouldCreateLoadBalancerForPoolerService(role PostgresRole, spec *acidv1.PostgresSpec) bool {
+
+	switch role {
+
+	case Replica:
+		// if the value is explicitly set in a Postgresql manifest, follow this setting
+		if spec.EnableReplicaPoolerLoadBalancer != nil {
+			return *spec.EnableReplicaPoolerLoadBalancer
+		}
+		// otherwise, follow the operator configuration
+		return c.OpConfig.EnableReplicaPoolerLoadBalancer
+
+	case Master:
+		if spec.EnableMasterPoolerLoadBalancer != nil {
+			return *spec.EnableMasterPoolerLoadBalancer
+		}
+		return c.OpConfig.EnableMasterPoolerLoadBalancer
+
+	default:
+		panic(fmt.Sprintf("Unknown role %v", role))
+	}
+}
+
 //delete connection pooler
 func (c *Cluster) deleteConnectionPooler(role PostgresRole) (err error) {
 	c.logger.Infof("deleting connection pooler spilo-role=%s", role)
@@ -811,6 +838,7 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, Look
 func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql, role PostgresRole) (
 	SyncReason, error) {
 
+	syncReason := make([]string, 0)
 	deployment, err := c.KubeClient.
 		Deployments(c.Namespace).
 		Get(context.TODO(), c.connectionPoolerName(role), metav1.GetOptions{})
@@ -865,25 +893,26 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 
 		if oldSpec != nil {
 			specSync, specReason = needSyncConnectionPoolerSpecs(oldConnectionPooler, newConnectionPooler, c.logger)
+			syncReason = append(syncReason, specReason...)
 		}
 
 		defaultsSync, defaultsReason := c.needSyncConnectionPoolerDefaults(&c.Config, newConnectionPooler, deployment)
-		reason := append(specReason, defaultsReason...)
+		syncReason = append(syncReason, defaultsReason...)
 
 		if specSync || defaultsSync {
 			c.logger.Infof("Update connection pooler deployment %s, reason: %+v",
-				c.connectionPoolerName(role), reason)
+				c.connectionPoolerName(role), syncReason)
 			newDeploymentSpec, err := c.generateConnectionPoolerDeployment(c.ConnectionPooler[role])
 			if err != nil {
 				msg := "could not generate deployment for connection pooler: %v"
-				return reason, fmt.Errorf(msg, err)
+				return syncReason, fmt.Errorf(msg, err)
 			}
 
 			deployment, err := updateConnectionPoolerDeployment(c.KubeClient,
 				newDeploymentSpec)
 
 			if err != nil {
-				return reason, err
+				return syncReason, err
 			}
 			c.ConnectionPooler[role].Deployment = deployment
 		}
@@ -898,31 +927,40 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
 		c.ConnectionPooler[role].Deployment = deployment
 	}
 
-	service, err := c.KubeClient.
-		Services(c.Namespace).
-		Get(context.TODO(), c.connectionPoolerName(role), metav1.GetOptions{})
-
-	if err != nil && k8sutil.ResourceNotFound(err) {
-		msg := "Service %s for connection pooler synchronization is not found, create it"
-		c.logger.Warningf(msg, c.connectionPoolerName(role))
-
-		serviceSpec := c.generateConnectionPoolerService(c.ConnectionPooler[role])
-		service, err := c.KubeClient.
-			Services(serviceSpec.Namespace).
-			Create(context.TODO(), serviceSpec, metav1.CreateOptions{})
-
-		if err != nil {
-			return NoSync, err
+	if svc, err := c.KubeClient.Services(c.Namespace).Get(context.TODO(), c.connectionPoolerName(role), metav1.GetOptions{}); err == nil {
+		c.ConnectionPooler[role].Service = svc
+		desiredSvc := c.generateConnectionPoolerService(c.ConnectionPooler[role])
+		if match, reason := k8sutil.SameService(svc, desiredSvc); !match {
+			syncReason = append(syncReason, reason)
+			c.logServiceChanges(role, svc, desiredSvc, false, reason)
+			updatedService, err := c.updateService(role, svc, desiredSvc)
+			if err != nil {
+				return syncReason, fmt.Errorf("could not update %s service to match desired state: %v", role, err)
+			}
+			c.ConnectionPooler[role].Service = updatedService
+			c.logger.Infof("%s service %q is in the desired state now", role, util.NameFromMeta(desiredSvc.ObjectMeta))
 		}
-		c.ConnectionPooler[role].Service = service
+		return NoSync, nil
+	}
 
-	} else if err != nil {
+	if !k8sutil.ResourceNotFound(err) {
 		msg := "could not get connection pooler service to sync: %v"
 		return NoSync, fmt.Errorf(msg, err)
-	} else {
-		// Service updates are not supported and probably not that useful anyway
-		c.ConnectionPooler[role].Service = service
 	}
 
+	c.ConnectionPooler[role].Service = nil
+	msg := "Service %s for connection pooler synchronization is not found, create it"
+	c.logger.Warningf(msg, c.connectionPoolerName(role))
+
+	serviceSpec := c.generateConnectionPoolerService(c.ConnectionPooler[role])
+	service, err := c.KubeClient.
+		Services(serviceSpec.Namespace).
+		Create(context.TODO(), serviceSpec, metav1.CreateOptions{})
+
+	if err != nil {
+		return NoSync, err
+	}
+	c.ConnectionPooler[role].Service = service
+
 	return NoSync, nil
 }

pkg/cluster/connection_pooler_test.go

@@ -27,10 +27,6 @@ func boolToPointer(value bool) *bool {
 	return &value
 }
 
-func int32ToPointer(value int32) *int32 {
-	return &value
-}
-
 func deploymentUpdated(cluster *Cluster, err error, reason SyncReason) error {
 	for _, role := range [2]PostgresRole{Master, Replica} {
 
@@ -294,7 +290,7 @@ func TestConnectionPoolerCreateDeletion(t *testing.T) {
 					ConnectionPoolerDefaultCPULimit:      "100m",
 					ConnectionPoolerDefaultMemoryRequest: "100Mi",
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
-					NumberOfInstances:                    int32ToPointer(1),
+					NumberOfInstances:                    k8sutil.Int32ToPointer(1),
 				},
 				PodManagementPolicy: "ordered_ready",
 				Resources: config.Resources{
@@ -401,7 +397,7 @@ func TestConnectionPoolerSync(t *testing.T) {
 					ConnectionPoolerDefaultCPULimit:      "100m",
 					ConnectionPoolerDefaultMemoryRequest: "100Mi",
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
-					NumberOfInstances:                    int32ToPointer(1),
+					NumberOfInstances:                    k8sutil.Int32ToPointer(1),
 				},
 				PodManagementPolicy: "ordered_ready",
 				Resources: config.Resources{
@@ -639,7 +635,7 @@ func TestConnectionPoolerSync(t *testing.T) {
 	for _, tt := range tests {
 		tt.cluster.OpConfig.ConnectionPooler.Image = tt.defaultImage
 		tt.cluster.OpConfig.ConnectionPooler.NumberOfInstances =
-			int32ToPointer(tt.defaultInstances)
+			k8sutil.Int32ToPointer(tt.defaultInstances)
 
 		t.Logf("running test for %s [%s]", testName, tt.subTest)
 
@@ -664,7 +660,7 @@ func TestConnectionPoolerPodSpec(t *testing.T) {
 					ReplicationUsername: replicationUserName,
 				},
 				ConnectionPooler: config.ConnectionPooler{
-					MaxDBConnections:                     int32ToPointer(60),
+					MaxDBConnections:                     k8sutil.Int32ToPointer(60),
 					ConnectionPoolerDefaultCPURequest:    "100m",
 					ConnectionPoolerDefaultCPULimit:      "100m",
 					ConnectionPoolerDefaultMemoryRequest: "100Mi",

pkg/cluster/k8sres.go

@@ -102,15 +102,15 @@ func (c *Cluster) serviceAddress(role PostgresRole) string {
 	return ""
 }
 
-func (c *Cluster) servicePort(role PostgresRole) string {
+func (c *Cluster) servicePort(role PostgresRole) int32 {
 	service, exist := c.Services[role]
 
 	if exist {
-		return fmt.Sprint(service.Spec.Ports[0].Port)
+		return service.Spec.Ports[0].Port
 	}
 
-	c.logger.Warningf("No service for role %s", role)
-	return ""
+	c.logger.Warningf("No service for role %s - defaulting to port 5432", role)
+	return pgPort
 }
 
 func (c *Cluster) podDisruptionBudgetName() string {
@@ -1699,23 +1699,7 @@ func (c *Cluster) generateService(role PostgresRole, spec *acidv1.PostgresSpec)
 	}
 
 	if c.shouldCreateLoadBalancerForService(role, spec) {
-
-		// spec.AllowedSourceRanges evaluates to the empty slice of zero length
-		// when omitted or set to 'null'/empty sequence in the PG manifest
-		if len(spec.AllowedSourceRanges) > 0 {
-			serviceSpec.LoadBalancerSourceRanges = spec.AllowedSourceRanges
-		} else {
-			// safe default value: lock a load balancer only to the local address unless overridden explicitly
-			serviceSpec.LoadBalancerSourceRanges = []string{localHost}
-		}
-
-		c.logger.Debugf("final load balancer source ranges as seen in a service spec (not necessarily applied): %q", serviceSpec.LoadBalancerSourceRanges)
-		serviceSpec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyType(c.OpConfig.ExternalTrafficPolicy)
-		serviceSpec.Type = v1.ServiceTypeLoadBalancer
-	} else if role == Replica {
-		// before PR #258, the replica service was only created if allocated a LB
-		// now we always create the service but warn if the LB is absent
-		c.logger.Debugf("No load balancer created for the replica service")
+		c.configureLoadBalanceService(&serviceSpec, spec.AllowedSourceRanges)
 	}
 
 	service := &v1.Service{
@@ -1731,6 +1715,21 @@ func (c *Cluster) generateService(role PostgresRole, spec *acidv1.PostgresSpec)
 	return service
 }
 
+func (c *Cluster) configureLoadBalanceService(serviceSpec *v1.ServiceSpec, sourceRanges []string) {
+	// spec.AllowedSourceRanges evaluates to the empty slice of zero length
+	// when omitted or set to 'null'/empty sequence in the PG manifest
+	if len(sourceRanges) > 0 {
+		serviceSpec.LoadBalancerSourceRanges = sourceRanges
+	} else {
+		// safe default value: lock a load balancer only to the local address unless overridden explicitly
+		serviceSpec.LoadBalancerSourceRanges = []string{localHost}
+	}
+
+	c.logger.Debugf("final load balancer source ranges as seen in a service spec (not necessarily applied): %q", serviceSpec.LoadBalancerSourceRanges)
+	serviceSpec.ExternalTrafficPolicy = v1.ServiceExternalTrafficPolicyType(c.OpConfig.ExternalTrafficPolicy)
+	serviceSpec.Type = v1.ServiceTypeLoadBalancer
+}
+
 func (c *Cluster) generateServiceAnnotations(role PostgresRole, spec *acidv1.PostgresSpec) map[string]string {
 	annotations := make(map[string]string)
 

pkg/cluster/k8sres_test.go

@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/kubernetes/fake"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
 )
@@ -1484,6 +1485,188 @@ func TestGenerateService(t *testing.T) {
 
 }
 
+func newLBFakeClient() (k8sutil.KubernetesClient, *fake.Clientset) {
+	clientSet := fake.NewSimpleClientset()
+
+	return k8sutil.KubernetesClient{
+		DeploymentsGetter: clientSet.AppsV1(),
+		ServicesGetter:    clientSet.CoreV1(),
+	}, clientSet
+}
+
+func getServices(serviceType v1.ServiceType, sourceRanges []string, extTrafficPolicy, clusterName string) []v1.ServiceSpec {
+	return []v1.ServiceSpec{
+		v1.ServiceSpec{
+			ExternalTrafficPolicy:    v1.ServiceExternalTrafficPolicyType(extTrafficPolicy),
+			LoadBalancerSourceRanges: sourceRanges,
+			Ports:                    []v1.ServicePort{{Name: "postgresql", Port: 5432, TargetPort: intstr.IntOrString{IntVal: 5432}}},
+			Type:                     serviceType,
+		},
+		v1.ServiceSpec{
+			ExternalTrafficPolicy:    v1.ServiceExternalTrafficPolicyType(extTrafficPolicy),
+			LoadBalancerSourceRanges: sourceRanges,
+			Ports:                    []v1.ServicePort{{Name: clusterName + "-pooler", Port: 5432, TargetPort: intstr.IntOrString{IntVal: 5432}}},
+			Selector:                 map[string]string{"connection-pooler": clusterName + "-pooler"},
+			Type:                     serviceType,
+		},
+		v1.ServiceSpec{
+			ExternalTrafficPolicy:    v1.ServiceExternalTrafficPolicyType(extTrafficPolicy),
+			LoadBalancerSourceRanges: sourceRanges,
+			Ports:                    []v1.ServicePort{{Name: "postgresql", Port: 5432, TargetPort: intstr.IntOrString{IntVal: 5432}}},
+			Selector:                 map[string]string{"spilo-role": "replica", "application": "spilo", "cluster-name": clusterName},
+			Type:                     serviceType,
+		},
+		v1.ServiceSpec{
+			ExternalTrafficPolicy:    v1.ServiceExternalTrafficPolicyType(extTrafficPolicy),
+			LoadBalancerSourceRanges: sourceRanges,
+			Ports:                    []v1.ServicePort{{Name: clusterName + "-pooler-repl", Port: 5432, TargetPort: intstr.IntOrString{IntVal: 5432}}},
+			Selector:                 map[string]string{"connection-pooler": clusterName + "-pooler-repl"},
+			Type:                     serviceType,
+		},
+	}
+}
+
+func TestEnableLoadBalancers(t *testing.T) {
+	testName := "Test enabling LoadBalancers"
+	client, _ := newLBFakeClient()
+	clusterName := "acid-test-cluster"
+	namespace := "default"
+	clusterNameLabel := "cluster-name"
+	roleLabel := "spilo-role"
+	roles := []PostgresRole{Master, Replica}
+	sourceRanges := []string{"192.186.1.2/22"}
+	extTrafficPolicy := "Cluster"
+
+	tests := []struct {
+		subTest          string
+		config           config.Config
+		pgSpec           acidv1.Postgresql
+		expectedServices []v1.ServiceSpec
+	}{
+		{
+			subTest: "LBs enabled in config, disabled in manifest",
+			config: config.Config{
+				ConnectionPooler: config.ConnectionPooler{
+					ConnectionPoolerDefaultCPURequest:    "100m",
+					ConnectionPoolerDefaultCPULimit:      "100m",
+					ConnectionPoolerDefaultMemoryRequest: "100Mi",
+					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
+					NumberOfInstances:                    k8sutil.Int32ToPointer(1),
+				},
+				EnableMasterLoadBalancer:        true,
+				EnableMasterPoolerLoadBalancer:  true,
+				EnableReplicaLoadBalancer:       true,
+				EnableReplicaPoolerLoadBalancer: true,
+				ExternalTrafficPolicy:           extTrafficPolicy,
+				Resources: config.Resources{
+					ClusterLabels:    map[string]string{"application": "spilo"},
+					ClusterNameLabel: clusterNameLabel,
+					PodRoleLabel:     roleLabel,
+				},
+			},
+			pgSpec: acidv1.Postgresql{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      clusterName,
+					Namespace: namespace,
+				},
+				Spec: acidv1.PostgresSpec{
+					AllowedSourceRanges:             sourceRanges,
+					EnableConnectionPooler:          util.True(),
+					EnableReplicaConnectionPooler:   util.True(),
+					EnableMasterLoadBalancer:        util.False(),
+					EnableMasterPoolerLoadBalancer:  util.False(),
+					EnableReplicaLoadBalancer:       util.False(),
+					EnableReplicaPoolerLoadBalancer: util.False(),
+					NumberOfInstances:               1,
+					Resources: acidv1.Resources{
+						ResourceRequests: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+						ResourceLimits:   acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+					},
+					TeamID: "acid",
+					Volume: acidv1.Volume{
+						Size: "1G",
+					},
+				},
+			},
+			expectedServices: getServices(v1.ServiceTypeClusterIP, nil, "", clusterName),
+		},
+		{
+			subTest: "LBs enabled in manifest, disabled in config",
+			config: config.Config{
+				ConnectionPooler: config.ConnectionPooler{
+					ConnectionPoolerDefaultCPURequest:    "100m",
+					ConnectionPoolerDefaultCPULimit:      "100m",
+					ConnectionPoolerDefaultMemoryRequest: "100Mi",
+					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
+					NumberOfInstances:                    k8sutil.Int32ToPointer(1),
+				},
+				EnableMasterLoadBalancer:        false,
+				EnableMasterPoolerLoadBalancer:  false,
+				EnableReplicaLoadBalancer:       false,
+				EnableReplicaPoolerLoadBalancer: false,
+				ExternalTrafficPolicy:           extTrafficPolicy,
+				Resources: config.Resources{
+					ClusterLabels:    map[string]string{"application": "spilo"},
+					ClusterNameLabel: clusterNameLabel,
+					PodRoleLabel:     roleLabel,
+				},
+			},
+			pgSpec: acidv1.Postgresql{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      clusterName,
+					Namespace: namespace,
+				},
+				Spec: acidv1.PostgresSpec{
+					AllowedSourceRanges:             sourceRanges,
+					EnableConnectionPooler:          util.True(),
+					EnableReplicaConnectionPooler:   util.True(),
+					EnableMasterLoadBalancer:        util.True(),
+					EnableMasterPoolerLoadBalancer:  util.True(),
+					EnableReplicaLoadBalancer:       util.True(),
+					EnableReplicaPoolerLoadBalancer: util.True(),
+					NumberOfInstances:               1,
+					Resources: acidv1.Resources{
+						ResourceRequests: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+						ResourceLimits:   acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+					},
+					TeamID: "acid",
+					Volume: acidv1.Volume{
+						Size: "1G",
+					},
+				},
+			},
+			expectedServices: getServices(v1.ServiceTypeLoadBalancer, sourceRanges, extTrafficPolicy, clusterName),
+		},
+	}
+
+	for _, tt := range tests {
+		var cluster = New(
+			Config{
+				OpConfig: tt.config,
+			}, client, tt.pgSpec, logger, eventRecorder)
+
+		cluster.Name = clusterName
+		cluster.Namespace = namespace
+		cluster.ConnectionPooler = map[PostgresRole]*ConnectionPoolerObjects{}
+		generatedServices := make([]v1.ServiceSpec, 0)
+		for _, role := range roles {
+			cluster.syncService(role)
+			cluster.ConnectionPooler[role] = &ConnectionPoolerObjects{
+				Name:        cluster.connectionPoolerName(role),
+				ClusterName: cluster.ClusterName,
+				Namespace:   cluster.Namespace,
+				Role:        role,
+			}
+			cluster.syncConnectionPoolerWorker(&tt.pgSpec, &tt.pgSpec, role)
+			generatedServices = append(generatedServices, cluster.Services[role].Spec)
+			generatedServices = append(generatedServices, cluster.ConnectionPooler[role].Service.Spec)
+		}
+		if !reflect.DeepEqual(tt.expectedServices, generatedServices) {
+			t.Errorf("%s %s: expected %#v but got %#v", testName, tt.subTest, tt.expectedServices, generatedServices)
+		}
+	}
+}
+
 func TestGenerateCapabilities(t *testing.T) {
 
 	testName := "TestGenerateCapabilities"

pkg/cluster/resources.go

@@ -275,7 +275,7 @@ func (c *Cluster) createService(role PostgresRole) (*v1.Service, error) {
 	return service, nil
 }
 
-func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error {
+func (c *Cluster) updateService(role PostgresRole, oldService *v1.Service, newService *v1.Service) (*v1.Service, error) {
 	var (
 		svc *v1.Service
 		err error
@@ -283,11 +283,7 @@ func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error
 
 	c.setProcessName("updating %v service", role)
 
-	if c.Services[role] == nil {
-		return fmt.Errorf("there is no service in the cluster")
-	}
-
-	serviceName := util.NameFromMeta(c.Services[role].ObjectMeta)
+	serviceName := util.NameFromMeta(oldService.ObjectMeta)
 
 	// update the service annotation in order to propagate ELB notation.
 	if len(newService.ObjectMeta.Annotations) > 0 {
@@ -301,39 +297,38 @@ func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error
 				"")
 
 			if err != nil {
-				return fmt.Errorf("could not replace annotations for the service %q: %v", serviceName, err)
+				return nil, fmt.Errorf("could not replace annotations for the service %q: %v", serviceName, err)
 			}
 		} else {
-			return fmt.Errorf("could not form patch for the service metadata: %v", err)
+			return nil, fmt.Errorf("could not form patch for the service metadata: %v", err)
 		}
 	}
 
 	// now, patch the service spec, but when disabling LoadBalancers do update instead
 	// patch does not work because of LoadBalancerSourceRanges field (even if set to nil)
-	oldServiceType := c.Services[role].Spec.Type
+	oldServiceType := oldService.Spec.Type
 	newServiceType := newService.Spec.Type
 	if newServiceType == "ClusterIP" && newServiceType != oldServiceType {
-		newService.ResourceVersion = c.Services[role].ResourceVersion
-		newService.Spec.ClusterIP = c.Services[role].Spec.ClusterIP
+		newService.ResourceVersion = oldService.ResourceVersion
+		newService.Spec.ClusterIP = oldService.Spec.ClusterIP
 		svc, err = c.KubeClient.Services(serviceName.Namespace).Update(context.TODO(), newService, metav1.UpdateOptions{})
 		if err != nil {
-			return fmt.Errorf("could not update service %q: %v", serviceName, err)
+			return nil, fmt.Errorf("could not update service %q: %v", serviceName, err)
 		}
 	} else {
 		patchData, err := specPatch(newService.Spec)
 		if err != nil {
-			return fmt.Errorf("could not form patch for the service %q: %v", serviceName, err)
+			return nil, fmt.Errorf("could not form patch for the service %q: %v", serviceName, err)
 		}
 
 		svc, err = c.KubeClient.Services(serviceName.Namespace).Patch(
 			context.TODO(), serviceName.Name, types.MergePatchType, patchData, metav1.PatchOptions{}, "")
 		if err != nil {
-			return fmt.Errorf("could not patch service %q: %v", serviceName, err)
+			return nil, fmt.Errorf("could not patch service %q: %v", serviceName, err)
 		}
 	}
-	c.Services[role] = svc
 
-	return nil
+	return svc, nil
 }
 
 func (c *Cluster) deleteService(role PostgresRole) error {

pkg/cluster/sync.go

@@ -174,9 +174,11 @@ func (c *Cluster) syncService(role PostgresRole) error {
 		desiredSvc := c.generateService(role, &c.Spec)
 		if match, reason := k8sutil.SameService(svc, desiredSvc); !match {
 			c.logServiceChanges(role, svc, desiredSvc, false, reason)
-			if err = c.updateService(role, desiredSvc); err != nil {
+			updatedSvc, err := c.updateService(role, svc, desiredSvc)
+			if err != nil {
 				return fmt.Errorf("could not update %s service to match desired state: %v", role, err)
 			}
+			c.Services[role] = updatedSvc
 			c.logger.Infof("%s service %q is in the desired state now", role, util.NameFromMeta(desiredSvc.ObjectMeta))
 		}
 		return nil

pkg/cluster/util_test.go

@@ -57,7 +57,7 @@ func TestInheritedAnnotations(t *testing.T) {
 					ConnectionPoolerDefaultCPULimit:      "100m",
 					ConnectionPoolerDefaultMemoryRequest: "100Mi",
 					ConnectionPoolerDefaultMemoryLimit:   "100Mi",
-					NumberOfInstances:                    int32ToPointer(1),
+					NumberOfInstances:                    k8sutil.Int32ToPointer(1),
 				},
 				PodManagementPolicy: "ordered_ready",
 				Resources: config.Resources{