Merge pull request #2739 from bflad/sqs-queue-policy-eventual-consist…

…ency

resource/aws_sqs_queue_policy: Prevent missing policy error on read

aws/resource_aws_sqs_queue_policy.go

@@ -3,10 +3,13 @@ package aws
 import (
 	"fmt"
 	"log"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sqs"
+	"github.com/hashicorp/terraform/helper/resource"
 	"github.com/hashicorp/terraform/helper/schema"
+	"github.com/jen20/awspolicyequivalence"
 )
 
 func resourceAwsSqsQueuePolicy() *schema.Resource {
@@ -40,18 +43,54 @@ func resourceAwsSqsQueuePolicy() *schema.Resource {
 
 func resourceAwsSqsQueuePolicyUpsert(d *schema.ResourceData, meta interface{}) error {
 	conn := meta.(*AWSClient).sqsconn
+	policy := d.Get("policy").(string)
 	url := d.Get("queue_url").(string)
 
-	_, err := conn.SetQueueAttributes(&sqs.SetQueueAttributesInput{
+	sqaInput := &sqs.SetQueueAttributesInput{
 		QueueUrl: aws.String(url),
 		Attributes: aws.StringMap(map[string]string{
-			"Policy": d.Get("policy").(string),
+			sqs.QueueAttributeNamePolicy: policy,
 		}),
-	})
+	}
+	log.Printf("[DEBUG] Updating SQS attributes: %s", sqaInput)
+	_, err := conn.SetQueueAttributes(sqaInput)
 	if err != nil {
 		return fmt.Errorf("Error updating SQS attributes: %s", err)
 	}
 
+	// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_SetQueueAttributes.html
+	// When you change a queue's attributes, the change can take up to 60 seconds
+	// for most of the attributes to propagate throughout the Amazon SQS system.
+	gqaInput := &sqs.GetQueueAttributesInput{
+		QueueUrl:       aws.String(url),
+		AttributeNames: []*string{aws.String(sqs.QueueAttributeNamePolicy)},
+	}
+	notUpdatedError := fmt.Errorf("SQS attribute %s not updated", sqs.QueueAttributeNamePolicy)
+	err = resource.Retry(1*time.Minute, func() *resource.RetryError {
+		log.Printf("[DEBUG] Reading SQS attributes: %s", gqaInput)
+		out, err := conn.GetQueueAttributes(gqaInput)
+		if err != nil {
+			return resource.NonRetryableError(err)
+		}
+		queuePolicy, ok := out.Attributes[sqs.QueueAttributeNamePolicy]
+		if !ok {
+			log.Printf("[DEBUG] SQS attribute %s not found - retrying", sqs.QueueAttributeNamePolicy)
+			return resource.RetryableError(notUpdatedError)
+		}
+		equivalent, err := awspolicy.PoliciesAreEquivalent(*queuePolicy, policy)
+		if err != nil {
+			return resource.NonRetryableError(err)
+		}
+		if !equivalent {
+			log.Printf("[DEBUG] SQS attribute %s not updated - retrying", sqs.QueueAttributeNamePolicy)
+			return resource.RetryableError(notUpdatedError)
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
 	d.SetId(url)
 
 	return resourceAwsSqsQueuePolicyRead(d, meta)
@@ -62,7 +101,7 @@ func resourceAwsSqsQueuePolicyRead(d *schema.ResourceData, meta interface{}) err
 
 	out, err := conn.GetQueueAttributes(&sqs.GetQueueAttributesInput{
 		QueueUrl:       aws.String(d.Id()),
-		AttributeNames: []*string{aws.String("Policy")},
+		AttributeNames: []*string{aws.String(sqs.QueueAttributeNamePolicy)},
 	})
 	if err != nil {
 		if isAWSErr(err, "AWS.SimpleQueueService.NonExistentQueue", "") {
@@ -76,12 +115,13 @@ func resourceAwsSqsQueuePolicyRead(d *schema.ResourceData, meta interface{}) err
 		return fmt.Errorf("Received empty response for SQS queue %s", d.Id())
 	}
 
-	policy, ok := out.Attributes["Policy"]
-	if !ok {
-		return fmt.Errorf("SQS Queue policy not found for %s", d.Id())
+	policy, ok := out.Attributes[sqs.QueueAttributeNamePolicy]
+	if ok {
+		d.Set("policy", policy)
+	} else {
+		d.Set("policy", "")
 	}
 
-	d.Set("policy", policy)
 	d.Set("queue_url", d.Id())
 
 	return nil
@@ -94,7 +134,7 @@ func resourceAwsSqsQueuePolicyDelete(d *schema.ResourceData, meta interface{}) e
 	_, err := conn.SetQueueAttributes(&sqs.SetQueueAttributesInput{
 		QueueUrl: aws.String(d.Id()),
 		Attributes: aws.StringMap(map[string]string{
-			"Policy": "",
+			sqs.QueueAttributeNamePolicy: "",
 		}),
 	})
 	if err != nil {