use sync methods while updating the cluster

pkg/cluster/cluster.go

@@ -28,7 +28,6 @@ import (
 	"github.com/zalando-incubator/postgres-operator/pkg/util/patroni"
 	"github.com/zalando-incubator/postgres-operator/pkg/util/teams"
 	"github.com/zalando-incubator/postgres-operator/pkg/util/users"
-	"github.com/zalando-incubator/postgres-operator/pkg/util/volumes"
 )
 
 var (
@@ -46,7 +45,7 @@ type Config struct {
 
 type kubeResources struct {
 	Services            map[PostgresRole]*v1.Service
-	Endpoint            *v1.Endpoints
+	Endpoints           map[PostgresRole]*v1.Endpoints
 	Secrets             map[types.UID]*v1.Secret
 	Statefulset         *v1beta1.StatefulSet
 	PodDisruptionBudget *policybeta1.PodDisruptionBudget
@@ -99,12 +98,15 @@ func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec spec.Postgresql
 	})
 
 	cluster := &Cluster{
-		Config:           cfg,
-		Postgresql:       pgSpec,
-		pgUsers:          make(map[string]spec.PgUser),
-		systemUsers:      make(map[string]spec.PgUser),
-		podSubscribers:   make(map[spec.NamespacedName]chan spec.PodEvent),
-		kubeResources:    kubeResources{Secrets: make(map[types.UID]*v1.Secret), Services: make(map[PostgresRole]*v1.Service)},
+		Config:         cfg,
+		Postgresql:     pgSpec,
+		pgUsers:        make(map[string]spec.PgUser),
+		systemUsers:    make(map[string]spec.PgUser),
+		podSubscribers: make(map[spec.NamespacedName]chan spec.PodEvent),
+		kubeResources: kubeResources{
+			Secrets:   make(map[types.UID]*v1.Secret),
+			Services:  make(map[PostgresRole]*v1.Service),
+			Endpoints: make(map[PostgresRole]*v1.Endpoints)},
 		masterLess:       false,
 		userSyncStrategy: users.DefaultUserSyncStrategy{},
 		deleteOptions:    &metav1.DeleteOptions{OrphanDependents: &orphanDependents},
@@ -203,17 +205,16 @@ func (c *Cluster) Create() error {
 
 	c.setStatus(spec.ClusterStatusCreating)
 
-	//service will create endpoint implicitly
-	ep, err = c.createEndpoint()
-	if err != nil {
-		return fmt.Errorf("could not create endpoint: %v", err)
-	}
-	c.logger.Infof("endpoint %q has been successfully created", util.NameFromMeta(ep.ObjectMeta))
-
 	for _, role := range []PostgresRole{Master, Replica} {
 		if role == Replica && !c.Spec.ReplicaLoadBalancer {
 			continue
 		}
+		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))
+
 		service, err = c.createService(role)
 		if err != nil {
 			return fmt.Errorf("could not create %s service: %v", role, err)
@@ -226,7 +227,7 @@ func (c *Cluster) Create() error {
 	}
 	c.logger.Infof("users have been initialized")
 
-	if err = c.applySecrets(); err != nil {
+	if err = c.syncSecrets(); err != nil {
 		return fmt.Errorf("could not create secrets: %v", err)
 	}
 	c.logger.Infof("secrets have been successfully created")
@@ -257,8 +258,8 @@ func (c *Cluster) Create() error {
 		}
 		c.logger.Infof("users have been successfully created")
 
-		if err = c.createDatabases(); err != nil {
-			return fmt.Errorf("could not create databases: %v", err)
+		if err = c.syncDatabases(); err != nil {
+			return fmt.Errorf("could not sync databases: %v", err)
 		}
 		c.logger.Infof("databases have been successfully created")
 	} else {
@@ -267,48 +268,13 @@ func (c *Cluster) Create() error {
 		}
 	}
 
-	err = c.listResources()
-	if err != nil {
+	if err := c.listResources(); err != nil {
 		c.logger.Errorf("could not list resources: %v", err)
 	}
 
 	return nil
 }
 
-func (c *Cluster) sameServiceWith(role PostgresRole, service *v1.Service) (match bool, reason string) {
-	//TODO: improve comparison
-	if c.Services[role].Spec.Type != service.Spec.Type {
-		return false, fmt.Sprintf("new %s service's type %q doesn't match the current one %q",
-			role, service.Spec.Type, c.Services[role].Spec.Type)
-	}
-	oldSourceRanges := c.Services[role].Spec.LoadBalancerSourceRanges
-	newSourceRanges := service.Spec.LoadBalancerSourceRanges
-	/* work around Kubernetes 1.6 serializing [] as nil. See https://github.com/kubernetes/kubernetes/issues/43203 */
-	if (len(oldSourceRanges) == 0) && (len(newSourceRanges) == 0) {
-		return true, ""
-	}
-	if !reflect.DeepEqual(oldSourceRanges, newSourceRanges) {
-		return false, fmt.Sprintf("new %s service's LoadBalancerSourceRange doesn't match the current one", role)
-	}
-
-	oldDNSAnnotation := c.Services[role].Annotations[constants.ZalandoDNSNameAnnotation]
-	newDNSAnnotation := service.Annotations[constants.ZalandoDNSNameAnnotation]
-	if oldDNSAnnotation != newDNSAnnotation {
-		return false, fmt.Sprintf("new %s service's %q annotation doesn't match the current one", role, constants.ZalandoDNSNameAnnotation)
-	}
-
-	return true, ""
-}
-
-func (c *Cluster) sameVolumeWith(volume spec.Volume) (match bool, reason string) {
-	if !reflect.DeepEqual(c.Spec.Volume, volume) {
-		reason = "new volume's specification doesn't match the current one"
-	} else {
-		match = true
-	}
-	return
-}
-
 func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *compareStatefulsetResult {
 	reasons := make([]string, 0)
 	var match, needsRollUpdate, needsReplace bool
@@ -406,6 +372,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *comp
 	if needsRollUpdate || needsReplace {
 		match = false
 	}
+
 	return &compareStatefulsetResult{match: match, reasons: reasons, rollingUpdate: needsRollUpdate, replace: needsReplace}
 }
 
@@ -417,12 +384,13 @@ func compareResources(a *v1.ResourceRequirements, b *v1.ResourceRequirements) (e
 	if equal && (b != nil) {
 		equal = compareResoucesAssumeFirstNotNil(b, a)
 	}
+
 	return
 }
 
 func compareResoucesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.ResourceRequirements) bool {
 	if b == nil || (len(b.Requests) == 0) {
-		return (len(a.Requests) == 0)
+		return len(a.Requests) == 0
 	}
 	for k, v := range a.Requests {
 		if (&v).Cmp(b.Requests[k]) != 0 {
@@ -440,108 +408,108 @@ 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 {
+func (c *Cluster) Update(oldSpec, newSpec *spec.Postgresql) error {
+	updateFailed := false
+
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
 	c.setStatus(spec.ClusterStatusUpdating)
+	c.Postgresql = *newSpec
 
-	/* Make sure we update when this function exits */
 	defer func() {
-		c.Postgresql = *newSpec
+		if updateFailed {
+			c.setStatus(spec.ClusterStatusUpdateFailed)
+		} else if c.Status != spec.ClusterStatusRunning {
+			c.setStatus(spec.ClusterStatusRunning)
+		}
 	}()
 
-	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 %q has been created", role, util.NameFromMeta(service.ObjectMeta))
-				}
-			}
-			// only proceed further if both old and new load balancer were present
-			if !(newSpec.Spec.ReplicaLoadBalancer && c.Spec.ReplicaLoadBalancer) {
-				continue
-			}
-		}
-		newService := c.generateService(role, &newSpec.Spec)
-		if match, reason := c.sameServiceWith(role, newService); !match {
-			c.logServiceChanges(role, c.Services[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 %q has been updated", role, util.NameFromMeta(c.Services[role].ObjectMeta))
+	if oldSpec.Spec.PgVersion != newSpec.Spec.PgVersion { // PG versions comparison
+		c.logger.Warningf("postgresql version change(%q -> %q) has no effect", oldSpec.Spec.PgVersion, newSpec.Spec.PgVersion)
+		//we need that hack to generate statefulset with the old version
+		newSpec.Spec.PgVersion = oldSpec.Spec.PgVersion
+	}
+
+	// Service
+	if !reflect.DeepEqual(c.generateService(Master, &oldSpec.Spec), c.generateService(Master, &newSpec.Spec)) ||
+		!reflect.DeepEqual(c.generateService(Replica, &oldSpec.Spec), c.generateService(Replica, &newSpec.Spec)) ||
+		oldSpec.Spec.ReplicaLoadBalancer != newSpec.Spec.ReplicaLoadBalancer {
+		c.logger.Debugf("syncing services")
+		if err := c.syncServices(); err != nil {
+			c.logger.Errorf("could not sync services: %v", err)
+			updateFailed = true
 		}
 	}
 
-	newStatefulSet, err := c.generateStatefulSet(newSpec.Spec)
-	if err != nil {
-		return fmt.Errorf("could not generate statefulset: %v", err)
-	}
-	cmp := c.compareStatefulSetWith(newStatefulSet)
+	if !reflect.DeepEqual(oldSpec.Spec.Users, newSpec.Spec.Users) {
+		c.logger.Debugf("syncing secrets")
+		if err := c.initUsers(); err != nil {
+			c.logger.Errorf("could not init users: %v", err)
+			updateFailed = true
+		}
 
-	if !cmp.match {
-		c.logStatefulSetChanges(c.Statefulset, newStatefulSet, true, cmp.reasons)
-		//TODO: mind the case of updating allowedSourceRanges
-		if !cmp.replace {
-			if err := c.updateStatefulSet(newStatefulSet); err != nil {
-				c.setStatus(spec.ClusterStatusUpdateFailed)
-				return fmt.Errorf("could not upate statefulset: %v", err)
-			}
-		} else {
-			if err := c.replaceStatefulSet(newStatefulSet); err != nil {
-				c.setStatus(spec.ClusterStatusUpdateFailed)
-				return fmt.Errorf("could not replace statefulset: %v", err)
+		c.logger.Debugf("syncing secrets")
+
+		//TODO: mind the secrets of the deleted/new users
+		if err := c.syncSecrets(); err != nil {
+			c.logger.Errorf("could not sync secrets: %v", err)
+			updateFailed = true
+		}
+
+		if !c.databaseAccessDisabled() {
+			c.logger.Debugf("syncing roles")
+			if err := c.syncRoles(true); err != nil {
+				c.logger.Errorf("could not sync roles: %v", err)
+				updateFailed = true
 			}
 		}
-		//TODO: if there is a change in numberOfInstances, make sure Pods have been created/deleted
-		c.logger.Infof("statefulset %q has been updated", util.NameFromMeta(c.Statefulset.ObjectMeta))
 	}
 
-	if c.Spec.PgVersion != newSpec.Spec.PgVersion { // PG versions comparison
-		c.logger.Warningf("postgresql version change(%q -> %q) is not allowed",
-			c.Spec.PgVersion, newSpec.Spec.PgVersion)
-		//TODO: rewrite pg version in tpr spec
-	}
+	// Volume
+	if oldSpec.Spec.Size != newSpec.Spec.Size {
+		c.logger.Debugf("syncing persistent volumes")
+		c.logVolumeChanges(oldSpec.Spec.Volume, newSpec.Spec.Volume)
 
-	if cmp.rollingUpdate {
-		c.logger.Infof("rolling update is needed")
-		// TODO: wait for actual streaming to the replica
-		if err := c.recreatePods(); err != nil {
-			c.setStatus(spec.ClusterStatusUpdateFailed)
-			return fmt.Errorf("could not recreate pods: %v", err)
+		if err := c.syncVolumes(); err != nil {
+			c.logger.Errorf("could not sync persistent volumes: %v", err)
+			updateFailed = true
 		}
-		c.logger.Infof("rolling update has been finished")
 	}
 
-	if match, reason := c.sameVolumeWith(newSpec.Spec.Volume); !match {
-		c.logVolumeChanges(c.Spec.Volume, newSpec.Spec.Volume, reason)
-		if err := c.resizeVolumes(newSpec.Spec.Volume, []volumes.VolumeResizer{&volumes.EBSVolumeResizer{}}); err != nil {
-			return fmt.Errorf("could not update volumes: %v", err)
+	// Statefulset
+	func() {
+		oldSs, err := c.generateStatefulSet(&oldSpec.Spec)
+		if err != nil {
+			c.logger.Errorf("could not generate old statefulset spec")
+			updateFailed = true
+			return
 		}
-		c.logger.Infof("volumes have been updated successfully")
-	}
 
-	if err := c.syncPodDisruptionBudget(true); err != nil {
-		c.setStatus(spec.ClusterStatusUpdateFailed)
-		return fmt.Errorf("could not update pod disruption budget: %v", err)
-	}
+		newSs, err := c.generateStatefulSet(&newSpec.Spec)
+		if err != nil {
+			c.logger.Errorf("could not generate new statefulset spec")
+			updateFailed = true
+			return
+		}
 
-	c.setStatus(spec.ClusterStatusRunning)
+		if !reflect.DeepEqual(oldSs, newSs) {
+			c.logger.Debugf("syncing statefulsets")
+			if err := c.syncStatefulSet(); err != nil {
+				c.logger.Errorf("could not sync statefulsets: %v", err)
+				updateFailed = true
+			}
+		}
+	}()
+
+	// Databases
+	if !reflect.DeepEqual(oldSpec.Spec.Databases, newSpec.Spec.Databases) {
+		c.logger.Infof("syncing databases")
+		if err := c.syncDatabases(); err != nil {
+			c.logger.Errorf("could not sync databases: %v", err)
+			updateFailed = true
+		}
+	}
 
 	return nil
 }
@@ -551,14 +519,15 @@ func (c *Cluster) Delete() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	if err := c.deleteEndpoint(); err != nil {
-		return fmt.Errorf("could not delete endpoint: %v", err)
-	}
-
 	for _, role := range []PostgresRole{Master, Replica} {
 		if role == Replica && !c.Spec.ReplicaLoadBalancer {
 			continue
 		}
+
+		if err := c.deleteEndpoint(role); err != nil {
+			return fmt.Errorf("could not delete %s endpoint: %v", role, err)
+		}
+
 		if err := c.deleteService(role); err != nil {
 			return fmt.Errorf("could not delete %s service: %v", role, err)
 		}
@@ -715,7 +684,8 @@ func (c *Cluster) GetStatus() *spec.ClusterStatus {
 
 		MasterService:       c.GetServiceMaster(),
 		ReplicaService:      c.GetServiceReplica(),
-		Endpoint:            c.GetEndpoint(),
+		MasterEndpoint:      c.GetEndpointMaster(),
+		ReplicaEndpoint:     c.GetEndpointReplica(),
 		StatefulSet:         c.GetStatefulSet(),
 		PodDisruptionBudget: c.GetPodDisruptionBudget(),
 		CurrentProcess:      c.GetCurrentProcess(),