AWS Glue worker_type feature

aws/resource_aws_glue_job.go

@@ -3,6 +3,7 @@ package aws
 import (
 	"fmt"
 	"log"
+	"regexp"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/glue"
@@ -25,7 +26,7 @@ func resourceAwsGlueJob() *schema.Resource {
 				Type:          schema.TypeInt,
 				Optional:      true,
 				Computed:      true,
-				ConflictsWith: []string{"max_capacity"},
+				ConflictsWith: []string{"max_capacity", "number_of_workers", "worker_type"},
 				Deprecated:    "Please use attribute `max_capacity' instead. This attribute might be removed in future releases.",
 				ValidateFunc:  validation.IntAtLeast(2),
 			},
@@ -86,7 +87,7 @@ func resourceAwsGlueJob() *schema.Resource {
 				Type:          schema.TypeFloat,
 				Optional:      true,
 				Computed:      true,
-				ConflictsWith: []string{"allocated_capacity"},
+				ConflictsWith: []string{"allocated_capacity", "number_of_workers", "worker_type"},
 			},
 			"max_retries": {
 				Type:         schema.TypeInt,
@@ -113,6 +114,18 @@ func resourceAwsGlueJob() *schema.Resource {
 				Type:     schema.TypeString,
 				Optional: true,
 			},
+			"worker_type": {
+				Type:          schema.TypeString,
+				Optional:      true,
+				ConflictsWith: []string{"allocated_capacity", "max_capacity", "allocated_capacity"},
+				ValidateFunc:  validation.StringMatch(regexp.MustCompile(`^Standard|G.1X|G.2X`), "Must be one of Standard, G.1X or G.2X. See https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-jobs-job.html"),
+			},
+			"number_of_workers": {
+				Type:          schema.TypeInt,
+				Optional:      true,
+				ConflictsWith: []string{"allocated_capacity", "max_capacity", "allocated_capacity"},
+				ValidateFunc:  validation.IntAtLeast(2),
+			},
 		},
 	}
 }
@@ -167,6 +180,14 @@ func resourceAwsGlueJobCreate(d *schema.ResourceData, meta interface{}) error {
 		input.SecurityConfiguration = aws.String(v.(string))
 	}
 
+	if v, ok := d.GetOk("worker_type"); ok {
+		input.WorkerType = aws.String(v.(string))
+	}
+
+	if v, ok := d.GetOk("number_of_workers"); ok {
+		input.NumberOfWorkers = aws.Int64(int64(v.(int)))
+	}
+
 	log.Printf("[DEBUG] Creating Glue Job: %s", input)
 	_, err := conn.CreateJob(input)
 	if err != nil {
@@ -225,6 +246,9 @@ func resourceAwsGlueJobRead(d *schema.ResourceData, meta interface{}) error {
 		return fmt.Errorf("error setting security_configuration: %s", err)
 	}
 
+	d.Set("worker_type", job.WorkerType)
+	d.Set("number_of_workers", int(aws.Int64Value(job.NumberOfWorkers)))
+
 	// TODO: Deprecated fields - remove in next major version
 	d.Set("allocated_capacity", int(aws.Int64Value(job.AllocatedCapacity)))
 
@@ -240,13 +264,16 @@ func resourceAwsGlueJobUpdate(d *schema.ResourceData, meta interface{}) error {
 		Timeout: aws.Int64(int64(d.Get("timeout").(int))),
 	}
 
-	if v, ok := d.GetOk("max_capacity"); ok {
-		jobUpdate.MaxCapacity = aws.Float64(v.(float64))
-	}
-
-	if d.HasChange("allocated_capacity") {
-		jobUpdate.MaxCapacity = aws.Float64(float64(d.Get("allocated_capacity").(int)))
-		log.Printf("[WARN] Using deprecated `allocated_capacity' attribute.")
+	if v, ok := d.GetOk("number_of_workers"); ok {
+		jobUpdate.NumberOfWorkers = aws.Int64(int64(v.(int)))
+	} else {
+		if v, ok := d.GetOk("max_capacity"); ok {
+			jobUpdate.MaxCapacity = aws.Float64(v.(float64))
+		}
+		if d.HasChange("allocated_capacity") {
+			jobUpdate.MaxCapacity = aws.Float64(float64(d.Get("allocated_capacity").(int)))
+			log.Printf("[WARN] Using deprecated `allocated_capacity' attribute.")
+		}
 	}
 
 	if v, ok := d.GetOk("connections"); ok {
@@ -279,6 +306,10 @@ func resourceAwsGlueJobUpdate(d *schema.ResourceData, meta interface{}) error {
 		jobUpdate.SecurityConfiguration = aws.String(v.(string))
 	}
 
+	if v, ok := d.GetOk("worker_type"); ok {
+		jobUpdate.WorkerType = aws.String(v.(string))
+	}
+
 	input := &glue.UpdateJobInput{
 		JobName:   aws.String(d.Id()),
 		JobUpdate: jobUpdate,

aws/resource_aws_glue_job_test.go

@@ -379,6 +379,47 @@ func TestAccAWSGlueJob_SecurityConfiguration(t *testing.T) {
 	})
 }
 
+func TestAccAWSGlueJob_WorkerType(t *testing.T) {
+	var job glue.Job
+
+	rName := fmt.Sprintf("tf-acc-test-%s", acctest.RandString(5))
+	resourceName := "aws_glue_job.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSGlueJobDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccAWSGlueJobConfig_WorkerType(rName, "Standard"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSGlueJobExists(resourceName, &job),
+					resource.TestCheckResourceAttr(resourceName, "worker_type", "Standard"),
+				),
+			},
+			{
+				Config: testAccAWSGlueJobConfig_WorkerType(rName, "G.1X"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSGlueJobExists(resourceName, &job),
+					resource.TestCheckResourceAttr(resourceName, "worker_type", "G.1X"),
+				),
+			},
+			{
+				Config: testAccAWSGlueJobConfig_WorkerType(rName, "G.2X"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSGlueJobExists(resourceName, &job),
+					resource.TestCheckResourceAttr(resourceName, "worker_type", "G.2X"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccAWSGlueJob_PythonShell(t *testing.T) {
 	var job glue.Job
 
@@ -742,6 +783,25 @@ resource "aws_glue_job" "test" {
 `, testAccAWSGlueJobConfig_Base(rName), rName, securityConfiguration)
 }
 
+func testAccAWSGlueJobConfig_WorkerType(rName string, workerType string) string {
+	return fmt.Sprintf(`
+%s
+
+resource "aws_glue_job" "test" {
+  name                   = "%s"
+  role_arn               = "${aws_iam_role.test.arn}"
+  worker_type 			 = "%s"
+  number_of_workers      = 10
+
+  command {
+    script_location = "testscriptlocation"
+  }
+
+  depends_on = ["aws_iam_role_policy_attachment.test"]
+}
+`, testAccAWSGlueJobConfig_Base(rName), rName, workerType)
+}
+
 func testAccAWSGlueJobConfig_PythonShell(rName string) string {
 	return fmt.Sprintf(`
 %s

website/docs/r/glue_job.html.markdown

@@ -61,6 +61,8 @@ be removed in future releases, please use `max_capacity` instead.
 * `role_arn` – (Required) The ARN of the IAM role associated with this job.
 * `timeout` – (Optional) The job timeout in minutes. The default is 2880 minutes (48 hours).
 * `security_configuration` - (Optional) The name of the Security Configuration to be associated with the job.
+* `worker_type` - (Optional) The type of predefined worker that is allocated when a job runs. Accepts a value of Standard, G.1X, or G.2X.
+* `number_of_workers` - (Optional) The number of workers of a defined workerType that are allocated when a job runs.
 
 ### command Argument Reference