Merge branch 'dms_event_subscription' of ssh://github.com/hawknewton/…

…terraform-provider-aws into hawknewton-dms_event_subscription

aws/provider.go

@@ -475,6 +475,7 @@ func Provider() terraform.ResourceProvider {
  "aws_dlm_lifecycle_policy": resourceAwsDlmLifecyclePolicy(),
  "aws_dms_certificate": resourceAwsDmsCertificate(),
  "aws_dms_endpoint": resourceAwsDmsEndpoint(),
+ "aws_dms_event_subscription": resourceAwsDmsEventSubscription(),
  "aws_dms_replication_instance": resourceAwsDmsReplicationInstance(),
  "aws_dms_replication_subnet_group": resourceAwsDmsReplicationSubnetGroup(),
  "aws_dms_replication_task": resourceAwsDmsReplicationTask(),

aws/resource_aws_dms_event_subscription.go

@@ -0,0 +1,252 @@
+package aws
+
+import (
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/aws/aws-sdk-go/aws"
+ 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/terraform-providers/terraform-provider-aws/aws/internal/keyvaluetags"
+)
+
+func resourceAwsDmsEventSubscription() *schema.Resource {
+ return &schema.Resource{
+ Create: resourceAwsDmsEventSubscriptionCreate,
+ Read: resourceAwsDmsEventSubscriptionRead,
+ Update: resourceAwsDmsEventSubscriptionUpdate,
+ Delete: resourceAwsDmsEventSubscriptionDelete,
+ Timeouts: &schema.ResourceTimeout{
+ Create: schema.DefaultTimeout(10 * time.Minute),
+ Delete: schema.DefaultTimeout(10 * time.Minute),
+ Update: schema.DefaultTimeout(10 * time.Minute),
+ },
+
+ Importer: &schema.ResourceImporter{
+ State: schema.ImportStatePassthrough,
+ },
+
+ Schema: map[string]*schema.Schema{
+ "enabled": {
+ Type: schema.TypeBool,
+ Optional: true,
+ Default: true,
+ },
+ "event_categories": {
+ Type: schema.TypeSet,
+ Elem: &schema.Schema{Type: schema.TypeString},
+ Set: schema.HashString,
+ Required: true,
+ },
+ "name": {
+ Type: schema.TypeString,
+ Required: true,
+ ForceNew: true,
+ },
+ "sns_topic_arn": {
+ Type: schema.TypeString,
+ Required: true,
+ },
+ "source_ids": {
+ Type: schema.TypeSet,
+ Elem: &schema.Schema{Type: schema.TypeString},
+ Set: schema.HashString,
+ ForceNew: true,
+ Optional: true,
+ },
+ "source_type": {
+ Type: schema.TypeString,
+ Optional: true,
+ // The API suppors modification but doing so loses all source_ids
+ ForceNew: true,
+ },
+ "tags": {
+ Type: schema.TypeMap,
+ Optional: true,
+ },
+ },
+ }
+}
+
+func resourceAwsDmsEventSubscriptionCreate(d *schema.ResourceData, meta interface{}) error {
+ conn := meta.(*AWSClient).dmsconn
+
+ request := &dms.CreateEventSubscriptionInput{
+ 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)),
+ Tags: keyvaluetags.New(d.Get("tags").(map[string]interface{})).IgnoreAws().DatabasemigrationserviceTags(),
+ }
+
+ if v, ok := d.GetOk("event_categories"); ok {
+ request.EventCategories = expandStringList(v.(*schema.Set).List())
+ }
+
+ if v, ok := d.GetOk("source_ids"); ok {
+ request.SourceIds = expandStringList(v.(*schema.Set).List())
+ }
+
+ _, err := conn.CreateEventSubscription(request)
+
+ if err != nil {
+ return fmt.Errorf("Error creating DMS event subscription: %s", 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"},
+ Refresh: resourceAwsDmsEventSubscriptionStateRefreshFunc(conn, d.Id()),
+ Timeout: d.Timeout(schema.TimeoutCreate),
+ 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) creation: %s", d.Id(), err)
+ }
+
+ return resourceAwsDmsEventSubscriptionUpdate(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 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)
+
+ 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.TimeoutCreate),
+ MinTimeout: 10 * time.Second,
+ Delay: 10 * time.Second, // Wait 30 secs before starting
+ }
+
+ // 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)
+ }
+
+ return resourceAwsDmsEventSubscriptionRead(d, meta)
+}
+
+func resourceAwsDmsEventSubscriptionRead(d *schema.ResourceData, meta interface{}) error {
+ conn := meta.(*AWSClient).dmsconn
+
+ request := &dms.DescribeEventSubscriptionsInput{
+ SubscriptionName: aws.String(d.Id()),
+ }
+
+ log.Println("[DEBUG] DMS read event subscription:", request)
+
+ response, err := conn.DescribeEventSubscriptions(request)
+
+ if isAWSErr(err, dms.ErrCodeResourceNotFoundFault, "") {
+ log.Printf("[WARN] DMS event subscription (%s) not found, removing from state", d.Id())
+ d.SetId("")
+ return nil
+ }
+
+ if err != nil {
+ return fmt.Errorf("Error reading DMS event subscription: %s", err)
+ }
+
+ if response == nil || response.EventSubscriptionsList == nil || len(response.EventSubscriptionsList) == 0 {
+ log.Printf("[WARN] DMS event subscription (%s) not found, removing from state", d.Id())
+ d.SetId("")
+ return nil
+ }
+
+ subscription := response.EventSubscriptionsList[0]
+
+ 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))
+
+ return nil
+}
+
+func resourceAwsDmsEventSubscriptionDelete(d *schema.ResourceData, meta interface{}) error {
+ conn := meta.(*AWSClient).dmsconn
+
+ request := &dms.DeleteEventSubscriptionInput{
+ SubscriptionName: aws.String(d.Get("name").(string)),
+ }
+
+ log.Println("[DEBUG] DMS event subscription delete:", request)
+
+ _, err := conn.DeleteEventSubscription(request)
+
+ if err != nil {
+ return fmt.Errorf("Error deleting DMS event subscription: %s", err)
+ }
+
+ stateConf := &resource.StateChangeConf{
+ Pending: []string{"deleting"},
+ Target: []string{},
+ Refresh: resourceAwsDmsEventSubscriptionStateRefreshFunc(conn, d.Id()),
+ Timeout: d.Timeout(schema.TimeoutCreate),
+ 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 nil
+}
+
+func resourceAwsDmsEventSubscriptionStateRefreshFunc(conn *dms.DatabaseMigrationService, name string) resource.StateRefreshFunc {
+ return func() (interface{}, string, error) {
+ v, err := conn.DescribeEventSubscriptions(&dms.DescribeEventSubscriptionsInput{
+ SubscriptionName: aws.String(name),
+ })
+
+ if isAWSErr(err, dms.ErrCodeResourceNotFoundFault, "") {
+ return nil, "", nil
+ }
+
+ if err != nil {
+ return nil, "", err
+ }
+
+ if v == nil || len(v.EventSubscriptionsList) == 0 || v.EventSubscriptionsList[0] == nil {
+ return nil, "", nil
+ }
+
+ return v, aws.StringValue(v.EventSubscriptionsList[0].Status), nil
+ }
+}