Extend Operator to PSQL-HA

Makefile

@@ -3,7 +3,7 @@
 # To re-generate a bundle for another specific version without changing the standard setup, you can:
 # - use the VERSION as arg of the bundle target (e.g make bundle VERSION=0.0.8)
 # - use environment variables to overwrite this value (e.g export VERSION=0.0.8)
-VERSION ?= 0.5.0
+VERSION ?= 0.5.1
 
 # CHANNELS define the bundle channels used in the bundle.
 # Add a new line here if you would like to change its default config. (E.g CHANNELS = "candidate,fast,stable")
@@ -299,13 +299,22 @@ run-automation-cloning:
 	fi; \
 	go test $(DEFAULT_CLONING_ROOT)$$folders -v -timeout 90m
 
-.PHONY: run-automation-provisioning
+.PHONY: run-automation-provisioning-si
 DEFAULT_PROVISIONING_ROOT := ./automation/tests/provisioning/
-PROVISIONING_FOLDERS := ...
-run-automation-provisioning:
+# change this list to every folder but the pg-ha_test one
+SI_PROVISIONING_FOLDERS := ...
+run-automation-provisioning-si:
 	@read -p "Enter the test directories with spacing to run (mongo-si_test mssql-si_test mysql-si_test pg-si_test). Else all directories will be run: " folders; \
 	if [ -z "$$folders" ]; then \
-		folders="$(PROVISIONING_FOLDERS)"; \
+		folders="$(SI_PROVISIONING_FOLDERS)"; \
 	fi; \
 	go test $(DEFAULT_PROVISIONING_ROOT)$$folders -v -timeout 90m
 
+.PHONY: run-automation-provisioning-ha
+HA_PROVISIONING_FOLDERS := pg-ha_test
+run-automation-provisioning-ha:
+	@read -p "Enter the test directories with spacing to run (pg-ha_test). Else all directories will be run: " folders; \
+	if [ -z "$$folders" ]; then \
+		folders="$(HA_PROVISIONING_FOLDERS)"; \
+	fi; \
+	go test $(DEFAULT_PROVISIONING_ROOT)$$folders -v -timeout 90m

README.md

@@ -156,7 +156,9 @@ spec:
     size: 10
     timezone: "UTC"
     type: postgres
-
+    # isHighAvailability is an optional parameter. In case nothing is specified, it is set to false
+    isHighAvailability: false
+
     # You can specify any (or none) of these types of profiles: compute, software, network, dbParam
     # If not specified, the corresponding Out-of-Box (OOB) profile will be used wherever applicable
     # Name is case-sensitive. ID is the UUID of the profile. Profile should be in the "READY" state
@@ -214,6 +216,9 @@ spec:
     # Cluster id of the cluster where the Database has to be provisioned
     # Can be fetched from the GET /clusters endpoint
     clusterId: "Nutanix Cluster Id"
+    # isHighAvailability is an optional parameter. In case nothing is specified, it is set to false
+    isHighAvailability: false
+
     # You can specify any (or none) of these types of profiles: compute, software, network, dbParam
     # If not specified, the corresponding Out-of-Box (OOB) profile will be used wherever applicable
     # Name is case-sensitive. ID is the UUID of the profile. Profile should be in the "READY" state
@@ -285,6 +290,26 @@ additionalArguments:
   windows_domain_profile_id: <domain-profile-id>   # NO Default. Must specify vm_db_server_user.
   vm_db_server_user: <vm-db-server-use>            # NO Default. Must specify windows_domain_profile_id.
   vm_win_license_key: <licenseKey>                 # NO Default.
+
+# Postgres High Availability
+additionalArguments:
+  listener_port:           "1111"                  # Default: "5432"
+  failover_mode:           "Manual"                # Default: "Automatic"
+	proxy_read_port:         "1111"                  # Default: "5001"
+	listener_port:           "1111"                  # Default: "5432"
+	proxy_write_port:        "1111"                  # Default: "5000",
+	enable_synchronous_mode: "true"                  # Default: "true",
+	auto_tune_staging_drive: "false"                 # Default: true",
+	backup_policy:           "primary_only"          # Default: "primary_only"
+	provision_virtual_ip":   "true"                  # Default: "true"
+	deploy_haproxy:          "true"                  # Default: "true"
+	node_type:               "haproxy"               # Default: "database"
+	allocate_pg_hugepage:    "false"                 # Default: "true"
+	cluster_database:        "false"                 # Default: "true"
+	archive_wal_expire_days: "7"                     # Default: "-1"
+	enable_peer_auth:        "false"                 # Default: "true"
+	cluster_name:            "<cluster-name>"               
+	patroni_cluster_name:    "<patroni>"
 ```
 
 Cloning Additional Arguments: 

api/v1alpha1/database_types.go

@@ -107,6 +107,24 @@ type Instance struct {
 	// +optional
 	// Additional database engine specific arguments
 	AdditionalArguments map[string]string `json:"additionalArguments"`
+	// +optional
+	IsHighAvailability bool `json:"isHighAvailability"`
+	// +optional
+	Nodes []*Node `json:"nodes,omitempty"`
+}
+
+type Node struct {
+	// +optional
+	VmName     string         `json:"vmName"`
+	Properties NodeProperties `json:"properties"`
+}
+
+type NodeProperties struct {
+	NodeType string `json:"node_type"`
+	// +optional
+	Role string `json:"role"`
+	// +optional
+	FailoverMode string `json:"failover_mode"`
 }
 
 type Clone struct {

api/v1alpha1/database_webhook.go

@@ -97,13 +97,13 @@ func (r *Database) ValidateDelete() (admission.Warnings, error) {
 }
 
 /* Checks if configured additional arguments are valid or not and returns the corresponding additional arguments. If error is nil valid, else invalid */
-func additionalArgumentsValidationCheck(isClone bool, dbType string, specifiedAdditionalArguments map[string]string) error {
+func additionalArgumentsValidationCheck(isClone bool, dbType string, isHA bool, specifiedAdditionalArguments map[string]string) error {
 	// Empty additionalArguments is always valid
 	if specifiedAdditionalArguments == nil {
 		return nil
 	}
 
-	allowedAdditionalArguments, err := util.GetAllowedAdditionalArguments(isClone, dbType)
+	allowedAdditionalArguments, err := util.GetAllowedAdditionalArguments(isClone, dbType, isHA)
 
 	// Invalid type returns error
 	if err != nil {

api/v1alpha1/node_helpers.go

@@ -0,0 +1,59 @@
+package v1alpha1
+
+// validate the Node and NodeProperties passed are valid
+// e.g validate vmNames being unique, properties correctly defined, etc.
+// one day move to common/util
+
+import (
+	"fmt"
+)
+
+var (
+	typeOptions     = map[string]bool{"database": true, "haproxy": true}
+	roleOptions     = map[string]bool{"primary": true, "secondary": true}
+	failoverOptions = map[string]bool{"Automatic": true, "Manual": true}
+)
+
+func ValidateNodes(nodes []*Node, isHighAvailability bool) error {
+	if !isHighAvailability {
+		return nil
+	}
+
+	databaseNodeCount := 0
+	vmNames := make(map[string]bool) // for validating that vmnames are unique
+	for _, node := range nodes {
+		if node.Properties.NodeType == "database" {
+			databaseNodeCount++
+		}
+
+		if err := ValidateNodeProperties(node.Properties); err != nil {
+			return err
+		}
+
+		if _, ok := vmNames[node.VmName]; ok {
+			return fmt.Errorf("vmName %s is already specified", node.VmName)
+		}
+		vmNames[node.VmName] = true
+	}
+
+	if databaseNodeCount < 3 {
+		return fmt.Errorf("high Availability requires at least 3 nodes database nodes")
+	}
+	return nil
+}
+
+func ValidateNodeProperties(np NodeProperties) error {
+	if !typeOptions[np.NodeType] {
+		return fmt.Errorf("invalid NodeType in Node Properties: %s", np.NodeType)
+	}
+
+	if !roleOptions[np.Role] {
+		return fmt.Errorf("invalid Role in Node Properties: %s", np.Role)
+	}
+
+	if !failoverOptions[np.FailoverMode] {
+		return fmt.Errorf("invalid FailoverMode in Node Properties: %s", np.FailoverMode)
+	}
+
+	return nil
+}

api/v1alpha1/webhook_helpers.go

@@ -91,9 +91,12 @@ func (v *CloningWebhookHandler) validateCreate(spec *DatabaseSpec, errors *field
 		}
 	}
 
-	if err := additionalArgumentsValidationCheck(spec.IsClone, clone.Type, clone.AdditionalArguments); err != nil {
+	// HA is not supported when cloning
+	isHighAvailability := false
+	if err := additionalArgumentsValidationCheck(spec.IsClone, clone.Type, isHighAvailability, clone.AdditionalArguments); err != nil {
 		*errors = append(*errors, field.Invalid(clonePath.Child("additionalArguments"), clone.AdditionalArguments, err.Error()))
 	}
+
 	databaselog.Info("Exiting validateCreate for clone")
 }
 
@@ -230,10 +233,15 @@ func (v *ProvisioningWebhookHandler) validateCreate(spec *DatabaseSpec, errors *
 		))
 	}
 
-	if err := additionalArgumentsValidationCheck(spec.IsClone, instance.Type, instance.AdditionalArguments); err != nil {
+	if err := additionalArgumentsValidationCheck(spec.IsClone, instance.Type, instance.IsHighAvailability, instance.AdditionalArguments); err != nil {
 		*errors = append(*errors, field.Invalid(instancePath.Child("additionalArguments"), instance.AdditionalArguments, err.Error()))
 	}
 
+	// Validate nodes for HA
+	if err := ValidateNodes(instance.Nodes, instance.IsHighAvailability); err != nil {
+		*errors = append(*errors, field.Invalid(instancePath.Child("nodes"), instance.Nodes, err.Error()))
+	}
+
 	databaselog.Info("Exiting validateCreate for provisioning")
 }
 

api/v1alpha1/webhook_suite_test.go

@@ -60,6 +60,7 @@ const (
 	CREDENTIAL_SECRET = "database-secret"
 	TIMEZONE          = "UTC"
 	SIZE              = 10
+	HA                = false
 )
 
 func TestAPIs(t *testing.T) {
@@ -353,6 +354,139 @@ var _ = Describe("Webhook Tests", func() {
 				Expect(errMsg).To(ContainSubstring(fmt.Sprintf("additional arguments validation for type: %s failed!", common.DATABASE_TYPE_MSSQL)))
 			})
 		})
+
+		When("Postgres specified with IsHighAvailability", func() {
+			It("Should have zero nodes and IsHighAvailability set to true", func() {
+				db := createDefaultDatabase("db15")
+				db.Spec.Instance.AdditionalArguments = map[string]string{
+					"failover_mode":           "Automatic",
+					"proxy_read_port":         "5001",
+					"listener_port":           "5432",
+					"proxy_write_port":        "5000",
+					"enable_synchronous_mode": "true",
+					"auto_tune_staging_drive": "true",
+					"backup_policy":           "primary_only",
+					"provision_virtual_ip":    "true",
+					"deploy_haproxy":          "true",
+					"node_type":               "database",
+					"allocate_pg_hugepage":    "false",
+					"cluster_database":        "false",
+					"archive_wal_expire_days": "-1",
+					"enable_peer_auth":        "false",
+					"cluster_name":            "psqlcluster",
+					"patroni_cluster_name":    "patroni",
+				}
+				db.Spec.Instance.IsHighAvailability = true
+				db.Spec.Instance.Nodes = nil
+
+				err := k8sClient.Create(context.Background(), db)
+				Expect(err).To(HaveOccurred())
+			})
+
+			It("Should have 5 nodes and IsHighAvailability set to true", func() {
+				db := createDefaultDatabase("db16")
+				db.Spec.Instance.AdditionalArguments = map[string]string{
+					"failover_mode":           "Automatic",
+					"proxy_read_port":         "5001",
+					"listener_port":           "5432",
+					"proxy_write_port":        "5000",
+					"enable_synchronous_mode": "true",
+					"auto_tune_staging_drive": "true",
+					"backup_policy":           "primary_only",
+					"provision_virtual_ip":    "true",
+					"deploy_haproxy":          "true",
+					"node_type":               "database",
+					"allocate_pg_hugepage":    "false",
+					"cluster_database":        "false",
+					"archive_wal_expire_days": "-1",
+					"enable_peer_auth":        "false",
+					"cluster_name":            "psqlcluster",
+					"patroni_cluster_name":    "patroni",
+				}
+				primaryProp := createDefaultNodeProperties("database", "primary")
+				secondaryProp := createDefaultNodeProperties("database", "secondary")
+				proxyProp := createDefaultNodeProperties("haproxy", "secondary")
+				db.Spec.Instance.IsHighAvailability = true
+				db.Spec.Instance.Nodes = []*Node{
+					{
+						VmName:     "VM1",
+						Properties: *primaryProp,
+					},
+					{
+						VmName:     "VM2",
+						Properties: *secondaryProp,
+					},
+					{
+						VmName:     "VM3",
+						Properties: *secondaryProp,
+					},
+					{
+						VmName:     "VM4",
+						Properties: *proxyProp,
+					},
+					{
+						VmName:     "VM5",
+						Properties: *proxyProp,
+					},
+				}
+
+				err := k8sClient.Create(context.Background(), db)
+				Expect(err).ToNot(HaveOccurred())
+			})
+
+			It("Should throw error when given 2 nodes", func() {
+				db := createDefaultDatabase("db17")
+				db.Spec.Instance.AdditionalArguments = map[string]string{
+					"failover_mode":           "Automatic",
+					"proxy_read_port":         "5001",
+					"listener_port":           "5432",
+					"proxy_write_port":        "5000",
+					"enable_synchronous_mode": "true",
+					"auto_tune_staging_drive": "true",
+					"backup_policy":           "primary_only",
+					"provision_virtual_ip":    "true",
+					"deploy_haproxy":          "true",
+					"node_type":               "database",
+					"allocate_pg_hugepage":    "false",
+					"cluster_database":        "false",
+					"archive_wal_expire_days": "-1",
+					"enable_peer_auth":        "false",
+					"cluster_name":            "psqlcluster",
+					"patroni_cluster_name":    "patroni",
+				}
+				primaryProp := createDefaultNodeProperties("database", "primary")
+				secondaryProp := createDefaultNodeProperties("database", "secondary")
+				db.Spec.Instance.IsHighAvailability = true
+				db.Spec.Instance.Nodes = []*Node{
+					{
+						VmName:     "VM1",
+						Properties: *primaryProp,
+					},
+					{
+						VmName:     "VM2",
+						Properties: *secondaryProp,
+					},
+				}
+
+				err := k8sClient.Create(context.Background(), db)
+				Expect(err).To(HaveOccurred())
+			})
+
+			It("Should error out for invalid Postgres additionalArguments and IsHighAvailability set to true", func() {
+				db := createDefaultDatabase("db18")
+				db.Spec.Instance.AdditionalArguments = map[string]string{
+					"listener_port": "5432",
+					"invalid":       "invalid",
+				}
+				db.Spec.Instance.IsHighAvailability = true
+
+				err := k8sClient.Create(context.Background(), db)
+				Expect(err).To(HaveOccurred())
+				errMsg := err.(*errors.StatusError).ErrStatus.Message
+				Expect(errMsg).To(ContainSubstring(fmt.Sprintf("additional arguments validation for type: %s failed!", common.DATABASE_TYPE_POSTGRES)))
+			})
+		})
+
 	})
 
 	Context("Clone checks", func() {
@@ -595,6 +729,7 @@ var _ = Describe("Webhook Tests", func() {
 				Expect(errMsg).To(ContainSubstring(fmt.Sprintf("additional arguments validation for type: %s failed!", common.DATABASE_TYPE_MSSQL)))
 			})
 		})
+
 	})
 })
 
@@ -615,6 +750,7 @@ func createDefaultDatabase(metadataName string) *Database {
 				Type:                common.DATABASE_TYPE_POSTGRES,
 				Profiles:            &(Profiles{}),
 				AdditionalArguments: map[string]string{},
+				IsHighAvailability:  HA,
 			},
 		},
 	}
@@ -643,3 +779,11 @@ func createDefaultClone(metadataName string) *Database {
 		},
 	}
 }
+
+func createDefaultNodeProperties(node_type, role string) *NodeProperties {
+	return &NodeProperties{
+		NodeType:     node_type,
+		Role:         role,
+		FailoverMode: "Automatic",
+	}
+}