add event_stream block to google_storage_transfer_job schema (#8894) (#16004)

* add event_stream block to google_storage_transfer_job schema

* refactor optional/required fields

* add flatteners/expanders

* add flatteners/expanders and tests

* update docs

* remove ForceNew

* add event_stream in update function

* add ConflictsWith for schedule/event_stream

* use ExactlyOneOf and update docs and tests

* Finish event_stream tests

* Update mmv1/third_party/terraform/services/storagetransfer/resource_storage_transfer_job_test.go



* add randomstrings to config

* remove copyright headers

---------

Signed-off-by: Modular Magician <[email protected]>
Co-authored-by: Sarah French <[email protected]>

Author: modular-magician

.changelog/8894.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+storagetransfer: added `event_stream` field to `google_storage_transfer_job` resource
+```

google/services/storagetransfer/resource_storage_transfer_job.go

@@ -82,6 +82,33 @@ func ResourceStorageTransferJob() *schema.Resource {
 				ForceNew:    true,
 				Description: `The project in which the resource belongs. If it is not provided, the provider project is used.`,
 			},
+			"event_stream": {
+				Type:          schema.TypeList,
+				Optional:      true,
+				MaxItems:      1,
+				ConflictsWith: []string{"schedule"},
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"name": {
+							Type:        schema.TypeString,
+							Required:    true,
+							Description: "Specifies a unique name of the resource such as AWS SQS ARN in the form 'arn:aws:sqs:region:account_id:queue_name', or Pub/Sub subscription resource name in the form 'projects/{project}/subscriptions/{sub}'",
+						},
+						"event_stream_start_time": {
+							Type:         schema.TypeString,
+							Optional:     true,
+							Description:  "Specifies the date and time that Storage Transfer Service starts listening for events from this stream. If no start time is specified or start time is in the past, Storage Transfer Service starts listening immediately",
+							ValidateFunc: validation.IsRFC3339Time,
+						},
+						"event_stream_expiration_time": {
+							Type:         schema.TypeString,
+							Optional:     true,
+							Description:  "Specifies the data and time at which Storage Transfer Service stops listening for events from this stream. After this time, any transfers in progress will complete, but no new transfers are initiated",
+							ValidateFunc: validation.IsRFC3339Time,
+						},
+					},
+				},
+			},
 			"transfer_spec": {
 				Type:     schema.TypeList,
 				Required: true,
@@ -195,9 +222,10 @@ func ResourceStorageTransferJob() *schema.Resource {
 				Description: `Notification configuration.`,
 			},
 			"schedule": {
-				Type:     schema.TypeList,
-				Optional: true,
-				MaxItems: 1,
+				Type:          schema.TypeList,
+				Optional:      true,
+				MaxItems:      1,
+				ConflictsWith: []string{"event_stream"},
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"schedule_start_date": {
@@ -567,6 +595,7 @@ func resourceStorageTransferJobCreate(d *schema.ResourceData, meta interface{})
 		ProjectId:          project,
 		Status:             d.Get("status").(string),
 		Schedule:           expandTransferSchedules(d.Get("schedule").([]interface{})),
+		EventStream:        expandEventStream(d.Get("event_stream").([]interface{})),
 		TransferSpec:       expandTransferSpecs(d.Get("transfer_spec").([]interface{})),
 		NotificationConfig: expandTransferJobNotificationConfig(d.Get("notification_config").([]interface{})),
 	}
@@ -642,6 +671,11 @@ func resourceStorageTransferJobRead(d *schema.ResourceData, meta interface{}) er
 		return err
 	}
 
+	err = d.Set("event_stream", flattenTransferEventStream(res.EventStream))
+	if err != nil {
+		return err
+	}
+
 	err = d.Set("transfer_spec", flattenTransferSpec(res.TransferSpec, d))
 	if err != nil {
 		return err
@@ -670,6 +704,13 @@ func resourceStorageTransferJobUpdate(d *schema.ResourceData, meta interface{})
 	transferJob := &storagetransfer.TransferJob{}
 	fieldMask := []string{}
 
+	if d.HasChange("event_stream") {
+		fieldMask = append(fieldMask, "event_stream")
+		if v, ok := d.GetOk("event_stream"); ok {
+			transferJob.EventStream = expandEventStream(v.([]interface{}))
+		}
+	}
+
 	if d.HasChange("description") {
 		fieldMask = append(fieldMask, "description")
 		if v, ok := d.GetOk("description"); ok {
@@ -899,6 +940,39 @@ func flattenTransferSchedule(transferSchedule *storagetransfer.Schedule) []map[s
 	return []map[string]interface{}{data}
 }
 
+func expandEventStream(e []interface{}) *storagetransfer.EventStream {
+	if len(e) == 0 || e[0] == nil {
+		return nil
+	}
+
+	eventStream := e[0].(map[string]interface{})
+	return &storagetransfer.EventStream{
+		Name:                      eventStream["name"].(string),
+		EventStreamStartTime:      eventStream["event_stream_start_time"].(string),
+		EventStreamExpirationTime: eventStream["event_stream_expiration_time"].(string),
+	}
+}
+
+func flattenTransferEventStream(eventStream *storagetransfer.EventStream) []map[string]interface{} {
+	if eventStream == nil || reflect.DeepEqual(eventStream, &storagetransfer.EventStream{}) {
+		return nil
+	}
+
+	data := map[string]interface{}{
+		"name": eventStream.Name,
+	}
+
+	if eventStream.EventStreamStartTime != "" {
+		data["event_stream_start_time"] = eventStream.EventStreamStartTime
+	}
+
+	if eventStream.EventStreamExpirationTime != "" {
+		data["event_stream_expiration_time"] = eventStream.EventStreamExpirationTime
+	}
+
+	return []map[string]interface{}{data}
+}
+
 func expandGcsData(gcsDatas []interface{}) *storagetransfer.GcsData {
 	if len(gcsDatas) == 0 || gcsDatas[0] == nil {
 		return nil

google/services/storagetransfer/resource_storage_transfer_job_test.go

@@ -201,6 +201,59 @@ func TestAccStorageTransferJob_transferOptions(t *testing.T) {
 	})
 }
 
+func TestAccStorageTransferJob_eventStream(t *testing.T) {
+	t.Parallel()
+
+	testDataSourceBucketName := acctest.RandString(t, 10)
+	testDataSinkName := acctest.RandString(t, 10)
+	testTransferJobDescription := acctest.RandString(t, 10)
+	testPubSubTopicName := fmt.Sprintf("tf-test-topic-%s", acctest.RandString(t, 10))
+	testEventStreamPubSubTopicName := fmt.Sprintf("tf-test-topic-%s", acctest.RandString(t, 10))
+	testPubSubSubscriptionName := fmt.Sprintf("tf-test-subscription-%s", acctest.RandString(t, 10))
+	eventStreamStart := []string{"2014-10-02T15:01:23Z", "2019-10-02T15:01:23Z"}
+	eventStreamEnd := []string{"2022-10-02T15:01:23Z", "2032-10-02T15:01:23Z"}
+
+	acctest.VcrTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.AccTestPreCheck(t) },
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories(t),
+		CheckDestroy:             testAccStorageTransferJobDestroyProducer(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccStorageTransferJob_basic(envvar.GetTestProjectFromEnv(), testDataSourceBucketName, testDataSinkName, testTransferJobDescription, testPubSubTopicName),
+			},
+			{
+				ResourceName:      "google_storage_transfer_job.transfer_job",
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccStorageTransferJob_eventStream(envvar.GetTestProjectFromEnv(), testDataSourceBucketName, testDataSinkName, testEventStreamPubSubTopicName, testPubSubSubscriptionName, testTransferJobDescription, eventStreamStart[0], eventStreamEnd[0]),
+			},
+			{
+				ResourceName:      "google_storage_transfer_job.transfer_job",
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccStorageTransferJob_eventStream(envvar.GetTestProjectFromEnv(), testDataSourceBucketName, testDataSinkName, testEventStreamPubSubTopicName, testPubSubSubscriptionName, testTransferJobDescription, eventStreamStart[1], eventStreamEnd[0]),
+			},
+			{
+				ResourceName:      "google_storage_transfer_job.transfer_job",
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccStorageTransferJob_eventStream(envvar.GetTestProjectFromEnv(), testDataSourceBucketName, testDataSinkName, testEventStreamPubSubTopicName, testPubSubSubscriptionName, testTransferJobDescription, eventStreamStart[1], eventStreamEnd[1]),
+			},
+			{
+				ResourceName:      "google_storage_transfer_job.transfer_job",
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccStorageTransferJob_objectConditions(t *testing.T) {
 	t.Parallel()
 
@@ -383,6 +436,99 @@ resource "google_storage_transfer_job" "transfer_job" {
 `, project, dataSourceBucketName, project, dataSinkBucketName, project, transferJobDescription, project)
 }
 
+func testAccStorageTransferJob_eventStream(project string, dataSourceBucketName string, dataSinkBucketName string, pubsubTopicName string, pubsubSubscriptionName string, transferJobDescription string, eventStreamStart string, eventStreamEnd string) string {
+	return fmt.Sprintf(`
+data "google_storage_transfer_project_service_account" "default" {
+  project = "%s"
+}
+
+resource "google_storage_bucket" "data_source" {
+  name          = "%s"
+  project       = "%s"
+  location      = "US"
+  force_destroy = true
+  uniform_bucket_level_access = true
+}
+
+resource "google_storage_bucket_iam_member" "data_source" {
+  bucket = google_storage_bucket.data_source.name
+  role   = "roles/storage.admin"
+  member = "serviceAccount:${data.google_storage_transfer_project_service_account.default.email}"
+}
+
+resource "google_storage_bucket" "data_sink" {
+  name          = "%s"
+  project       = "%s"
+  location      = "US"
+  force_destroy = true
+  uniform_bucket_level_access = true
+}
+
+resource "google_storage_bucket_iam_member" "data_sink" {
+  bucket = google_storage_bucket.data_sink.name
+  role   = "roles/storage.admin"
+  member = "serviceAccount:${data.google_storage_transfer_project_service_account.default.email}"
+}
+
+resource "google_pubsub_subscription_iam_member" "editor" {
+  subscription = google_pubsub_subscription.example.name
+  role         = "roles/editor"
+  member       = "serviceAccount:${data.google_storage_transfer_project_service_account.default.email}"
+}
+
+resource "google_pubsub_topic" "example" {
+  name = "%s"
+}
+
+resource "google_pubsub_subscription" "example" {
+  name  = "%s"
+  topic = google_pubsub_topic.example.name
+
+  ack_deadline_seconds = 20
+
+  labels = {
+    foo = "bar"
+  }
+
+  push_config {
+    push_endpoint = "https://example.com/push"
+
+    attributes = {
+      x-goog-version = "v1"
+    }
+  }
+}
+
+resource "google_storage_transfer_job" "transfer_job" {
+  description = "%s"
+  project     = "%s"
+
+  event_stream {
+    name = google_pubsub_subscription.example.id
+    event_stream_start_time = "%s"
+    event_stream_expiration_time = "%s"
+  }
+
+  transfer_spec {
+    gcs_data_source {
+      bucket_name = google_storage_bucket.data_source.name
+      path  = "foo/bar/"
+    }
+    gcs_data_sink {
+      bucket_name = google_storage_bucket.data_sink.name
+      path  = "foo/bar/"
+    }
+  }
+
+  depends_on = [
+    google_storage_bucket_iam_member.data_source,
+    google_storage_bucket_iam_member.data_sink,
+    google_pubsub_subscription_iam_member.editor,
+  ]
+}
+`, project, dataSourceBucketName, project, dataSinkBucketName, project, pubsubTopicName, pubsubSubscriptionName, transferJobDescription, project, eventStreamStart, eventStreamEnd)
+}
+
 func testAccStorageTransferJob_omitNotificationConfig(project string, dataSourceBucketName string, dataSinkBucketName string, transferJobDescription string) string {
 	return fmt.Sprintf(`
 data "google_storage_transfer_project_service_account" "default" {

website/docs/r/storage_transfer_job.html.markdown

@@ -116,10 +116,12 @@ The following arguments are supported:
 
 * `transfer_spec` - (Required) Transfer specification. Structure [documented below](#nested_transfer_spec).
 
-* `schedule` - (Required) Schedule specification defining when the Transfer Job should be scheduled to start, end and what time to run. Structure [documented below](#nested_schedule).
-
 - - -
 
+* `schedule` - (Optional) Schedule specification defining when the Transfer Job should be scheduled to start, end and what time to run. Structure [documented below](#nested_schedule). Either `schedule` or `event_stream` must be set.
+
+* `event_stream` - (Optional) Specifies the Event-driven transfer options. Event-driven transfers listen to an event stream to transfer updated files. Structure [documented below](#nested_event_stream) Either `event_stream` or `schedule` must be set.
+
 * `project` - (Optional) The project in which the resource belongs. If it
 	is not provided, the provider project is used.
 
@@ -161,6 +163,14 @@ The following arguments are supported:
 
 * `repeat_interval` - (Optional) Interval between the start of each scheduled transfer. If unspecified, the default value is 24 hours. This value may not be less than 1 hour. A duration in seconds with up to nine fractional digits, terminated by 's'. Example: "3.5s".
 
+<a name="nested_event_stream"></a>The `event_stream` block supports:
+
+* `name` - (Required) Specifies a unique name of the resource such as AWS SQS ARN in the form 'arn:aws:sqs:region:account_id:queue_name', or Pub/Sub subscription resource name in the form 'projects/{project}/subscriptions/{sub}'.
+
+* `event_stream_start_time` - (Optional) Specifies the date and time that Storage Transfer Service starts listening for events from this stream. If no start time is specified or start time is in the past, Storage Transfer Service starts listening immediately. A timestamp in RFC3339 UTC "Zulu" format, with nanosecond resolution and up to nine fractional digits. Examples: "2014-10-02T15:01:23Z" and "2014-10-02T15:01:23.045123456Z".
+
+* `event_stream_expiration_time` - (Optional) Specifies the data and time at which Storage Transfer Service stops listening for events from this stream. After this time, any transfers in progress will complete, but no new transfers are initiated.A timestamp in RFC3339 UTC "Zulu" format, with nanosecond resolution and up to nine fractional digits. Examples: "2014-10-02T15:01:23Z" and "2014-10-02T15:01:23.045123456Z".
+
 <a name="nested_object_conditions"></a>The `object_conditions` block supports:
 
 * `max_time_elapsed_since_last_modification` - (Optional) A duration in seconds with up to nine fractional digits, terminated by 's'. Example: "3.5s".