Dedicated zookeeper support (#45)

* Added support for (soon to be released) dedicated zookeeper feature for Instaclustr managed Kafka clusters

* Forgot to remove commenting

* Addressed Arjun's comments

* bump version

* bumped the Makefile version to 1.5.0

Co-authored-by: Hendra Gunadi <[email protected]>

Makefile

@@ -1,5 +1,5 @@
 BIN_NAME="terraform-provider-instaclustr"
-VERSION=v1.4.0
+VERSION=v1.5.0
 
 .PHONY: install clean all build test testacc
 

README.md

@@ -163,6 +163,9 @@ ssl_enabled_protocols, ssl_truststore_password, ssl_protocol, security_protocol,
 truststore|Base64 encoded version of the TLS trust store (in JKS format) used to connect to your Kafka Cluster. Only required if connecting to a Non-Instaclustr managed Kafka Cluster with TLS enabled|Kafka Connect
 master_nodes|The number of Master nodes in a generated Redis Cluster.|Redis|Required (Integers)
 replica_nodes|The number of Replica nodes in a generated Redis Cluster.|Redis|Required (Integers)
+dedicated_zookeeper|Indicate whether this Kafka cluster should allocate dedicated Zookeeper nodes|Kafka|false
+zookeeper_node_size|If `dedicated_zookeeper` is true, then it is the node size for the dedicated Zookeeper nodes. Have a look [here](https://www.instaclustr.com/support/api-integrations/api-reference/provisioning-api/#section-create-cluster) (Kafka bundle options table) for node size options. |Kafka
+zookeeper_node_count|If `dedicated_zookeeper` is true, then it indicates how many nodes are allocated to be Zookeeper nodes|Kafka
 
 ### Resource:  `instaclustr_firewall_rule`                             
 A resource for managing cluster firewall rules on Instaclustr Managed Platform. A firewall rule allows access to your Instaclustr cluster.

examples/main.tf

@@ -109,9 +109,12 @@ resource "instaclustr_cluster" "example_kafka" {
 
   bundle {
     bundle = "KAFKA"
-    version = "2.3.1"
+    version = "2.5.1"
     options = {
       auth_n_authz = true
+      dedicated_zookeeper = true
+      zookeeper_node_size = "zk-production-m5.large-60"
+      zookeeper_node_count = 3
     }
   }
 

instaclustr/resource_cluster.go

@@ -254,6 +254,18 @@ func resourceCluster() *schema.Resource {
 										Type:     schema.TypeString,
 										Optional: true,
 									},
+									"dedicated_zookeeper": {
+										Type:     schema.TypeBool,
+										Optional: true,
+									},
+									"zookeeper_node_size": {
+										Type:     schema.TypeString,
+										Optional: true,
+									},
+									"zookeeper_node_count": {
+										Type:     schema.TypeInt,
+										Optional: true,
+									},
 									"master_nodes": {
 										Type:     schema.TypeInt,
 										Optional: true,
@@ -371,7 +383,17 @@ func resourceClusterRead(d *schema.ResourceData, meta interface{}) error {
 	d.Set("cluster_id", cluster.ID)
 	d.Set("cluster_name", cluster.ClusterName)
 
-	nodeSize := cluster.DataCentres[0].Nodes[0].Size
+	nodeSize := ""
+	/* 
+	*  Ideally, we would like this information to be coming directly from the API cluster status.
+	*  Hence, this is a slightly hacky way of ignoring zookeeper node sizes (Kafka bundle specific).
+	*/
+	for _, node := range(cluster.DataCentres[0].Nodes) {
+		nodeSize = node.Size
+		if (!strings.HasPrefix(nodeSize, "zk-")) {
+			break
+		}
+	}
 	if len(cluster.DataCentres[0].ResizeTargetNodeSize) > 0 {
 		nodeSize = cluster.DataCentres[0].ResizeTargetNodeSize
 	}

instaclustr/structs.go

@@ -47,6 +47,9 @@ type BundleOptions struct {
 	Truststore                    string `json:"truststore,omitempty" mapstructure:"truststore"`
 	RedisMasterNodes              string `json:"masterNodes,omitempty" mapstructure:"master_nodes"`
 	RedisReplicaNodes             string `json:"replicaNodes,omitempty" mapstructure:"replica_nodes"`
+	DedicatedZookeeper            string `json:"dedicatedZookeeper,omitempty" mapstructure:"dedicated_zookeeper"`
+	ZookeeperNodeSize             string `json:"zookeeperNodeSize,omitempty" mapstructure:"zookeeper_node_size"`
+	ZookeeperNodeCount            string `json:"zookeeperNodeCount,omitempty" mapstructure:"zookeeper_node_count"`
 }
 
 type ClusterProvider struct {

test/data/kafka_user_create_cluster.tf

@@ -0,0 +1,31 @@
+// This is part of testing "kafka user" suite, 1 of 3
+provider "instaclustr" {
+  username = "%s"
+  api_key = "%s"
+  api_hostname = "%s"
+}
+
+resource "instaclustr_cluster" "kafka_cluster" {
+  cluster_name = "example_kafka_tf_test"
+  node_size = "t3.small-20-gp2"
+  data_centre = "US_WEST_2"
+  sla_tier = "NON_PRODUCTION"
+  cluster_network = "192.168.0.0/18"
+  cluster_provider = {
+    name = "AWS_VPC"
+  }
+  rack_allocation = {
+    number_of_racks = 3
+    nodes_per_rack = 1
+  }
+
+  bundle {
+    bundle = "KAFKA"
+    version = "2.5.1"
+    options = {
+      dedicated_zookeeper = true
+      zookeeper_node_size = "%s"
+      zookeeper_node_count = 3
+    }
+  }
+}

test/data/kafka_user_create_user.tf

@@ -0,0 +1,38 @@
+// This is part of testing "kafka user" suite, 2 of 3
+provider "instaclustr" {
+  username = "%s"
+  api_key = "%s"
+  api_hostname = "%s"
+}
+
+resource "instaclustr_cluster" "kafka_cluster" {
+  cluster_name = "example_kafka_tf_test"
+  node_size = "t3.small-20-gp2"
+  data_centre = "US_WEST_2"
+  sla_tier = "NON_PRODUCTION"
+  cluster_network = "192.168.0.0/18"
+  cluster_provider = {
+    name = "AWS_VPC"
+  }
+  rack_allocation = {
+    number_of_racks = 3
+    nodes_per_rack = 1
+  }
+
+  bundle {
+    bundle = "KAFKA"
+    version = "2.5.1"
+    options = {
+      dedicated_zookeeper = true
+      zookeeper_node_size = "%s"
+      zookeeper_node_count = 3
+    }
+  }
+}
+
+resource "instaclustr_kafka_user" "kafka_user_charlie" {
+  cluster_id = "${instaclustr_cluster.kafka_cluster.cluster_id}"
+  username = "%s"
+  password = "%s"
+  initial_permissions = "none"
+}

test/data/kafka_user.tf → test/data/kafka_user_user_list.tf

@@ -1,3 +1,4 @@
+// This is part of testing "kafka user" suite, 3 of 3
 provider "instaclustr" {
   username = "%s"
   api_key = "%s"
@@ -20,19 +21,22 @@ resource "instaclustr_cluster" "kafka_cluster" {
 
   bundle {
     bundle = "KAFKA"
-    version = "2.3.1"
+    version = "2.5.1"
+    options = {
+      dedicated_zookeeper = true
+      zookeeper_node_size = "%s"
+      zookeeper_node_count = 3
+    }
   }
 }
-/*
+
 resource "instaclustr_kafka_user" "kafka_user_charlie" {
   cluster_id = "${instaclustr_cluster.kafka_cluster.cluster_id}"
-  username = "@@KAFKA_USERNAME@@"
-  password = "@@KAFKA_USER_PASSWORD@@"
+  username = "%s"
+  password = "%s"
   initial_permissions = "none"
 }
-*/
-/*
+
 data "instaclustr_kafka_user_list" "kafka_user_list" {
   cluster_id = "${instaclustr_cluster.kafka_cluster.cluster_id}"
 }
-*/

test/resource_kafka_user_test.go

@@ -6,7 +6,6 @@ import (
 	"os"
 	"testing"
 	"time"
-	"strings"
 	"strconv"
 
 	"github.com/hashicorp/terraform/helper/resource"
@@ -19,23 +18,22 @@ func TestKafkaUserResource(t *testing.T) {
 		"instaclustr": instaclustr.Provider(),
 	}
 
-	tfFile, _ := ioutil.ReadFile("data/kafka_user.tf")
-	oriConfig := string(tfFile)
+	configBytes1, _ := ioutil.ReadFile("data/kafka_user_create_cluster.tf")
+	configBytes2, _ := ioutil.ReadFile("data/kafka_user_create_user.tf")
+	configBytes3, _ := ioutil.ReadFile("data/kafka_user_user_list.tf")
 	username := os.Getenv("IC_USERNAME")
 	apiKey := os.Getenv("IC_API_KEY")
 	hostname := getOptionalEnv("IC_API_URL", instaclustr.DefaultApiHostname)
 
-	kafka_username := "charlie"
+	kafkaUsername := "charlie"
 	oldPassword := "charlie123!"
 	newPassword := "charlie123standard!"
-	const KAFKA_USERNAME_PATTERN = "@@KAFKA_USERNAME@@"
-	const KAFKA_USER_PASSWORD_PATTERN = "@@KAFKA_USER_PASSWORD@@"	
+	zookeeperNodeSize := "zk-developer-t3.small-20"
 
-	createClusterConfig := fmt.Sprintf(oriConfig, username, apiKey, hostname)
-	createKafkaUserConfigRaw := strings.Replace(strings.Replace(createClusterConfig, "/*", "", 1), "*/", "", 1)
-	createKafkaUserConfig := strings.ReplaceAll(strings.ReplaceAll(createKafkaUserConfigRaw, KAFKA_USERNAME_PATTERN, kafka_username), KAFKA_USER_PASSWORD_PATTERN, oldPassword) 
-	createKafkaUserListConfig := strings.Replace(strings.Replace(createKafkaUserConfig, "/*", "", 1), "*/", "", 1)
-	updateKafkaUserConfig := strings.Replace(createKafkaUserConfig, oldPassword, newPassword, 1)
+	createClusterConfig := fmt.Sprintf(string(configBytes1), username, apiKey, hostname, zookeeperNodeSize)
+	createKafkaUserConfig := fmt.Sprintf(string(configBytes2), username, apiKey, hostname, zookeeperNodeSize, kafkaUsername, oldPassword)
+	createKafkaUserListConfig := fmt.Sprintf(string(configBytes3), username, apiKey, hostname, zookeeperNodeSize, kafkaUsername, oldPassword)
+	updateKafkaUserConfig := fmt.Sprintf(string(configBytes3), username, apiKey, hostname, zookeeperNodeSize, kafkaUsername, newPassword)
 
 	resource.Test(t, resource.TestCase{
 		Providers:    testProviders,
@@ -71,7 +69,7 @@ func TestKafkaUserResource(t *testing.T) {
 			// i.e., we need to destroy the kafka user resources first.
 			{
 				Config: createClusterConfig,
-				Check: checkKafkaUserDeleted(kafka_username, hostname, username, apiKey),
+				Check: checkKafkaUserDeleted(kafkaUsername, hostname, username, apiKey),
 			},
 		},
 	})