enable changing encryption without creating new resource for redshift cluster

aws/resource_aws_redshift_cluster.go

@@ -153,8 +153,7 @@ func resourceAwsRedshiftCluster() *schema.Resource {
 			"encrypted": {
 				Type:     schema.TypeBool,
 				Optional: true,
-				Computed: true,
-				ForceNew: true,
+				Default:  false,
 			},
 
 			"enhanced_vpc_routing": {
@@ -167,7 +166,6 @@ func resourceAwsRedshiftCluster() *schema.Resource {
 				Type:         schema.TypeString,
 				Optional:     true,
 				Computed:     true,
-				ForceNew:     true,
 				ValidateFunc: validateArn,
 			},
 
@@ -713,6 +711,16 @@ func resourceAwsRedshiftClusterUpdate(d *schema.ResourceData, meta interface{})
 		requestUpdate = true
 	}
 
+	if d.HasChange("encrypted") {
+		req.Encrypted = aws.Bool(d.Get("encrypted").(bool))
+		requestUpdate = true
+	}
+
+	if d.Get("encrypted").(bool) && d.HasChange("kms_key_id") {
+		req.KmsKeyId = aws.String(d.Get("kms_key_id").(string))
+		requestUpdate = true
+	}
+
 	if requestUpdate {
 		log.Printf("[INFO] Modifying Redshift Cluster: %s", d.Id())
 		log.Printf("[DEBUG] Redshift Cluster Modify options: %s", req)

aws/resource_aws_redshift_cluster_test.go

@@ -1,6 +1,7 @@
 package aws
 
 import (
+	"errors"
 	"fmt"
 	"log"
 	"regexp"
@@ -560,6 +561,69 @@ func TestAccAWSRedshiftCluster_changeAvailabilityZone(t *testing.T) {
 	})
 }
 
+func TestAccAWSRedshiftCluster_changeEncryption1(t *testing.T) {
+	var cluster1, cluster2 redshift.Cluster
+
+	ri := acctest.RandInt()
+	preConfig := testAccAWSRedshiftClusterConfig_basic(ri)
+	postConfig := testAccAWSRedshiftClusterConfig_encrypted(ri)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSRedshiftClusterDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: preConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSRedshiftClusterExists("aws_redshift_cluster.default", &cluster1),
+					resource.TestCheckResourceAttr("aws_redshift_cluster.default", "encrypted", "false"),
+				),
+			},
+
+			{
+				Config: postConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSRedshiftClusterExists("aws_redshift_cluster.default", &cluster2),
+					testAccCheckAWSRedshiftClusterNotRecreated(&cluster1, &cluster2),
+					resource.TestCheckResourceAttr("aws_redshift_cluster.default", "encrypted", "true"),
+				),
+			},
+		},
+	})
+}
+
+func TestAccAWSRedshiftCluster_changeEncryption2(t *testing.T) {
+	var cluster1, cluster2 redshift.Cluster
+
+	ri := acctest.RandInt()
+	preConfig := testAccAWSRedshiftClusterConfig_encrypted(ri)
+	postConfig := testAccAWSRedshiftClusterConfig_unencrypted(ri)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSRedshiftClusterDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: preConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSRedshiftClusterExists("aws_redshift_cluster.default", &cluster1),
+					resource.TestCheckResourceAttr("aws_redshift_cluster.default", "encrypted", "true"),
+				),
+			},
+			{
+				Config: postConfig,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSRedshiftClusterExists("aws_redshift_cluster.default", &cluster2),
+					testAccCheckAWSRedshiftClusterNotRecreated(&cluster1, &cluster2),
+					resource.TestCheckResourceAttr("aws_redshift_cluster.default", "encrypted", "false"),
+				),
+			},
+		},
+	})
+}
+
 func testAccCheckAWSRedshiftClusterDestroy(s *terraform.State) error {
 	for _, rs := range s.RootModule().Resources {
 		if rs.Type != "aws_redshift_cluster" {
@@ -834,6 +898,21 @@ func TestResourceAWSRedshiftClusterMasterPasswordValidation(t *testing.T) {
 	}
 }
 
+func testAccCheckAWSRedshiftClusterNotRecreated(i, j *redshift.Cluster) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		// In lieu of some other uniquely identifying attribute from the API that always changes
+		// when a cluster is destroyed and recreated with the same identifier, we use the SSH key
+		// as it will get regenerated when a cluster is destroyed.
+		// Certain update operations (e.g KMS encrypting a cluster) will change ClusterCreateTime.
+		// Clusters with the same identifier can/will have an overlapping Endpoint.Address.
+		if aws.StringValue(i.ClusterPublicKey) != aws.StringValue(j.ClusterPublicKey) {
+			return errors.New("Redshift Cluster was recreated")
+		}
+
+		return nil
+	}
+}
+
 func testAccAWSRedshiftClusterConfig_updateNodeCount(rInt int) string {
 	return fmt.Sprintf(`
 resource "aws_redshift_cluster" "default" {
@@ -883,6 +962,82 @@ resource "aws_redshift_cluster" "default" {
 }`, rInt)
 }
 
+func testAccAWSRedshiftClusterConfig_encrypted(rInt int) string {
+	return fmt.Sprintf(`
+resource "aws_kms_key" "foo" {
+	description = "Terraform acc test %d"
+	policy = <<POLICY
+	{
+	"Version": "2012-10-17",
+	"Id": "kms-tf-1",
+	"Statement": [
+		{
+		"Sid": "Enable IAM User Permissions",
+		"Effect": "Allow",
+		"Principal": {
+			"AWS": "*"
+		},
+		"Action": "kms:*",
+		"Resource": "*"
+		}
+	]
+	}
+	POLICY
+	}
+
+resource "aws_redshift_cluster" "default" {
+  cluster_identifier = "tf-redshift-cluster-%d"
+  availability_zone = "us-west-2a"
+  database_name = "mydb"
+  master_username = "foo_test"
+  master_password = "Mustbe8characters"
+  node_type = "dc1.large"
+  automated_snapshot_retention_period = 0
+  allow_version_upgrade = false
+  skip_final_snapshot = true
+  encrypted = true
+  kms_key_id = "${aws_kms_key.foo.arn}"
+}`, rInt, rInt)
+}
+
+func testAccAWSRedshiftClusterConfig_unencrypted(rInt int) string {
+	// This is used along with the terraform config created testAccAWSRedshiftClusterConfig_encrypted, to test removal of encryption.
+	//Removing the kms key here causes the key to be deleted before the redshift cluster is unencrypted, resulting in an unstable cluster. This is to be kept for the time-being unti we find a better way to handle this.
+	return fmt.Sprintf(`
+resource "aws_kms_key" "foo" {
+	description = "Terraform acc test %d"
+	policy = <<POLICY
+	{
+	"Version": "2012-10-17",
+	"Id": "kms-tf-1",
+	"Statement": [
+		{
+		"Sid": "Enable IAM User Permissions",
+		"Effect": "Allow",
+		"Principal": {
+			"AWS": "*"
+		},
+		"Action": "kms:*",
+		"Resource": "*"
+		}
+	]
+	}
+	POLICY
+	}
+
+resource "aws_redshift_cluster" "default" {
+  cluster_identifier = "tf-redshift-cluster-%d"
+  availability_zone = "us-west-2a"
+  database_name = "mydb"
+  master_username = "foo_test"
+  master_password = "Mustbe8characters"
+  node_type = "dc1.large"
+  automated_snapshot_retention_period = 0
+  allow_version_upgrade = false
+  skip_final_snapshot = true
+}`, rInt, rInt)
+}
+
 func testAccAWSRedshiftClusterConfigWithFinalSnapshot(rInt int) string {
 	return fmt.Sprintf(`
 resource "aws_redshift_cluster" "default" {