add unit test and documentation for finalizers (#2509)

* add unit test and documentation for finalizers
* error msg with lower case and cover sync case
* try to avoid adding json-patch dependency
* use Update to remove finalizer
* changing status and finalizer during create
* do not call Delete() twice

Author: FxKu

docs/reference/operator_parameters.md

@@ -328,6 +328,14 @@ configuration they are grouped under the `kubernetes` key.
   drained if the node_readiness_label is not used. This option if set to `false`
   will not add the `spilo-role=master` selector to the PDB.
 
+* **enable_finalizers**
+  By default, a deletion of the Postgresql resource will trigger a cleanup of
+  all child resources. However, if the database cluster is in a broken state
+  (e.g. failed initialization) and the operator cannot fully sync it, there can
+  be leftovers from a DELETE event. By enabling finalizers the Operator will
+  ensure all managed resources are deleted prior to the Postgresql resource.
+  The default is `false`.
+
 * **enable_pod_disruption_budget**
   PDB is enabled by default to protect the cluster from voluntarily disruptions
   and hence unwanted DB downtime. However, on some cloud providers it could be

pkg/apis/acid.zalan.do/v1/const.go

@@ -1,6 +1,6 @@
 package v1
 
-// 	ClusterStatusUnknown etc : status of a Postgres cluster known to the operator
+// ClusterStatusUnknown etc : status of a Postgres cluster known to the operator
 const (
 	ClusterStatusUnknown      = ""
 	ClusterStatusCreating     = "Creating"

pkg/cluster/__debug_bin4170763078

@@ -13,7 +13,6 @@ import (
 	"sync"
 	"time"
 
-	jsonpatch "github.com/evanphx/json-patch"
 	"github.com/sirupsen/logrus"
 	acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
 
@@ -248,9 +247,10 @@ func (c *Cluster) Create() (err error) {
 	defer c.mu.Unlock()
 
 	var (
-		service *v1.Service
-		ep      *v1.Endpoints
-		ss      *appsv1.StatefulSet
+		pgCreateStatus *acidv1.Postgresql
+		service        *v1.Service
+		ep             *v1.Endpoints
+		ss             *appsv1.StatefulSet
 	)
 
 	defer func() {
@@ -261,11 +261,15 @@ func (c *Cluster) Create() (err error) {
 		}
 	}()
 
-	c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusCreating)
+	pgCreateStatus, err = c.KubeClient.SetPostgresCRDStatus(c.clusterName(), acidv1.ClusterStatusCreating)
+	if err != nil {
+		return fmt.Errorf("could not set cluster status: %v", err)
+	}
+	c.setSpec(pgCreateStatus)
+
 	if c.OpConfig.EnableFinalizers != nil && *c.OpConfig.EnableFinalizers {
-		c.logger.Info("Adding finalizer.")
-		if err = c.AddFinalizer(); err != nil {
-			return fmt.Errorf("could not add Finalizer: %v", err)
+		if err = c.addFinalizer(); err != nil {
+			return fmt.Errorf("could not add finalizer: %v", err)
 		}
 	}
 	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Create", "Started creation of new cluster resources")
@@ -771,60 +775,45 @@ func (c *Cluster) compareServices(old, new *v1.Service) (bool, string) {
 	return true, ""
 }
 
-// AddFinalizer patches the postgresql CR to add our finalizer.
-func (c *Cluster) AddFinalizer() error {
-	if c.HasFinalizer() {
-		c.logger.Debugf("Finalizer %s already exists.", finalizerName)
+// addFinalizer patches the postgresql CR to add finalizer
+func (c *Cluster) addFinalizer() error {
+	if c.hasFinalizer() {
 		return nil
 	}
 
-	currentSpec := c.DeepCopy()
-	newSpec := c.DeepCopy()
-	newSpec.ObjectMeta.SetFinalizers(append(newSpec.ObjectMeta.Finalizers, finalizerName))
-	patchBytes, err := getPatchBytes(currentSpec, newSpec)
-	if err != nil {
-		return fmt.Errorf("Unable to produce patch to add finalizer: %v", err)
-	}
-
-	updatedSpec, err := c.KubeClient.AcidV1ClientSet.AcidV1().Postgresqls(c.clusterNamespace()).Patch(
-		context.TODO(), c.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
+	c.logger.Infof("adding finalizer %s", finalizerName)
+	finalizers := append(c.ObjectMeta.Finalizers, finalizerName)
+	newSpec, err := c.KubeClient.SetFinalizer(c.clusterName(), c.DeepCopy(), finalizers)
 	if err != nil {
-		return fmt.Errorf("Could not add finalizer: %v", err)
+		return fmt.Errorf("error adding finalizer: %v", err)
 	}
 
-	// update the spec, maintaining the new resourceVersion.
-	c.setSpec(updatedSpec)
+	// update the spec, maintaining the new resourceVersion
+	c.setSpec(newSpec)
 	return nil
 }
 
-// RemoveFinalizer patches postgresql CR to remove finalizer.
-func (c *Cluster) RemoveFinalizer() error {
-	if !c.HasFinalizer() {
-		c.logger.Debugf("No finalizer %s exists to remove.", finalizerName)
+// removeFinalizer patches postgresql CR to remove finalizer
+func (c *Cluster) removeFinalizer() error {
+	if !c.hasFinalizer() {
 		return nil
 	}
-	currentSpec := c.DeepCopy()
-	newSpec := c.DeepCopy()
-	newSpec.ObjectMeta.SetFinalizers(removeString(newSpec.ObjectMeta.Finalizers, finalizerName))
-	patchBytes, err := getPatchBytes(currentSpec, newSpec)
-	if err != nil {
-		return fmt.Errorf("Unable to produce patch to remove finalizer: %v", err)
-	}
 
-	updatedSpec, err := c.KubeClient.AcidV1ClientSet.AcidV1().Postgresqls(c.clusterNamespace()).Patch(
-		context.TODO(), c.Name, types.MergePatchType, patchBytes, metav1.PatchOptions{})
+	c.logger.Infof("removing finalizer %s", finalizerName)
+	finalizers := util.RemoveString(c.ObjectMeta.Finalizers, finalizerName)
+	newSpec, err := c.KubeClient.SetFinalizer(c.clusterName(), c.DeepCopy(), finalizers)
 	if err != nil {
-		return fmt.Errorf("Could not remove finalizer: %v", err)
+		return fmt.Errorf("error removing finalizer: %v", err)
 	}
 
 	// update the spec, maintaining the new resourceVersion.
-	c.setSpec(updatedSpec)
+	c.setSpec(newSpec)
 
 	return nil
 }
 
-// HasFinalizer checks if our finalizer is currently set or not
-func (c *Cluster) HasFinalizer() bool {
+// hasFinalizer checks if finalizer is currently set or not
+func (c *Cluster) hasFinalizer() bool {
 	for _, finalizer := range c.ObjectMeta.Finalizers {
 		if finalizer == finalizerName {
 			return true
@@ -833,36 +822,6 @@ func (c *Cluster) HasFinalizer() bool {
 	return false
 }
 
-// Iterate through slice and remove certain string, then return cleaned slice
-func removeString(slice []string, s string) (result []string) {
-	for _, item := range slice {
-		if item == s {
-			continue
-		}
-		result = append(result, item)
-	}
-	return result
-}
-
-// getPatchBytes will produce a JSONpatch between the two parameters of type acidv1.Postgresql
-func getPatchBytes(oldSpec, newSpec *acidv1.Postgresql) ([]byte, error) {
-	oldData, err := json.Marshal(oldSpec)
-	if err != nil {
-		return nil, fmt.Errorf("failed to Marshal oldSpec for postgresql %s/%s: %v", oldSpec.Namespace, oldSpec.Name, err)
-	}
-
-	newData, err := json.Marshal(newSpec)
-	if err != nil {
-		return nil, fmt.Errorf("failed to Marshal newSpec for postgresql %s/%s: %v", newSpec.Namespace, newSpec.Name, err)
-	}
-
-	patchBytes, err := jsonpatch.CreateMergePatch(oldData, newData)
-	if err != nil {
-		return nil, fmt.Errorf("failed to CreateMergePatch for postgresl %s/%s: %v", oldSpec.Namespace, oldSpec.Name, err)
-	}
-	return patchBytes, nil
-}
-
 // Update changes Kubernetes objects according to the new specification. Unlike the sync case, the missing object
 // (i.e. service) is treated as an error
 // logical backup cron jobs are an exception: a user-initiated Update can enable a logical backup job
@@ -1106,40 +1065,41 @@ func syncResources(a, b *v1.ResourceRequirements) bool {
 // before the pods, it will be re-created by the current master pod and will remain, obstructing the
 // creation of the new cluster with the same name. Therefore, the endpoints should be deleted last.
 func (c *Cluster) Delete() error {
+	var anyErrors = false
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Delete", "Started deletion of cluster resources")
 
 	if err := c.deleteStreams(); err != nil {
+		anyErrors = true
 		c.logger.Warningf("could not delete event streams: %v", err)
-		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not delete event streams: %v", err)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not delete event streams: %v", err)
 	}
-	var anyErrors = false
 
 	// delete the backup job before the stateful set of the cluster to prevent connections to non-existing pods
 	// deleting the cron job also removes pods and batch jobs it created
 	if err := c.deleteLogicalBackupJob(); err != nil {
 		anyErrors = true
 		c.logger.Warningf("could not remove the logical backup k8s cron job; %v", err)
-		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not remove the logical backup k8s cron job; %v", err)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not remove the logical backup k8s cron job; %v", err)
 	}
 
 	if err := c.deleteStatefulSet(); err != nil {
 		anyErrors = true
 		c.logger.Warningf("could not delete statefulset: %v", err)
-		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not delete statefulset: %v", err)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not delete statefulset: %v", err)
 	}
 
 	if err := c.deleteSecrets(); err != nil {
 		anyErrors = true
 		c.logger.Warningf("could not delete secrets: %v", err)
-		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not delete secrets: %v", err)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not delete secrets: %v", err)
 	}
 
 	if err := c.deletePodDisruptionBudget(); err != nil {
 		anyErrors = true
 		c.logger.Warningf("could not delete pod disruption budget: %v", err)
-		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not delete pod disruption budget: %v", err)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not delete pod disruption budget: %v", err)
 	}
 
 	for _, role := range []PostgresRole{Master, Replica} {
@@ -1148,21 +1108,21 @@ func (c *Cluster) Delete() error {
 			if err := c.deleteEndpoint(role); err != nil {
 				anyErrors = true
 				c.logger.Warningf("could not delete %s endpoint: %v", role, err)
-				c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not delete %s endpoint: %v", role, err)
+				c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not delete %s endpoint: %v", role, err)
 			}
 		}
 
 		if err := c.deleteService(role); err != nil {
 			anyErrors = true
 			c.logger.Warningf("could not delete %s service: %v", role, err)
-			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not delete %s service: %v", role, err)
+			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not delete %s service: %v", role, err)
 		}
 	}
 
 	if err := c.deletePatroniClusterObjects(); err != nil {
 		anyErrors = true
 		c.logger.Warningf("could not remove leftover patroni objects; %v", err)
-		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not remove leftover patroni objects; %v", err)
+		c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not remove leftover patroni objects; %v", err)
 	}
 
 	// Delete connection pooler objects anyway, even if it's not mentioned in the
@@ -1172,20 +1132,19 @@ func (c *Cluster) Delete() error {
 		if err := c.deleteConnectionPooler(role); err != nil {
 			anyErrors = true
 			c.logger.Warningf("could not remove connection pooler: %v", err)
-			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "Could not remove connection pooler: %v", err)
+			c.eventRecorder.Eventf(c.GetReference(), v1.EventTypeWarning, "Delete", "could not remove connection pooler: %v", err)
 		}
 	}
 
 	// If we are done deleting our various resources we remove the finalizer to let K8S finally delete the Postgres CR
 	if anyErrors {
-		c.eventRecorder.Event(c.GetReference(), v1.EventTypeWarning, "Delete", "Some resources could be successfully deleted yet")
+		c.eventRecorder.Event(c.GetReference(), v1.EventTypeWarning, "Delete", "some resources could be successfully deleted yet")
 		return fmt.Errorf("some error(s) occured when deleting resources, NOT removing finalizer yet")
 	}
-	if err := c.RemoveFinalizer(); err != nil {
-		return fmt.Errorf("Done cleaning up, but error when trying to remove our finalizer: %v", err)
+	if err := c.removeFinalizer(); err != nil {
+		return fmt.Errorf("done cleaning up, but error when removing finalizer: %v", err)
 	}
 
-	c.logger.Info("Done cleaning up our resources, removed finalizer.")
 	return nil
 }
 

pkg/cluster/cluster.go

@@ -1,11 +1,13 @@
 package cluster
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -33,7 +35,10 @@ const (
 )
 
 var logger = logrus.New().WithField("test", "cluster")
-var eventRecorder = record.NewFakeRecorder(1)
+
+// eventRecorder needs buffer for TestCreate which emit events for
+// 1 cluster, primary endpoint, 2 services, the secrets, the statefulset and pods being ready
+var eventRecorder = record.NewFakeRecorder(7)
 
 var cl = New(
 	Config{
@@ -79,6 +84,79 @@ var cl = New(
 	eventRecorder,
 )
 
+func TestCreate(t *testing.T) {
+	clientSet := fake.NewSimpleClientset()
+	acidClientSet := fakeacidv1.NewSimpleClientset()
+	clusterName := "cluster-with-finalizer"
+	clusterNamespace := "test"
+
+	client := k8sutil.KubernetesClient{
+		DeploymentsGetter:            clientSet.AppsV1(),
+		EndpointsGetter:              clientSet.CoreV1(),
+		PersistentVolumeClaimsGetter: clientSet.CoreV1(),
+		PodDisruptionBudgetsGetter:   clientSet.PolicyV1(),
+		PodsGetter:                   clientSet.CoreV1(),
+		PostgresqlsGetter:            acidClientSet.AcidV1(),
+		ServicesGetter:               clientSet.CoreV1(),
+		SecretsGetter:                clientSet.CoreV1(),
+		StatefulSetsGetter:           clientSet.AppsV1(),
+	}
+
+	pg := acidv1.Postgresql{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      clusterName,
+			Namespace: clusterNamespace,
+		},
+		Spec: acidv1.PostgresSpec{
+			Volume: acidv1.Volume{
+				Size: "1Gi",
+			},
+		},
+	}
+
+	pod := v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("%s-0", clusterName),
+			Namespace: clusterNamespace,
+			Labels: map[string]string{
+				"application":  "spilo",
+				"cluster-name": clusterName,
+				"spilo-role":   "master",
+			},
+		},
+	}
+
+	// manually create resources which must be found by further API calls and are not created by cluster.Create()
+	client.Postgresqls(clusterNamespace).Create(context.TODO(), &pg, metav1.CreateOptions{})
+	client.Pods(clusterNamespace).Create(context.TODO(), &pod, metav1.CreateOptions{})
+
+	var cluster = New(
+		Config{
+			OpConfig: config.Config{
+				PodManagementPolicy: "ordered_ready",
+				Resources: config.Resources{
+					ClusterLabels:         map[string]string{"application": "spilo"},
+					ClusterNameLabel:      "cluster-name",
+					DefaultCPURequest:     "300m",
+					DefaultCPULimit:       "300m",
+					DefaultMemoryRequest:  "300Mi",
+					DefaultMemoryLimit:    "300Mi",
+					PodRoleLabel:          "spilo-role",
+					ResourceCheckInterval: time.Duration(3),
+					ResourceCheckTimeout:  time.Duration(10),
+				},
+				EnableFinalizers: util.True(),
+			},
+		}, client, pg, logger, eventRecorder)
+
+	err := cluster.Create()
+	assert.NoError(t, err)
+
+	if !cluster.hasFinalizer() {
+		t.Errorf("%s - expected finalizer not found on cluster", t.Name())
+	}
+}
+
 func TestStatefulSetAnnotations(t *testing.T) {
 	spec := acidv1.PostgresSpec{
 		TeamID: "myapp", NumberOfInstances: 1,