Skip to content

Commit

Permalink
resource/aws_dms_event_subscription: Finish initial implementation
Browse files Browse the repository at this point in the history 
Reference: #7170 (review)

Output from acceptance testing:

```
--- PASS: TestAccAWSDmsEventSubscription_basic (481.92s)
--- PASS: TestAccAWSDmsEventSubscription_disappears (500.31s)
--- PASS: TestAccAWSDmsEventSubscription_Enabled (584.73s)
--- PASS: TestAccAWSDmsEventSubscription_EventCategories (601.53s)
--- PASS: TestAccAWSDmsEventSubscription_Tags (540.87s)
```
  • Loading branch information
@bflad
bflad committed Apr 15, 2020
1 parent e9447a9 commit 92200ba
Show file tree
Hide file tree
Showing 4 changed files with 383 additions and 215 deletions.
131 changes: 81 additions & 50 deletions aws/resource_aws_dms_event_subscription.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,11 @@ import (
"time"

"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/arn"
dms "github.com/aws/aws-sdk-go/service/databasemigrationservice"
"github.com/hashicorp/terraform-plugin-sdk/helper/resource"
"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
"github.com/hashicorp/terraform-plugin-sdk/helper/validation"
"github.com/terraform-providers/terraform-provider-aws/aws/internal/keyvaluetags"
)

Expand All @@ -29,6 +31,10 @@ func resourceAwsDmsEventSubscription() *schema.Resource {
},

Schema: map[string]*schema.Schema{
"arn": {
Type: schema.TypeString,
Computed: true,
},
"enabled": {
Type: schema.TypeBool,
Optional: true,
Expand All @@ -41,13 +47,15 @@ func resourceAwsDmsEventSubscription() *schema.Resource {
Required: true,
},
"name": {
Type: schema.TypeString,
Required: true,
ForceNew: true,
Type: schema.TypeString,
Required: true,
ForceNew: true,
ValidateFunc: validation.StringLenBetween(1, 255),
},
"sns_topic_arn": {
Type: schema.TypeString,
Required: true,
Type: schema.TypeString,
Required: true,
ValidateFunc: validateArn,
},
"source_ids": {
Type: schema.TypeSet,
Expand All @@ -61,11 +69,12 @@ func resourceAwsDmsEventSubscription() *schema.Resource {
Optional: true,
// The API suppors modification but doing so loses all source_ids
ForceNew: true,
ValidateFunc: validation.StringInSlice([]string{
"replication-instance",
"replication-task",
}, false),
},
"tags": {
Type: schema.TypeMap,
Optional: true,
},
"tags": tagsSchema(),
},
}
}
Expand All @@ -92,13 +101,11 @@ func resourceAwsDmsEventSubscriptionCreate(d *schema.ResourceData, meta interfac
_, err := conn.CreateEventSubscription(request)

if err != nil {
return fmt.Errorf("Error creating DMS event subscription: %s", err)
return fmt.Errorf("error creating DMS Event Subscription (%s): %w", d.Get("name").(string), err)
}

d.SetId(d.Get("name").(string))

log.Println("[DEBUG] DMS create event subscription", request)

stateConf := &resource.StateChangeConf{
Pending: []string{"creating", "modifying"},
Target: []string{"active"},
Expand All @@ -108,50 +115,56 @@ func resourceAwsDmsEventSubscriptionCreate(d *schema.ResourceData, meta interfac
Delay: 10 * time.Second,
}

// Wait, catching any errors
_, err = stateConf.WaitForState()
if err != nil {
return fmt.Errorf("error waiting for DMS event subscription (%s) creation: %s", d.Id(), err)
return fmt.Errorf("error waiting for DMS Event Subscription (%s) creation: %w", d.Id(), err)
}

return resourceAwsDmsEventSubscriptionUpdate(d, meta)
return resourceAwsDmsEventSubscriptionRead(d, meta)
}

func resourceAwsDmsEventSubscriptionUpdate(d *schema.ResourceData, meta interface{}) error {
conn := meta.(*AWSClient).dmsconn

request := &dms.ModifyEventSubscriptionInput{
Enabled: aws.Bool(d.Get("enabled").(bool)),
SnsTopicArn: aws.String(d.Get("sns_topic_arn").(string)),
SubscriptionName: aws.String(d.Get("name").(string)),
SourceType: aws.String(d.Get("source_type").(string)),
}
if d.HasChanges("enabled", "event_categories", "sns_topic_arn", "source_type") {
request := &dms.ModifyEventSubscriptionInput{
Enabled: aws.Bool(d.Get("enabled").(bool)),
SnsTopicArn: aws.String(d.Get("sns_topic_arn").(string)),
SubscriptionName: aws.String(d.Get("name").(string)),
SourceType: aws.String(d.Get("source_type").(string)),
}

if v, ok := d.GetOk("event_categories"); ok {
request.EventCategories = expandStringList(v.(*schema.Set).List())
}
if v, ok := d.GetOk("event_categories"); ok {
request.EventCategories = expandStringList(v.(*schema.Set).List())
}

log.Println("[DEBUG] DMS update event subscription:", request)
_, err := conn.ModifyEventSubscription(request)

_, err := conn.ModifyEventSubscription(request)
if err != nil {
return fmt.Errorf("error updating DMS Event Subscription (%s): %w", d.Id(), err)
}

if err != nil {
return fmt.Errorf("Error updating DMS event subscription: %s", err)
}
stateConf := &resource.StateChangeConf{
Pending: []string{"modifying"},
Target: []string{"active"},
Refresh: resourceAwsDmsEventSubscriptionStateRefreshFunc(conn, d.Id()),
Timeout: d.Timeout(schema.TimeoutUpdate),
MinTimeout: 10 * time.Second,
Delay: 10 * time.Second,
}

stateConf := &resource.StateChangeConf{
Pending: []string{"modifying"},
Target: []string{"active"},
Refresh: resourceAwsDmsEventSubscriptionStateRefreshFunc(conn, d.Id()),
Timeout: d.Timeout(schema.TimeoutCreate),
MinTimeout: 10 * time.Second,
Delay: 10 * time.Second, // Wait 30 secs before starting
_, err = stateConf.WaitForState()
if err != nil {
return fmt.Errorf("error waiting for DMS Event Subscription (%s) modification: %w", d.Id(), err)
}
}

// Wait, catching any errors
_, err = stateConf.WaitForState()
if err != nil {
return fmt.Errorf("error waiting for DMS event subscription (%s) modification: %s", d.Id(), err)
if d.HasChange("tags") {
o, n := d.GetChange("tags")

if err := keyvaluetags.DatabasemigrationserviceUpdateTags(conn, d.Get("arn").(string), o, n); err != nil {
return fmt.Errorf("error updating DMS Event Subscription (%s) tags: %s", d.Get("arn").(string), err)
}
}

return resourceAwsDmsEventSubscriptionRead(d, meta)
Expand All @@ -164,8 +177,6 @@ func resourceAwsDmsEventSubscriptionRead(d *schema.ResourceData, meta interface{
SubscriptionName: aws.String(d.Id()),
}

log.Println("[DEBUG] DMS read event subscription:", request)

response, err := conn.DescribeEventSubscriptions(request)

if isAWSErr(err, dms.ErrCodeResourceNotFoundFault, "") {
Expand All @@ -178,52 +189,72 @@ func resourceAwsDmsEventSubscriptionRead(d *schema.ResourceData, meta interface{
return fmt.Errorf("Error reading DMS event subscription: %s", err)
}

if response == nil || response.EventSubscriptionsList == nil || len(response.EventSubscriptionsList) == 0 {
if response == nil || len(response.EventSubscriptionsList) == 0 || response.EventSubscriptionsList[0] == nil {
log.Printf("[WARN] DMS event subscription (%s) not found, removing from state", d.Id())
d.SetId("")
return nil
}

subscription := response.EventSubscriptionsList[0]

arn := arn.ARN{
Partition: meta.(*AWSClient).partition,
Service: "dms",
Region: meta.(*AWSClient).region,
AccountID: meta.(*AWSClient).accountid,
Resource: fmt.Sprintf("es:%s", d.Id()),
}.String()
d.Set("arn", arn)

d.Set("enabled", subscription.Enabled)
d.Set("sns_topic_arn", subscription.SnsTopicArn)
d.Set("source_type", subscription.SourceType)
d.Set("name", d.Id())
d.Set("event_categories", flattenStringList(subscription.EventCategoriesList))
d.Set("source_ids", flattenStringList(subscription.SourceIdsList))

tags, err := keyvaluetags.DatabasemigrationserviceListTags(conn, arn)

if err != nil {
return fmt.Errorf("error listing tags for DMS Event Subscription (%s): %s", arn, err)
}

if err := d.Set("tags", tags.IgnoreAws().Map()); err != nil {
return fmt.Errorf("error setting tags: %s", err)
}

return nil
}

func resourceAwsDmsEventSubscriptionDelete(d *schema.ResourceData, meta interface{}) error {
conn := meta.(*AWSClient).dmsconn

request := &dms.DeleteEventSubscriptionInput{
SubscriptionName: aws.String(d.Get("name").(string)),
SubscriptionName: aws.String(d.Id()),
}

log.Println("[DEBUG] DMS event subscription delete:", request)

_, err := conn.DeleteEventSubscription(request)

if isAWSErr(err, dms.ErrCodeResourceNotFoundFault, "") {
return nil
}

if err != nil {
return fmt.Errorf("Error deleting DMS event subscription: %s", err)
return fmt.Errorf("error deleting DMS Event Subscription (%s): %w", d.Id(), err)
}

stateConf := &resource.StateChangeConf{
Pending: []string{"deleting"},
Target: []string{},
Refresh: resourceAwsDmsEventSubscriptionStateRefreshFunc(conn, d.Id()),
Timeout: d.Timeout(schema.TimeoutCreate),
Timeout: d.Timeout(schema.TimeoutDelete),
MinTimeout: 10 * time.Second,
Delay: 10 * time.Second,
}

// Wait, catching any errors
_, err = stateConf.WaitForState()
if err != nil {
return fmt.Errorf("error waiting for DMS event subscription (%s) deletion: %s", d.Id(), err)
return fmt.Errorf("error waiting for DMS Event Subscription (%s) deletion: %w", d.Id(), err)
}

return nil
Expand Down
Loading

0 comments on commit 92200ba

Please sign in to comment.