Implement replicaLoadBalancer boolean flag. (#38)

The flag adds a replica service with the name cluster_name-repl and
a DNS name that defaults to {cluster}-repl.{team}.{hostedzone}.

The implementation converted Service field of the cluster into a map
with one or two elements and deals with the cases when the new flag
is changed on a running cluster
(the update and the sync should create or delete the replica service).
In order to pick up master and replica service and master endpoint
when listing cluster resources.

* Update the spec when updating the cluster.

pkg/cluster/cluster.go

@@ -34,7 +34,7 @@ var (
 	userRegexp         = regexp.MustCompile(`^[a-z0-9]([-_a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-_a-z0-9]*[a-z0-9])?)*$`)
 )
 
-//TODO: remove struct duplication
+// Config contains operator-wide clients and configuration used from a cluster. TODO: remove struct duplication.
 type Config struct {
 	KubeClient          *kubernetes.Clientset //TODO: move clients to the better place?
 	RestClient          *rest.RESTClient
@@ -45,7 +45,7 @@ type Config struct {
 }
 
 type kubeResources struct {
-	Service     *v1.Service
+	Service     map[PostgresRole]*v1.Service
 	Endpoint    *v1.Endpoints
 	Secrets     map[types.UID]*v1.Secret
 	Statefulset *v1beta1.StatefulSet
@@ -77,9 +77,10 @@ type compareStatefulsetResult struct {
 	reasons       []string
 }
 
+// New creates a new cluster. This function should be called from a controller.
 func New(cfg Config, pgSpec spec.Postgresql, logger *logrus.Entry) *Cluster {
 	lg := logger.WithField("pkg", "cluster").WithField("cluster-name", pgSpec.Metadata.Name)
-	kubeResources := kubeResources{Secrets: make(map[types.UID]*v1.Secret)}
+	kubeResources := kubeResources{Secrets: make(map[types.UID]*v1.Secret), Service: make(map[PostgresRole]*v1.Service)}
 	orphanDependents := true
 
 	podEventsQueue := cache.NewFIFO(func(obj interface{}) (string, error) {
@@ -108,7 +109,7 @@ func New(cfg Config, pgSpec spec.Postgresql, logger *logrus.Entry) *Cluster {
 	return cluster
 }
 
-func (c *Cluster) ClusterName() spec.NamespacedName {
+func (c *Cluster) clusterName() spec.NamespacedName {
 	return util.NameFromMeta(c.Metadata)
 }
 
@@ -136,7 +137,7 @@ func (c *Cluster) setStatus(status spec.PostgresStatus) {
 	}
 
 	if err != nil {
-		c.logger.Warningf("could not set status for cluster '%s': %s", c.ClusterName(), err)
+		c.logger.Warningf("could not set status for cluster '%s': %s", c.clusterName(), err)
 	}
 }
 
@@ -155,11 +156,10 @@ func (c *Cluster) initUsers() error {
 		return fmt.Errorf("could not init human users: %v", err)
 	}
 
-	c.logger.Debugf("Initialized users: %# v", util.Pretty(c.pgUsers))
-
 	return nil
 }
 
+// Create creates the new kubernetes objects associated with the cluster.
 func (c *Cluster) Create() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -182,11 +182,16 @@ func (c *Cluster) Create() error {
 	}
 	c.logger.Infof("endpoint '%s' has been successfully created", util.NameFromMeta(ep.ObjectMeta))
 
-	service, err := c.createService()
-	if err != nil {
-		return fmt.Errorf("could not create service: %v", err)
+	for _, role := range []PostgresRole{Master, Replica} {
+		if role == Replica && !c.Spec.ReplicaLoadBalancer {
+			continue
+		}
+		service, err := c.createService(role)
+		if err != nil {
+			return fmt.Errorf("could not create %s service: %v", role, err)
+		}
+		c.logger.Infof("%s service '%s' has been successfully created", role, util.NameFromMeta(service.ObjectMeta))
 	}
-	c.logger.Infof("service '%s' has been successfully created", util.NameFromMeta(service.ObjectMeta))
 
 	if err = c.initUsers(); err != nil {
 		return err
@@ -226,7 +231,7 @@ func (c *Cluster) Create() error {
 		}
 	}
 
-	err = c.ListResources()
+	err = c.listResources()
 	if err != nil {
 		c.logger.Errorf("could not list resources: %s", err)
 	}
@@ -234,14 +239,19 @@ func (c *Cluster) Create() error {
 	return nil
 }
 
-func (c *Cluster) sameServiceWith(service *v1.Service) (match bool, reason string) {
+func (c *Cluster) sameServiceWith(role PostgresRole, service *v1.Service) (match bool, reason string) {
 	//TODO: improve comparison
-	if !reflect.DeepEqual(c.Service.Spec.LoadBalancerSourceRanges, service.Spec.LoadBalancerSourceRanges) {
-		reason = "new service's LoadBalancerSourceRange doesn't match the current one"
-	} else {
-		match = true
+	match = true
+	old := c.Service[role].Spec.LoadBalancerSourceRanges
+	new := service.Spec.LoadBalancerSourceRanges
+	/* work around Kubernetes 1.6 serializing [] as nil. See https://github.com/kubernetes/kubernetes/issues/43203 */
+	if (len(old) == 0) && (len(new) == 0) {
+		return true, ""
 	}
-	return
+	if !reflect.DeepEqual(old, new) {
+		return false, fmt.Sprintf("new %s service's LoadBalancerSourceRange doesn't match the current one", role)
+	}
+	return true, ""
 }
 
 func (c *Cluster) sameVolumeWith(volume spec.Volume) (match bool, reason string) {
@@ -377,6 +387,8 @@ func compareResoucesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.Resource
 
 }
 
+// Update changes Kubernetes objects according to the new specification. Unlike the sync case, the missing object.
+// (i.e. service) is treated as an error.
 func (c *Cluster) Update(newSpec *spec.Postgresql) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -385,14 +397,46 @@ func (c *Cluster) Update(newSpec *spec.Postgresql) error {
 	c.logger.Debugf("Cluster update from version %s to %s",
 		c.Metadata.ResourceVersion, newSpec.Metadata.ResourceVersion)
 
-	newService := c.genService(newSpec.Spec.AllowedSourceRanges)
-	if match, reason := c.sameServiceWith(newService); !match {
-		c.logServiceChanges(c.Service, newService, true, reason)
-		if err := c.updateService(newService); err != nil {
-			c.setStatus(spec.ClusterStatusUpdateFailed)
-			return fmt.Errorf("could not update service: %v", err)
+	/* Make sure we update when this function exists */
+	defer func() {
+		c.Postgresql = *newSpec
+	}()
+
+	for _, role := range []PostgresRole{Master, Replica} {
+		if role == Replica {
+			if !newSpec.Spec.ReplicaLoadBalancer {
+				// old spec had a load balancer, but the new one doesn't
+				if c.Spec.ReplicaLoadBalancer {
+					err := c.deleteService(role)
+					if err != nil {
+						return fmt.Errorf("could not delete obsolete %s service: %v", role, err)
+					}
+					c.logger.Infof("deleted obsolete %s service", role)
+				}
+			} else {
+				if !c.Spec.ReplicaLoadBalancer {
+					// old spec didn't have a load balancer, but the one does
+					service, err := c.createService(role)
+					if err != nil {
+						return fmt.Errorf("could not create new %s service: %v", role, err)
+					}
+					c.logger.Infof("%s service '%s' has been created", role, util.NameFromMeta(service.ObjectMeta))
+				}
+			}
+			// only proceeed further if both old and new load balancer were present
+			if !(newSpec.Spec.ReplicaLoadBalancer && c.Spec.ReplicaLoadBalancer) {
+				continue
+			}
+		}
+		newService := c.genService(role, newSpec.Spec.AllowedSourceRanges)
+		if match, reason := c.sameServiceWith(role, newService); !match {
+			c.logServiceChanges(role, c.Service[role], newService, true, reason)
+			if err := c.updateService(role, newService); err != nil {
+				c.setStatus(spec.ClusterStatusUpdateFailed)
+				return fmt.Errorf("could not update %s service: %v", role, err)
+			}
+			c.logger.Infof("%s service '%s' has been updated", role, util.NameFromMeta(c.Service[role].ObjectMeta))
 		}
-		c.logger.Infof("service '%s' has been updated", util.NameFromMeta(c.Service.ObjectMeta))
 	}
 
 	newStatefulSet, err := c.genStatefulSet(newSpec.Spec)
@@ -448,6 +492,7 @@ func (c *Cluster) Update(newSpec *spec.Postgresql) error {
 	return nil
 }
 
+// Delete deletes the cluster and cleans up all objects associated with it (including statefulsets).
 func (c *Cluster) Delete() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -456,8 +501,13 @@ func (c *Cluster) Delete() error {
 		return fmt.Errorf("could not delete endpoint: %v", err)
 	}
 
-	if err := c.deleteService(); err != nil {
-		return fmt.Errorf("could not delete service: %v", err)
+	for _, role := range []PostgresRole{Master, Replica} {
+		if role == Replica && !c.Spec.ReplicaLoadBalancer {
+			continue
+		}
+		if err := c.deleteService(role); err != nil {
+			return fmt.Errorf("could not delete %s service: %v", role, err)
+		}
 	}
 
 	if err := c.deleteStatefulSet(); err != nil {
@@ -473,6 +523,7 @@ func (c *Cluster) Delete() error {
 	return nil
 }
 
+// ReceivePodEvent is called back by the controller in order to add the cluster's pod event to the queue.
 func (c *Cluster) ReceivePodEvent(event spec.PodEvent) {
 	c.podEventsQueue.Add(event)
 }
@@ -493,6 +544,7 @@ func (c *Cluster) processPodEvent(obj interface{}) error {
 	return nil
 }
 
+// Run starts the pod event dispatching for the given cluster.
 func (c *Cluster) Run(stopCh <-chan struct{}) {
 	go c.processPodEventQueue(stopCh)
 }

pkg/cluster/k8sres.go

@@ -15,9 +15,9 @@ import (
 )
 
 const (
-	PGBinariesLocationTemplate       = "/usr/lib/postgresql/%s/bin"
-	PatroniPGBinariesParameterName   = "pg_bin"
-	PatroniPGParametersParameterName = "parameters"
+	pgBinariesLocationTemplate       = "/usr/lib/postgresql/%s/bin"
+	patroniPGBinariesParameterName   = "bin_dir"
+	patroniPGParametersParameterName = "parameters"
 )
 
 type pgUser struct {
@@ -25,7 +25,7 @@ type pgUser struct {
 	Options  []string `json:"options"`
 }
 
-type PatroniDCS struct {
+type patroniDCS struct {
 	TTL                  uint32  `json:"ttl,omitempty"`
 	LoopWait             uint32  `json:"loop_wait,omitempty"`
 	RetryTimeout         uint32  `json:"retry_timeout,omitempty"`
@@ -36,7 +36,7 @@ type pgBootstrap struct {
 	Initdb []interface{}     `json:"initdb"`
 	Users  map[string]pgUser `json:"users"`
 	PgHBA  []string          `json:"pg_hba"`
-	DCS    PatroniDCS        `json:"dcs,omitempty"`
+	DCS    patroniDCS        `json:"dcs,omitempty"`
 }
 
 type spiloConfiguration struct {
@@ -185,9 +185,9 @@ PATRONI_INITDB_PARAMS:
 	}
 
 	config.PgLocalConfiguration = make(map[string]interface{})
-	config.PgLocalConfiguration[PatroniPGBinariesParameterName] = fmt.Sprintf(PGBinariesLocationTemplate, pg.PgVersion)
+	config.PgLocalConfiguration[patroniPGBinariesParameterName] = fmt.Sprintf(pgBinariesLocationTemplate, pg.PgVersion)
 	if len(pg.Parameters) > 0 {
-		config.PgLocalConfiguration[PatroniPGParametersParameterName] = pg.Parameters
+		config.PgLocalConfiguration[patroniPGParametersParameterName] = pg.Parameters
 	}
 	config.Bootstrap.Users = map[string]pgUser{
 		c.OpConfig.PamRoleName: {
@@ -425,14 +425,22 @@ func (c *Cluster) genSingleUserSecret(namespace string, pgUser spec.PgUser) *v1.
 	return &secret
 }
 
-func (c *Cluster) genService(allowedSourceRanges []string) *v1.Service {
+func (c *Cluster) genService(role PostgresRole, allowedSourceRanges []string) *v1.Service {
+
+	dnsNameFunction := c.masterDnsName
+	name := c.Metadata.Name
+	if role == Replica {
+		dnsNameFunction = c.replicaDnsName
+		name = name + "-repl"
+	}
+
 	service := &v1.Service{
 		ObjectMeta: v1.ObjectMeta{
-			Name:      c.Metadata.Name,
+			Name:      name,
 			Namespace: c.Metadata.Namespace,
-			Labels:    c.labelsSet(),
+			Labels:    c.roleLabelsSet(role),
 			Annotations: map[string]string{
-				constants.ZalandoDNSNameAnnotation: c.dnsName(),
+				constants.ZalandoDNSNameAnnotation: dnsNameFunction(),
 				constants.ElbTimeoutAnnotationName: constants.ElbTimeoutAnnotationValue,
 			},
 		},
@@ -442,16 +450,19 @@ func (c *Cluster) genService(allowedSourceRanges []string) *v1.Service {
 			LoadBalancerSourceRanges: allowedSourceRanges,
 		},
 	}
+	if role == Replica {
+		service.Spec.Selector = map[string]string{c.OpConfig.PodRoleLabel: string(Replica)}
+	}
 
 	return service
 }
 
-func (c *Cluster) genEndpoints() *v1.Endpoints {
+func (c *Cluster) genMasterEndpoints() *v1.Endpoints {
 	endpoints := &v1.Endpoints{
 		ObjectMeta: v1.ObjectMeta{
 			Name:      c.Metadata.Name,
 			Namespace: c.Metadata.Namespace,
-			Labels:    c.labelsSet(),
+			Labels:    c.roleLabelsSet(Master),
 		},
 	}
 

pkg/cluster/resources.go

@@ -24,20 +24,31 @@ func (c *Cluster) loadResources() error {
 	if err != nil {
 		return fmt.Errorf("could not get list of services: %v", err)
 	}
-	if len(services.Items) > 1 {
+	if len(services.Items) > 2 {
 		return fmt.Errorf("too many(%d) services for a cluster", len(services.Items))
-	} else if len(services.Items) == 1 {
-		c.Service = &services.Items[0]
+	}
+	for i, svc := range services.Items {
+		switch PostgresRole(svc.Labels[c.OpConfig.PodRoleLabel]) {
+		case Replica:
+			c.Service[Replica] = &services.Items[i]
+		default:
+			c.Service[Master] = &services.Items[i]
+		}
 	}
 
 	endpoints, err := c.KubeClient.Endpoints(ns).List(listOptions)
 	if err != nil {
 		return fmt.Errorf("could not get list of endpoints: %v", err)
 	}
-	if len(endpoints.Items) > 1 {
+	if len(endpoints.Items) > 2 {
 		return fmt.Errorf("too many(%d) endpoints for a cluster", len(endpoints.Items))
-	} else if len(endpoints.Items) == 1 {
-		c.Endpoint = &endpoints.Items[0]
+	}
+
+	for i, ep := range endpoints.Items {
+		if ep.Labels[c.OpConfig.PodRoleLabel] != string(Replica) {
+			c.Endpoint = &endpoints.Items[i]
+			break
+		}
 	}
 
 	secrets, err := c.KubeClient.Secrets(ns).List(listOptions)
@@ -58,14 +69,15 @@ func (c *Cluster) loadResources() error {
 	}
 	if len(statefulSets.Items) > 1 {
 		return fmt.Errorf("too many(%d) statefulsets for a cluster", len(statefulSets.Items))
-	} else if len(statefulSets.Items) == 1 {
+	}
+	if len(statefulSets.Items) == 1 {
 		c.Statefulset = &statefulSets.Items[0]
 	}
 
 	return nil
 }
 
-func (c *Cluster) ListResources() error {
+func (c *Cluster) listResources() error {
 	if c.Statefulset != nil {
 		c.logger.Infof("Found statefulset: %s (uid: %s)", util.NameFromMeta(c.Statefulset.ObjectMeta), c.Statefulset.UID)
 	}
@@ -78,8 +90,8 @@ func (c *Cluster) ListResources() error {
 		c.logger.Infof("Found endpoint: %s (uid: %s)", util.NameFromMeta(c.Endpoint.ObjectMeta), c.Endpoint.UID)
 	}
 
-	if c.Service != nil {
-		c.logger.Infof("Found service: %s (uid: %s)", util.NameFromMeta(c.Service.ObjectMeta), c.Service.UID)
+	for role, service := range c.Service {
+		c.logger.Infof("Found %s service: %s (uid: %s)", role, util.NameFromMeta(service.ObjectMeta), service.UID)
 	}
 
 	pods, err := c.listPods()
@@ -217,57 +229,56 @@ func (c *Cluster) deleteStatefulSet() error {
 	return nil
 }
 
-func (c *Cluster) createService() (*v1.Service, error) {
-	if c.Service != nil {
+func (c *Cluster) createService(role PostgresRole) (*v1.Service, error) {
+	if c.Service[role] != nil {
 		return nil, fmt.Errorf("service already exists in the cluster")
 	}
-	serviceSpec := c.genService(c.Spec.AllowedSourceRanges)
+	serviceSpec := c.genService(role, c.Spec.AllowedSourceRanges)
 
 	service, err := c.KubeClient.Services(serviceSpec.Namespace).Create(serviceSpec)
 	if err != nil {
 		return nil, err
 	}
-	c.Service = service
 
+	c.Service[role] = service
 	return service, nil
 }
 
-func (c *Cluster) updateService(newService *v1.Service) error {
-	if c.Service == nil {
+func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error {
+	if c.Service[role] == nil {
 		return fmt.Errorf("there is no service in the cluster")
 	}
-	serviceName := util.NameFromMeta(c.Service.ObjectMeta)
+	serviceName := util.NameFromMeta(c.Service[role].ObjectMeta)
 
 	patchData, err := specPatch(newService.Spec)
 	if err != nil {
 		return fmt.Errorf("could not form patch for the service '%s': %v", serviceName, err)
 	}
 
-	svc, err := c.KubeClient.Services(c.Service.Namespace).Patch(
-		c.Service.Name,
+	svc, err := c.KubeClient.Services(c.Service[role].Namespace).Patch(
+		c.Service[role].Name,
 		api.MergePatchType,
 		patchData, "")
 	if err != nil {
 		return fmt.Errorf("could not patch service '%s': %v", serviceName, err)
 	}
-	c.Service = svc
+	c.Service[role] = svc
 
 	return nil
 }
 
-func (c *Cluster) deleteService() error {
-	c.logger.Debugln("Deleting service")
-
-	if c.Service == nil {
-		return fmt.Errorf("there is no service in the cluster")
+func (c *Cluster) deleteService(role PostgresRole) error {
+	c.logger.Debugf("Deleting service %s", role)
+	if c.Service[role] == nil {
+		return fmt.Errorf("There is no %s service in the cluster", role)
 	}
-	err := c.KubeClient.Services(c.Service.Namespace).Delete(c.Service.Name, c.deleteOptions)
+	service := c.Service[role]
+	err := c.KubeClient.Services(service.Namespace).Delete(service.Name, c.deleteOptions)
 	if err != nil {
 		return err
 	}
-	c.logger.Infof("service '%s' has been deleted", util.NameFromMeta(c.Service.ObjectMeta))
-	c.Service = nil
-
+	c.logger.Infof("%s service '%s' has been deleted", role, util.NameFromMeta(service.ObjectMeta))
+	c.Service[role] = nil
 	return nil
 }
 
@@ -275,7 +286,7 @@ func (c *Cluster) createEndpoint() (*v1.Endpoints, error) {
 	if c.Endpoint != nil {
 		return nil, fmt.Errorf("endpoint already exists in the cluster")
 	}
-	endpointsSpec := c.genEndpoints()
+	endpointsSpec := c.genMasterEndpoints()
 
 	endpoints, err := c.KubeClient.Endpoints(endpointsSpec.Namespace).Create(endpointsSpec)
 	if err != nil {

pkg/cluster/sync.go

@@ -8,6 +8,8 @@ import (
 	"github.com/zalando-incubator/postgres-operator/pkg/util/volumes"
 )
 
+// Sync syncs the cluster, making sure the actual Kubernetes objects correspond to what is defined in the manifest.
+// Unlike the update, sync does not error out if some objects do not exist and takes care of creating them.
 func (c *Cluster) Sync() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -32,9 +34,20 @@ func (c *Cluster) Sync() error {
 	}
 
 	c.logger.Debugf("Syncing services")
-	if err := c.syncService(); err != nil {
-		if !k8sutil.ResourceAlreadyExists(err) {
-			return fmt.Errorf("coud not sync services: %v", err)
+	for _, role := range []PostgresRole{Master, Replica} {
+		if role == Replica && !c.Spec.ReplicaLoadBalancer {
+			if c.Service[role] != nil {
+				// delete the left over replica service
+				if err := c.deleteService(role); err != nil {
+					return fmt.Errorf("could not delete obsolete %s service: %v", role)
+				}
+			}
+			continue
+		}
+		if err := c.syncService(role); err != nil {
+			if !k8sutil.ResourceAlreadyExists(err) {
+				return fmt.Errorf("coud not sync %s service: %v", role, err)
+			}
 		}
 	}
 
@@ -48,16 +61,15 @@ func (c *Cluster) Sync() error {
 	if !c.databaseAccessDisabled() {
 		if err := c.initDbConn(); err != nil {
 			return fmt.Errorf("could not init db connection: %v", err)
-		} else {
-			c.logger.Debugf("Syncing roles")
-			if err := c.SyncRoles(); err != nil {
-				return fmt.Errorf("could not sync roles: %v", err)
-			}
+		}
+		c.logger.Debugf("Syncing roles")
+		if err := c.syncRoles(); err != nil {
+			return fmt.Errorf("could not sync roles: %v", err)
 		}
 	}
 
 	c.logger.Debugf("Syncing persistent volumes")
-	if err := c.SyncVolumes(); err != nil {
+	if err := c.syncVolumes(); err != nil {
 		return fmt.Errorf("could not sync persistent volumes: %v", err)
 	}
 
@@ -75,30 +87,30 @@ func (c *Cluster) syncSecrets() error {
 	return err
 }
 
-func (c *Cluster) syncService() error {
+func (c *Cluster) syncService(role PostgresRole) error {
 	cSpec := c.Spec
-	if c.Service == nil {
-		c.logger.Infof("could not find the cluster's service")
-		svc, err := c.createService()
+	if c.Service[role] == nil {
+		c.logger.Infof("could not find the cluster's %s service", role)
+		svc, err := c.createService(role)
 		if err != nil {
-			return fmt.Errorf("could not create missing service: %v", err)
+			return fmt.Errorf("could not create missing %s service: %v", role, err)
 		}
-		c.logger.Infof("Created missing service '%s'", util.NameFromMeta(svc.ObjectMeta))
+		c.logger.Infof("Created missing %s service '%s'", role, util.NameFromMeta(svc.ObjectMeta))
 
 		return nil
 	}
 
-	desiredSvc := c.genService(cSpec.AllowedSourceRanges)
-	match, reason := c.sameServiceWith(desiredSvc)
+	desiredSvc := c.genService(role, cSpec.AllowedSourceRanges)
+	match, reason := c.sameServiceWith(role, desiredSvc)
 	if match {
 		return nil
 	}
-	c.logServiceChanges(c.Service, desiredSvc, false, reason)
+	c.logServiceChanges(role, c.Service[role], desiredSvc, false, reason)
 
-	if err := c.updateService(desiredSvc); err != nil {
-		return fmt.Errorf("could not update service to match desired state: %v", err)
+	if err := c.updateService(role, desiredSvc); err != nil {
+		return fmt.Errorf("could not update %s service to match desired state: %v", role, err)
 	}
-	c.logger.Infof("service '%s' is in the desired state now", util.NameFromMeta(desiredSvc.ObjectMeta))
+	c.logger.Infof("%s service '%s' is in the desired state now", role, util.NameFromMeta(desiredSvc.ObjectMeta))
 
 	return nil
 }
@@ -181,7 +193,7 @@ func (c *Cluster) syncStatefulSet() error {
 	return nil
 }
 
-func (c *Cluster) SyncRoles() error {
+func (c *Cluster) syncRoles() error {
 	var userNames []string
 
 	if err := c.initUsers(); err != nil {
@@ -201,9 +213,9 @@ func (c *Cluster) SyncRoles() error {
 	return nil
 }
 
-/* SyncVolume reads all persistent volumes and checks that their size matches the one declared in the statefulset */
-func (c *Cluster) SyncVolumes() error {
-	act, err := c.VolumesNeedResizing(c.Spec.Volume)
+// syncVolumes reads all persistent volumes and checks that their size matches the one declared in the statefulset.
+func (c *Cluster) syncVolumes() error {
+	act, err := c.volumesNeedResizing(c.Spec.Volume)
 	if err != nil {
 		return fmt.Errorf("could not compare size of the volumes: %v", err)
 	}

pkg/cluster/types.go

@@ -0,0 +1,8 @@
+package cluster
+
+type PostgresRole string
+
+const (
+	Master  PostgresRole = "master"
+	Replica PostgresRole = "replica"
+)

pkg/cluster/util.go

@@ -82,14 +82,14 @@ func (c *Cluster) logStatefulSetChanges(old, new *v1beta1.StatefulSet, isUpdate
 	}
 }
 
-func (c *Cluster) logServiceChanges(old, new *v1.Service, isUpdate bool, reason string) {
+func (c *Cluster) logServiceChanges(role PostgresRole, old, new *v1.Service, isUpdate bool, reason string) {
 	if isUpdate {
-		c.logger.Infof("service '%s' has been changed",
-			util.NameFromMeta(old.ObjectMeta),
+		c.logger.Infof("%s service '%s' has been changed",
+			role, util.NameFromMeta(old.ObjectMeta),
 		)
 	} else {
-		c.logger.Infof("service '%s  is not in the desired state and needs to be updated",
-			util.NameFromMeta(old.ObjectMeta),
+		c.logger.Infof("%s service '%s  is not in the desired state and needs to be updated",
+			role, util.NameFromMeta(old.ObjectMeta),
 		)
 	}
 	c.logger.Debugf("diff\n%s\n", util.PrettyDiff(old.Spec, new.Spec))
@@ -145,7 +145,6 @@ func (c *Cluster) getTeamMembers() ([]string, error) {
 	if err != nil {
 		return nil, fmt.Errorf("could not get team info: %v", err)
 	}
-	c.logger.Debugf("Got from the Team API: %+v", *teamInfo)
 
 	return teamInfo.Members, nil
 }
@@ -263,14 +262,30 @@ func (c *Cluster) waitStatefulsetPodsReady() error {
 }
 
 func (c *Cluster) labelsSet() labels.Set {
-	lbls := c.OpConfig.ClusterLabels
+	lbls := make(map[string]string)
+	for k, v := range c.OpConfig.ClusterLabels {
+		lbls[k] = v
+	}
 	lbls[c.OpConfig.ClusterNameLabel] = c.Metadata.Name
 
 	return labels.Set(lbls)
 }
 
-func (c *Cluster) dnsName() string {
-	return strings.ToLower(c.OpConfig.DNSNameFormat.Format(
+func (c *Cluster) roleLabelsSet(role PostgresRole) labels.Set {
+	lbls := c.labelsSet()
+	lbls[c.OpConfig.PodRoleLabel] = string(role)
+	return lbls
+}
+
+func (c *Cluster) masterDnsName() string {
+	return strings.ToLower(c.OpConfig.MasterDNSNameFormat.Format(
+		"cluster", c.Spec.ClusterName,
+		"team", c.teamName(),
+		"hostedzone", c.OpConfig.DbHostedZone))
+}
+
+func (c *Cluster) replicaDnsName() string {
+	return strings.ToLower(c.OpConfig.ReplicaDNSNameFormat.Format(
 		"cluster", c.Spec.ClusterName,
 		"team", c.teamName(),
 		"hostedzone", c.OpConfig.DbHostedZone))

pkg/cluster/volumes.go

@@ -60,15 +60,14 @@ func (c *Cluster) listPersistentVolumes() ([]*v1.PersistentVolume, error) {
 	for _, pvc := range pvcs {
 		lastDash := strings.LastIndex(pvc.Name, "-")
 		if lastDash > 0 && lastDash < len(pvc.Name)-1 {
-			if pvcNumber, err := strconv.Atoi(pvc.Name[lastDash+1:]); err != nil {
+			pvcNumber, err := strconv.Atoi(pvc.Name[lastDash+1:])
+			if err != nil {
 				return nil, fmt.Errorf("could not convert last part of the persistent volume claim name %s to a number", pvc.Name)
-			} else {
-				if int32(pvcNumber) > lastPodIndex {
-					c.logger.Debugf("Skipping persistent volume %s corresponding to a non-running pods", pvc.Name)
-					continue
-				}
 			}
-
+			if int32(pvcNumber) > lastPodIndex {
+				c.logger.Debugf("Skipping persistent volume %s corresponding to a non-running pods", pvc.Name)
+				continue
+			}
 		}
 		pv, err := c.KubeClient.PersistentVolumes().Get(pvc.Spec.VolumeName)
 		if err != nil {
@@ -139,7 +138,7 @@ func (c *Cluster) resizeVolumes(newVolume spec.Volume, resizers []volumes.Volume
 	return nil
 }
 
-func (c *Cluster) VolumesNeedResizing(newVolume spec.Volume) (bool, error) {
+func (c *Cluster) volumesNeedResizing(newVolume spec.Volume) (bool, error) {
 	volumes, manifestSize, err := c.listVolumesWithManifestSize(newVolume)
 	if err != nil {
 		return false, err