Deprecate old LB options, fix endpoint sync. (#287)

* Depreate old LB options, fix endpoint sync.

- deprecate useLoadBalancer, replicaLoadBalancer from the manifest
  and enable_load_balancer from the operator configuration. The old
  operator configuration options become no-op with this commit. For
  the old manifest options, `useLoadBalancer` and `replicaLoadBalancer`
  are still consulted,  but only in the absense of the new ones
  (enableMasterLoadBalancer and enableReplicaLoadBalancer).

- Make sure the endpoint being created during the sync receives proper
  addresses subset. This is more critical for the replicas, as for the
  masters Patroni will normally re-create the endpoint before the
  operator.

- Avoid creating the replica endpoint, since it will be created automatically
  by the corresponding service.
- Update the README and unit tests.

Code review by @mgomezch and @zerg-junior

pkg/cluster/cluster.go

@@ -170,6 +170,10 @@ func (c *Cluster) setStatus(status spec.PostgresStatus) {
 	}
 }
 
+func (c *Cluster) isNewCluster() bool {
+	return c.Status == spec.ClusterStatusCreating
+}
+
 // initUsers populates c.systemUsers and c.pgUsers maps.
 func (c *Cluster) initUsers() error {
 	c.setProcessName("initializing users")
@@ -255,11 +259,15 @@ func (c *Cluster) Create() error {
 		if c.Endpoints[role] != nil {
 			return fmt.Errorf("%s endpoint already exists in the cluster", role)
 		}
-		ep, err = c.createEndpoint(role)
-		if err != nil {
-			return fmt.Errorf("could not create %s endpoint: %v", role, err)
+		if role == Master {
+			// replica endpoint will be created by the replica service. Master endpoint needs to be created by us,
+			// since the corresponding master service doesn't define any selectors.
+			ep, err = c.createEndpoint(role)
+			if err != nil {
+				return fmt.Errorf("could not create %s endpoint: %v", role, err)
+			}
+			c.logger.Infof("endpoint %q has been successfully created", util.NameFromMeta(ep.ObjectMeta))
 		}
-		c.logger.Infof("endpoint %q has been successfully created", util.NameFromMeta(ep.ObjectMeta))
 
 		if c.Services[role] != nil {
 			return fmt.Errorf("service already exists in the cluster")

pkg/cluster/k8sres.go

@@ -683,12 +683,6 @@ func (c *Cluster) shouldCreateLoadBalancerForService(role PostgresRole, spec *sp
 
 	case Replica:
 
-		// deprecated option takes priority for backward compatibility
-		if spec.ReplicaLoadBalancer != nil {
-			c.logger.Debugf("The Postgres manifest for the cluster %v sets the deprecated `replicaLoadBalancer` param. Consider using the `enableReplicaLoadBalancer` instead.", c.Name)
-			return *spec.ReplicaLoadBalancer
-		}
-
 		// if the value is explicitly set in a Postgresql manifest, follow this setting
 		if spec.EnableReplicaLoadBalancer != nil {
 			return *spec.EnableReplicaLoadBalancer
@@ -699,16 +693,8 @@ func (c *Cluster) shouldCreateLoadBalancerForService(role PostgresRole, spec *sp
 
 	case Master:
 
-		if spec.UseLoadBalancer != nil {
-			c.logger.Debugf("The Postgres manifest for the cluster %v sets the deprecated `useLoadBalancer` param. Consider using the `enableMasterLoadBalancer` instead.", c.Name)
-			return *spec.UseLoadBalancer
-		}
-
-		// `enable_load_balancer`` governs LB for a master service
-		// there is no equivalent deprecated operator option for the replica LB
-		if c.OpConfig.EnableLoadBalancer != nil {
-			c.logger.Debugf("The operator configmap sets the deprecated `enable_load_balancer` param. Consider using the `enable_master_load_balancer` or `enable_replica_load_balancer` instead.")
-			return *c.OpConfig.EnableLoadBalancer
+		if spec.EnableMasterLoadBalancer != nil {
+			return *spec.EnableMasterLoadBalancer
 		}
 
 		return c.OpConfig.EnableMasterLoadBalancer

pkg/cluster/k8sres_test.go

@@ -65,23 +65,6 @@ func TestCreateLoadBalancerLogic(t *testing.T) {
 			opConfig: config.Config{EnableReplicaLoadBalancer: false},
 			result:   false,
 		},
-		{
-			subtest:  "old format, load balancer is enabled for replica",
-			role:     Replica,
-			spec:     &spec.PostgresSpec{ReplicaLoadBalancer: True()},
-			opConfig: config.Config{},
-			result:   true,
-		},
-		{
-			subtest: "old format has priority",
-			role:    Replica,
-			spec: &spec.PostgresSpec{
-				ReplicaLoadBalancer:       True(),
-				EnableReplicaLoadBalancer: False(),
-			},
-			opConfig: config.Config{},
-			result:   true,
-		},
 	}
 	for _, tt := range tests {
 		cluster.OpConfig = tt.opConfig

pkg/cluster/resources.go

@@ -344,10 +344,16 @@ func (c *Cluster) deleteService(role PostgresRole) error {
 }
 
 func (c *Cluster) createEndpoint(role PostgresRole) (*v1.Endpoints, error) {
+	var (
+		subsets []v1.EndpointSubset
+	)
 	c.setProcessName("creating endpoint")
-	subsets := make([]v1.EndpointSubset, 0)
-	if role == Master {
-		//TODO: set subsets to the master
+	if !c.isNewCluster() {
+		subsets = c.generateEndpointSubsets(role)
+	} else {
+		// Patroni will populate the master endpoint for the new cluster
+		// The replica endpoint will be filled-in by the service selector.
+		subsets = make([]v1.EndpointSubset, 0)
 	}
 	endpointsSpec := c.generateEndpoint(role, subsets)
 
@@ -361,6 +367,34 @@ func (c *Cluster) createEndpoint(role PostgresRole) (*v1.Endpoints, error) {
 	return endpoints, nil
 }
 
+func (c *Cluster) generateEndpointSubsets(role PostgresRole) []v1.EndpointSubset {
+	result := make([]v1.EndpointSubset, 0)
+	pods, err := c.getRolePods(role)
+	if err != nil {
+		if role == Master {
+			c.logger.Warningf("could not obtain the address for %s pod: %v", role, err)
+		} else {
+			c.logger.Warningf("could not obtain the addresses for %s pods: %v", role, err)
+		}
+		return result
+	}
+
+	endPointAddresses := make([]v1.EndpointAddress, 0)
+	for _, pod := range pods {
+		endPointAddresses = append(endPointAddresses, v1.EndpointAddress{IP: pod.Status.PodIP})
+	}
+	if len(endPointAddresses) > 0 {
+		result = append(result, v1.EndpointSubset{
+			Addresses: endPointAddresses,
+			Ports:     []v1.EndpointPort{{"postgresql", 5432, "TCP"}},
+		})
+	} else if role == Master {
+		c.logger.Warningf("master is not running, generated master endpoint does not contain any addresses")
+	}
+
+	return result
+}
+
 func (c *Cluster) createPodDisruptionBudget() (*policybeta1.PodDisruptionBudget, error) {
 	podDisruptionBudgetSpec := c.generatePodDisruptionBudget()
 	podDisruptionBudget, err := c.KubeClient.