Allow Elasticsearch to be used as an engine for DMS Endpoint

aws/resource_aws_dms_endpoint.go

@@ -69,6 +69,7 @@ func resourceAwsDmsEndpoint() *schema.Resource {
 					"db2",
 					"docdb",
 					"dynamodb",
+					"elasticsearch",
 					"mariadb",
 					"mongodb",
 					"mysql",
@@ -217,6 +218,41 @@ func resourceAwsDmsEndpoint() *schema.Resource {
 					},
 				},
 			},
+			"elasticsearch_settings": {
+				Type:     schema.TypeList,
+				Optional: true,
+				MaxItems: 1,
+				DiffSuppressFunc: func(k, old, new string, d *schema.ResourceData) bool {
+					if old == "1" && new == "0" {
+						return true
+					}
+					return false
+				},
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"service_access_role_arn": {
+							Type:     schema.TypeString,
+							Required: true,
+							ForceNew: true,
+						},
+						"endpoint_uri": {
+							Type:     schema.TypeString,
+							Required: true,
+							ForceNew: true,
+						},
+						"error_retry_duration": {
+							Type:     schema.TypeInt,
+							Optional: true,
+							Default:  "300",
+						},
+						"full_load_error_percentage": {
+							Type:     schema.TypeInt,
+							Optional: true,
+							Default:  "10",
+						},
+					},
+				},
+			},
 		},
 	}
 }
@@ -270,6 +306,13 @@ func resourceAwsDmsEndpointCreate(d *schema.ResourceData, meta interface{}) erro
 			BucketName:              aws.String(d.Get("s3_settings.0.bucket_name").(string)),
 			CompressionType:         aws.String(d.Get("s3_settings.0.compression_type").(string)),
 		}
+	case "elasticsearch":
+		request.ElasticsearchSettings = &dms.ElasticsearchSettings{
+			ServiceAccessRoleArn:    aws.String(d.Get("elasticsearch_settings.0.service_access_role_arn").(string)),
+			EndpointUri:             aws.String(d.Get("elasticsearch_settings.0.endpoint_uri").(string)),
+			ErrorRetryDuration:      aws.Int64(int64(d.Get("elasticsearch_settings.0.error_retry_duration").(int))),
+			FullLoadErrorPercentage: aws.Int64(int64(d.Get("elasticsearch_settings.0.full_load_error_percentage").(int))),
+		}
 	default:
 		request.Password = aws.String(d.Get("password").(string))
 		request.Port = aws.Int64(int64(d.Get("port").(int)))
@@ -477,6 +520,20 @@ func resourceAwsDmsEndpointUpdate(d *schema.ResourceData, meta interface{}) erro
 			request.EngineName = aws.String(d.Get("engine_name").(string)) // Must be included (should be 's3')
 			hasChanges = true
 		}
+	case "elasticsearch":
+		if d.HasChange("elasticsearch_settings.0.endpoint_uri") ||
+			d.HasChange("elasticsearch_settings.0.error_retry_duration") ||
+			d.HasChange("elasticsearch_settings.0.full_load_error_percentage") ||
+			d.HasChange("elasticsearch_settings.0.service_access_role_arn") {
+			request.ElasticsearchSettings = &dms.ElasticsearchSettings{
+				ServiceAccessRoleArn:    aws.String(d.Get("elasticsearch_settings.0.service_access_role_arn").(string)),
+				EndpointUri:             aws.String(d.Get("elasticsearch_settings.0.endpoint_uri").(string)),
+				ErrorRetryDuration:      aws.Int64(int64(d.Get("elasticsearch_settings.0.error_retry_duration").(int))),
+				FullLoadErrorPercentage: aws.Int64(int64(d.Get("elasticsearch_settings.0.full_load_error_percentage").(int))),
+			}
+			request.EngineName = aws.String(d.Get("engine_name").(string))
+			hasChanges = true
+		}
 	default:
 		if d.HasChange("database_name") {
 			request.DatabaseName = aws.String(d.Get("database_name").(string))
@@ -567,6 +624,10 @@ func resourceAwsDmsEndpointSetState(d *schema.ResourceData, endpoint *dms.Endpoi
 		if err := d.Set("s3_settings", flattenDmsS3Settings(endpoint.S3Settings)); err != nil {
 			return fmt.Errorf("Error setting s3_settings for DMS: %s", err)
 		}
+	case "elasticsearch":
+		if err := d.Set("elasticsearch_settings", flattenDmsElasticsearchSettings(endpoint.ElasticsearchSettings)); err != nil {
+			return fmt.Errorf("Error setting elasticsearch for DMS: %s", err)
+		}
 	default:
 		d.Set("database_name", endpoint.DatabaseName)
 		d.Set("extra_connection_attributes", endpoint.ExtraConnectionAttributes)
@@ -615,3 +676,18 @@ func flattenDmsS3Settings(settings *dms.S3Settings) []map[string]interface{} {
 
 	return []map[string]interface{}{m}
 }
+
+func flattenDmsElasticsearchSettings(settings *dms.ElasticsearchSettings) []map[string]interface{} {
+	if settings == nil {
+		return []map[string]interface{}{}
+	}
+
+	m := map[string]interface{}{
+		"service_access_role_arn":    aws.StringValue(settings.ServiceAccessRoleArn),
+		"endpoint_uri":               aws.StringValue(settings.EndpointUri),
+		"full_load_error_percentage": aws.Int64Value(settings.FullLoadErrorPercentage),
+		"error_retry_duration":       aws.Int64Value(settings.ErrorRetryDuration),
+	}
+
+	return []map[string]interface{}{m}
+}

aws/resource_aws_dms_endpoint_test.go

@@ -96,6 +96,37 @@ func TestAccAwsDmsEndpoint_S3(t *testing.T) {
 	})
 }
 
+func TestAccAwsDmsEndpoint_Elasticsearch(t *testing.T) {
+	resourceName := "aws_dms_endpoint.dms_endpoint"
+	randId := acctest.RandString(8) + "-elasticsearch"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: dmsEndpointDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: dmsEndpointElasticsearchConfig(randId),
+				Check: resource.ComposeTestCheckFunc(
+					checkDmsEndpointExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "elasticsearch_settings.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "elasticsearch_settings.0.endpoint_uri", "search-estest.us-west-2.es.amazonaws.com"),
+					resource.TestCheckResourceAttr(resourceName, "elasticsearch_settings.0.full_load_error_percentage", "10"),
+					resource.TestCheckResourceAttr(resourceName, "elasticsearch_settings.0.error_retry_duration", "300"),
+				),
+			},
+			{
+				ResourceName:            resourceName,
+				ImportState:             true,
+				ImportStateVerify:       true,
+				ImportStateVerifyIgnore: []string{"password"},
+			},
+			//  No update step due to:
+			//  InvalidParameterCombinationException: Elasticsearch endpoint cant be modified.
+		},
+	})
+}
+
 func TestAccAwsDmsEndpoint_DynamoDb(t *testing.T) {
 	resourceName := "aws_dms_endpoint.dms_endpoint"
 	randId := acctest.RandString(8) + "-dynamodb"
@@ -623,6 +654,75 @@ EOF
 `, randId)
 }
 
+func dmsEndpointElasticsearchConfig(randId string) string {
+	return fmt.Sprintf(`
+resource "aws_dms_endpoint" "dms_endpoint" {
+  endpoint_id                 = "tf-test-dms-endpoint-%[1]s"
+  endpoint_type               = "target"
+  engine_name                 = "elasticsearch"
+  ssl_mode                    = "none"
+
+  tags = {
+    Name   = "tf-test-elasticsearch-endpoint-%[1]s"
+    Update = "to-update"
+    Remove = "to-remove"
+  }
+
+  elasticsearch_settings {
+    service_access_role_arn    = "${aws_iam_role.iam_role.arn}"
+    endpoint_uri               = "search-estest.us-west-2.es.amazonaws.com"
+    full_load_error_percentage = 10
+    error_retry_duration       = 300
+  }
+
+  depends_on = ["aws_iam_role_policy.dms_elasticsearch_access"]
+}
+
+resource "aws_iam_role" "iam_role" {
+  name = "tf-test-iam-elasticsearch-role-%[1]s"
+
+  assume_role_policy = <<EOF
+{
+	"Version": "2012-10-17",
+	"Statement": [
+		{
+			"Action": "sts:AssumeRole",
+			"Principal": {
+				"Service": "dms.amazonaws.com"
+			},
+			"Effect": "Allow"
+		}
+	]
+}
+EOF
+}
+
+resource "aws_iam_role_policy" "dms_elasticsearch_access" {
+  name = "tf-test-iam-elasticsearch-role-policy-%[1]s"
+  role = "${aws_iam_role.iam_role.name}"
+
+  policy = <<EOF
+{
+	"Version": "2012-10-17",
+	"Statement": [
+		{
+			"Effect": "Allow",
+			"Action": [
+				 "es:ESHttpDelete",
+				 "es:ESHttpGet",
+				 "es:ESHttpHead",
+				 "es:ESHttpPost",
+				 "es:ESHttpPut"
+			],
+			"Resource": "*"
+		}
+	]
+}
+EOF
+}
+`, randId)
+}
+
 func dmsEndpointMongoDbConfig(randId string) string {
 	return fmt.Sprintf(`
 data "aws_kms_alias" "dms" {

website/docs/r/dms_endpoint.html.markdown

@@ -53,7 +53,7 @@ The following arguments are supported:
     - Must not contain two consecutive hyphens
 
 * `endpoint_type` - (Required) The type of endpoint. Can be one of `source | target`.
-* `engine_name` - (Required) The type of engine for the endpoint. Can be one of `aurora | azuredb | db2 | docdb | dynamodb | mariadb | mongodb | mysql | oracle | postgres | redshift | s3 | sqlserver | sybase`.
+* `engine_name` - (Required) The type of engine for the endpoint. Can be one of `aurora | azuredb | db2 | docdb | dynamodb | elasticsearch | mariadb | mongodb | mysql | oracle | postgres | redshift | s3 | sqlserver | sybase`.
 * `extra_connection_attributes` - (Optional) Additional attributes associated with the connection. For available attributes see [Using Extra Connection Attributes with AWS Database Migration Service](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Introduction.ConnectionAttributes.html).
 * `kms_key_arn` - (Required when `engine_name` is `mongodb`, optional otherwise) The Amazon Resource Name (ARN) for the KMS key that will be used to encrypt the connection parameters. If you do not specify a value for `kms_key_arn`, then AWS DMS will use your default encryption key. AWS KMS creates the default encryption key for your AWS account. Your AWS account has a different default encryption key for each AWS region.
 * `password` - (Optional) The password to be used to login to the endpoint database.
@@ -65,6 +65,7 @@ The following arguments are supported:
 * `service_access_role` - (Optional) The Amazon Resource Name (ARN) used by the service access IAM role for dynamodb endpoints.
 * `mongodb_settings` - (Optional) Settings for the source MongoDB endpoint. Available settings are `auth_type` (default: `password`), `auth_mechanism` (default: `default`), `nesting_level` (default: `none`), `extract_doc_id` (default: `false`), `docs_to_investigate` (default: `1000`) and `auth_source` (default: `admin`). For more details, see [Using MongoDB as a Source for AWS DMS](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.MongoDB.html).
 * `s3_settings` - (Optional) Settings for the target S3 endpoint. Available settings are `service_access_role_arn`, `external_table_definition`, `csv_row_delimiter` (default: `\\n`), `csv_delimiter` (default: `,`), `bucket_folder`, `bucket_name` and `compression_type` (default: `NONE`). For more details, see [Using Amazon S3 as a Target for AWS Database Migration Service](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.S3.html).
+* `elasticsearch_settings` - (Optional) Settings for the target Elasticsearch. Available settings are `service_access_role_arn`, `endpoint_uri`, `error_retry_duration` (default: `300`) and `full_load_error_percentage` (default: `10`). For more details, see [Using an Amazon Elasticsearch Service Cluster as a Target for AWS Database Migration Service](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Elasticsearch.html).
 
 ## Attributes Reference